Order Sidekick recovery fixture reads after queued saves

packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts

@@ -24,7 +24,7 @@ import { appendFile, mkdir, open, readFile, rm, writeFile } from "node:fs/promis
 import { randomBytes } from "node:crypto";
 import { join } from "node:path";
 import process from "node:process";
-import { ensure, main, Ok, spawn, suspend, until, withResolvers } from "effection";
+import { ensure, main, Ok, race, spawn, suspend, until, withResolvers } from "effection";
 import type { Operation, Result } from "effection";
 import { parseDurableEvent, serializeDurableEvent } from "@executablemd/durable-streams";
 import type { DurableEvent } from "@executablemd/durable-streams";
@@ -64,7 +64,11 @@ import { assembleReplProfile } from "../../../../src/repl-profile.ts";
 import { NO_PLUGINS } from "../../../../src/plugin-host.ts";
 import { rootReads } from "../../../../src/repl/sidekick/requests.ts";
 import { useSidekickOwner } from "../../../../src/repl/sidekick/owner.ts";
-import type { SidekickAssembly, SidekickOwner } from "../../../../src/repl/sidekick/owner.ts";
+import type {
+  SidekickAssembly,
+  SidekickOwner,
+  SidekickTurn,
+} from "../../../../src/repl/sidekick/owner.ts";
 import type { SubmitReceipt } from "../../../../src/repl/sidekick/controls.ts";
 import type { PairedChatStorage } from "../../../../src/repl/sidekick/repository.ts";
 import { createFakeAcp, makeRegistry } from "../../../support/fake-acp.ts";
@@ -327,8 +331,10 @@ function* useConversing(
   yield* ensure(() => until(writes));
   // The scripted fake does not await saves; snapshots keep its pending write
   // from adopting an identity before the queued asserting write publishes it.
+  // A read waits for the saves already queued when it was asked, so a later
+  // session resolution sees the record an earlier turn has already saved.
   const store: typeof disk = {
-    load: (key) => disk.load(key),
+    load: (key) => writes.then(() => disk.load(key)),
     save: (record) => {
       const snapshot = structuredClone(record);
       const saving = writes.then(() => disk.save(snapshot));
@@ -1119,12 +1125,27 @@ function* phases(): Operation<unknown> {
       };
     }
     if (holdWhen !== undefined) {
-      yield* spawn(function* (): Operation<void> {
-        yield* held.owner.message(said);
-      });
+      const sending = yield* spawn(() => held.owner.message(said));
       // The boundary, observed. Returning unwinds the owner, the conversation
-      // and the turn under the write that never happened.
-      yield* reached.operation;
+      // and the turn under the write that never happened. A message that
+      // settles first can no longer reach the boundary, so its answer is the
+      // phase's failure. Losing the race halts only the observer, not the
+      // owner still suspended under the held write.
+      const early = yield* race([
+        (function* (): Operation<undefined> {
+          yield* reached.operation;
+          return undefined;
+        })(),
+        (function* (): Operation<Result<SidekickTurn>> {
+          return yield* sending;
+        })(),
+      ]);
+      if (early !== undefined) {
+        const answer = early.ok
+          ? `completed with ${early.value.failure ?? "no failure"}`
+          : `refused: ${early.error.message}`;
+        throw new Error(`the message settled before the ${script.stop} boundary: ${answer}`);
+      }
       // Observed while the acknowledgement is still held: the provider took the
       // turn, and its own store already retains the native conversation it
       // created for it. Both are facts about the world this process is about to

packages/cli/tests/repl-sidekick-recovery.test.ts

@@ -1362,6 +1362,28 @@ describe("Tier R2 — cases 2, 5 and 6: a conversation across real process bound
 
     yield* root.discard();
   });
+
+  it("R2q fixture: a message that settles before its cut fails the phase rather than waiting", function* () {
+    const root = yield* createRepositoryRoot();
+    yield* runPhase("chat-open", root, "chat");
+
+    // One reply that reads nothing, so the exchange has one canonical Prompt
+    // and the second one this cut holds is never written.
+    let observed: unknown;
+    try {
+      yield* runPhase(
+        "chat-say",
+        root,
+        "chat",
+        script({ replies: ["<Output>Done.</Output>\n"], stop: "laterTurn" }),
+      );
+    } catch (error) {
+      observed = error;
+    }
+    expect(String(observed)).toContain("the message settled before the laterTurn boundary");
+
+    yield* root.discard();
+  });
 });
 
 describe("Tier R2 — a reservation binds what the writer has acknowledged", () => {