diff --git a/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts b/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts index b41822c16..165e886b8 100644 --- a/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts +++ b/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 { }; } if (holdWhen !== undefined) { - yield* spawn(function* (): Operation { - 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 { + yield* reached.operation; + return undefined; + })(), + (function* (): Operation> { + 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 diff --git a/packages/cli/tests/repl-sidekick-recovery.test.ts b/packages/cli/tests/repl-sidekick-recovery.test.ts index e69a8d07a..7f878baf9 100644 --- a/packages/cli/tests/repl-sidekick-recovery.test.ts +++ b/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: ["Done.\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", () => {