From 5541543a348ec793920ee02670cdbb1330e824ef Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 11 Sep 2026 15:18:27 +0200 Subject: [PATCH 1/8] =?UTF-8?q?=F0=9F=A4=96=20fix:=20admit=20fresh=20autom?= =?UTF-8?q?atic=20input=20after=20settled=20Stop?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replay the existing reviewed B layer on its corrected prerequisites: versioned settlement proof, original automatic admission CAS, guarded batch publication, and receipt-based budget and rollback accounting. Its existing CI fixtures remain part of this layer. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ --- .../agentSession.admissionGates.test.ts | 34 +- .../agentSession.autoCompaction.test.ts | 166 +++---- ...gentSession.compactionCancellation.test.ts | 431 ++++++++++++++---- .../agentSession.pinnedBudget.test.ts | 19 +- .../agentSession.preTurnMessages.test.ts | 9 +- .../agentSession.preparationAdmission.test.ts | 32 +- .../agentSession.preparedHistory.test.ts | 136 +++--- .../agentSession.queueDispatch.test.ts | 20 +- .../agentSession.scopedLifetimes.test.ts | 4 +- .../services/agentSession.tokenBudget.test.ts | 14 +- src/node/services/agentSession.ts | 156 ++++--- .../compactionCancellation.storage.test.ts | 108 ++++- src/node/services/compactionCancellation.ts | 64 ++- src/node/services/workspaceService.test.ts | 123 +++-- 14 files changed, 858 insertions(+), 458 deletions(-) diff --git a/src/node/services/agentSession.admissionGates.test.ts b/src/node/services/agentSession.admissionGates.test.ts index 1ca18a6e41..96b0ad3960 100644 --- a/src/node/services/agentSession.admissionGates.test.ts +++ b/src/node/services/agentSession.admissionGates.test.ts @@ -67,7 +67,7 @@ describe("AgentSession.sendMessage (admission gates)", () => { it("refuses at the pre-persist gate before any row lands when the epoch is stale", async () => { const workspaceId = "ws-epoch-prepersist"; const { session, historyService, streamMessage } = await createSessionHarness(workspaceId); - const appendMany = spyOn(historyService, "appendManyToHistory"); + const publication = spyOn(historyService, "acceptCompactionReplacement"); let acceptedCalls = 0; const result = await session.sendMessage( @@ -95,7 +95,7 @@ describe("AgentSession.sendMessage (admission gates)", () => { }); // Pre-acceptance refusal: nothing persisted, nothing accepted, no stream. expect(acceptedCalls).toBe(0); - expect(appendMany).not.toHaveBeenCalled(); + expect(publication).not.toHaveBeenCalled(); expect(streamMessage).not.toHaveBeenCalled(); const history = await historyService.getHistoryFromLatestBoundary(workspaceId); expect(history.success ? history.data : ["unexpected"]).toHaveLength(0); @@ -104,7 +104,18 @@ describe("AgentSession.sendMessage (admission gates)", () => { it("invokes the cancellation hook when the caller probe goes stale before acceptance", async () => { const workspaceId = "ws-caller-stale-cancel"; const { session, historyService, streamMessage } = await createSessionHarness(workspaceId); - const appendMany = spyOn(historyService, "appendManyToHistory"); + let published = false; + const publish = historyService.acceptCompactionReplacement.bind(historyService); + spyOn(historyService, "acceptCompactionReplacement").mockImplementationOnce( + (id, capture, operation, observer) => + publish(id, capture, operation, { + ...observer, + onCommitted: (receipt) => { + observer.onCommitted(receipt); + published = true; + }, + }) + ); const canceled: string[] = []; const result = await session.sendMessage( @@ -123,7 +134,7 @@ describe("AgentSession.sendMessage (admission gates)", () => { // which must roll the rows back AND surface the refusal through the cancellation hook — // a queued peer send's caller already returned success and this hook carries its budget // refund; without it the reservation would leak. - admissionStale: () => appendMany.mock.calls.length > 0, + admissionStale: () => published, onCanceled: (reason: string) => { canceled.push(reason); }, @@ -140,7 +151,18 @@ describe("AgentSession.sendMessage (admission gates)", () => { it("keeps the charge when a stale send's rollback did not commit", async () => { const workspaceId = "ws-caller-stale-rollback-failed"; const { session, historyService, streamMessage } = await createSessionHarness(workspaceId); - const appendMany = spyOn(historyService, "appendManyToHistory"); + let published = false; + const publish = historyService.acceptCompactionReplacement.bind(historyService); + spyOn(historyService, "acceptCompactionReplacement").mockImplementationOnce( + (id, capture, operation, observer) => + publish(id, capture, operation, { + ...observer, + onCommitted: (receipt) => { + observer.onCommitted(receipt); + published = true; + }, + }) + ); // Rollback deletion fails and the rows verifiably REMAIN: the cancellation hook must not // fire — a refunded reservation with durable rows would let the payload enter provider // context after a resume while no longer counting against the sender's budget. @@ -162,7 +184,7 @@ describe("AgentSession.sendMessage (admission gates)", () => { synthetic: true, }), ], - admissionStale: () => appendMany.mock.calls.length > 0, + admissionStale: () => published, onCanceled: (reason: string) => { canceled.push(reason); }, diff --git a/src/node/services/agentSession.autoCompaction.test.ts b/src/node/services/agentSession.autoCompaction.test.ts index 16827ebe86..5fa08e1ef1 100644 --- a/src/node/services/agentSession.autoCompaction.test.ts +++ b/src/node/services/agentSession.autoCompaction.test.ts @@ -1443,117 +1443,91 @@ describe("AgentSession on-send auto-compaction for synthetic guidance sends", () return predicate(); } - test("unresolved Stop refuses automatic guidance before creating a compaction continuation", async () => { - const workspaceId = "ws-unresolved-stop-guidance"; - const fixture = await createGuidanceHarness({ workspaceId }); - try { - expect(await fixture.session.interruptStream()).toEqual(Ok(undefined)); - const accepted = mock(() => undefined); + test.each([false, true])( + "applies compaction for fresh automatic guidance after settled Stop (%s)", + async (stopped) => { + const fixture = await createGuidanceHarness({ + workspaceId: "ws-auto-compaction-synthetic-guidance", + }); + + const monitor = (fixture.session as unknown as { compactionMonitor: CompactionMonitor }) + .compactionMonitor; + let firstCheck = true; + spyOn(monitor, "checkBeforeSend").mockImplementation(() => { + const high = firstCheck; + firstCheck = false; + return { + shouldShowWarning: high, + shouldForceCompact: high, + usagePercentage: high ? 95 : 1, + thresholdPercentage: 70, + contextTokens: high ? 95_000 : 1_000, + maxTokens: 100_000, + }; + }); + if (stopped) expect(await fixture.session.interruptStream()).toEqual(Ok(undefined)); + const result = await fixture.session.sendMessage( - "Guidance that would otherwise force compaction", + "Updated guidance from parent: focus on the failing tests.", { model: "openai:gpt-4o", agentId: "exec" }, { acceptanceOrigin: "automatic", synthetic: true, agentInitiated: true, startStreamInBackground: true, - onAccepted: accepted, } ); - expect(result.success).toBe(false); - expect(accepted).not.toHaveBeenCalled(); - expect(fixture.streamHistories).toEqual([]); - expect(await fixture.historyService.getHistoryFromLatestBoundary(workspaceId)).toEqual( - Ok([]) - ); - expect(fixture.session.isBusy()).toBe(false); - expect(fixture.session.queuedMessageEntryCount()).toBe(0); - } finally { - await fixture.session.dispose(); - } - }); - - test("applies compaction for fresh automatic guidance without Stop", async () => { - const fixture = await createGuidanceHarness({ - workspaceId: "ws-auto-compaction-synthetic-guidance", - }); - - const monitor = (fixture.session as unknown as { compactionMonitor: CompactionMonitor }) - .compactionMonitor; - let firstCheck = true; - spyOn(monitor, "checkBeforeSend").mockImplementation(() => { - const high = firstCheck; - firstCheck = false; - return { - shouldShowWarning: high, - shouldForceCompact: high, - usagePercentage: high ? 95 : 1, - thresholdPercentage: 70, - contextTokens: high ? 95_000 : 1_000, - maxTokens: 100_000, - }; - }); - - const result = await fixture.session.sendMessage( - "Updated guidance from parent: focus on the failing tests.", - { model: "openai:gpt-4o", agentId: "exec" }, - { - acceptanceOrigin: "automatic", - synthetic: true, - agentInitiated: true, - startStreamInBackground: true, - } - ); - expect(result.success).toBe(true); - - // First stream must carry the persisted compaction request. - await waitFor(() => fixture.streamHistories.length >= 1); - expect(fixture.streamHistories.length).toBeGreaterThanOrEqual(1); - const firstRequestHasCompactionRequest = fixture.streamHistories[0].some( - (message) => message.metadata?.muxMetadata?.type === "compaction-request" - ); - expect(firstRequestHasCompactionRequest).toBe(true); + expect(result.success).toBe(true); - // Compaction must complete: a boundary summary lands in durable history. - const boundaryLanded = await waitFor(async () => { - const historyResult = await fixture.historyService.getHistoryFromLatestBoundary( - "ws-auto-compaction-synthetic-guidance" + // First stream must carry the persisted compaction request. + await waitFor(() => fixture.streamHistories.length >= 1); + expect(fixture.streamHistories.length).toBeGreaterThanOrEqual(1); + const firstRequestHasCompactionRequest = fixture.streamHistories[0].some( + (message) => message.metadata?.muxMetadata?.type === "compaction-request" ); - return ( - historyResult.success && - historyResult.data.some((message) => message.metadata?.compactionBoundary === true) + expect(firstRequestHasCompactionRequest).toBe(true); + + // Compaction must complete: a boundary summary lands in durable history. + const boundaryLanded = await waitFor(async () => { + const historyResult = await fixture.historyService.getHistoryFromLatestBoundary( + "ws-auto-compaction-synthetic-guidance" + ); + return ( + historyResult.success && + historyResult.data.some((message) => message.metadata?.compactionBoundary === true) + ); + }); + expect(boundaryLanded).toBe(true); + + // The original guidance text is re-dispatched as the post-compaction follow-up. + const followUpDispatched = await waitFor(() => + fixture.streamHistories.some((history) => + history.some( + (message) => + message.role === "user" && + message.metadata?.muxMetadata?.type !== "compaction-request" && + message.parts.some( + (part) => part.type === "text" && part.text.includes("focus on the failing tests") + ) + ) + ) ); - }); - expect(boundaryLanded).toBe(true); + expect(followUpDispatched).toBe(true); - // The original guidance text is re-dispatched as the post-compaction follow-up. - const followUpDispatched = await waitFor(() => - fixture.streamHistories.some((history) => - history.some( - (message) => - message.role === "user" && - message.metadata?.muxMetadata?.type !== "compaction-request" && - message.parts.some( - (part) => part.type === "text" && part.text.includes("focus on the failing tests") - ) + expect( + fixture.events.some( + (event) => (event as { type?: string }).type === "auto-compaction-triggered" ) - ) - ); - expect(followUpDispatched).toBe(true); - - expect( - fixture.events.some( - (event) => (event as { type?: string }).type === "auto-compaction-triggered" - ) - ).toBe(true); - expect( - fixture.events.some( - (event) => (event as { type?: string }).type === "auto-compaction-completed" - ) - ).toBe(true); + ).toBe(true); + expect( + fixture.events.some( + (event) => (event as { type?: string }).type === "auto-compaction-completed" + ) + ).toBe(true); - await fixture.session.dispose(); - }); + await fixture.session.dispose(); + } + ); // Characterization: sends carrying preTurnMessages (family-message payloads) // intentionally skip on-send compaction. The trigger row references its diff --git a/src/node/services/agentSession.compactionCancellation.test.ts b/src/node/services/agentSession.compactionCancellation.test.ts index a4a452a8f8..7e4b2be683 100644 --- a/src/node/services/agentSession.compactionCancellation.test.ts +++ b/src/node/services/agentSession.compactionCancellation.test.ts @@ -20,11 +20,7 @@ import { CompactionCancellation, CompactionCancellationReadRefusedError, } from "./compactionCancellation"; -import { - createAgentSessionHarness, - createStartedTurnHandle, - type AgentSessionHarness, -} from "./agentSession.testHarness"; +import { createAgentSessionHarness, type AgentSessionHarness } from "./agentSession.testHarness"; const workspaceId = "cancellation-runtime"; const options = { model: "openai:gpt-4o", agentId: "exec" }; @@ -216,11 +212,11 @@ describe("compaction cancellation runtime", () => { } ); - test("unresolved Stop refuses fresh automatic input across restart until manual replacement", async () => { + test("settled Stop qualifies only durable fresh automatic input across restart", async () => { const h = await fixture(); expect(await h.session.interruptStream()).toEqual(Ok(undefined)); const stopped = await h.storage.read(); - expect(stopped).toMatchObject({ version: 1, scope: { kind: "unresolved" } }); + assert(stopped?.version === 2); await h.session.dispose(); const fresh = await createAgentSessionHarness({ workspaceId, @@ -228,86 +224,332 @@ describe("compaction cancellation runtime", () => { historyService: new HistoryService(h.config), }); fixtures.push(fresh); - const stream = spyOn(fresh.aiService, "streamMessage"); - const accepted = mock(() => undefined); expect( - ( - await fresh.session.sendMessage("fresh automatic input", options, { + await fresh.session.sendMessage("fresh automatic input", options, { + acceptanceOrigin: "automatic", + }) + ).toEqual(Ok(undefined)); + expect((await h.rows()).at(-1)?.metadata?.compactionReplacementNonce).toBe(stopped.nonce); + expect(await h.storage.read()).toBeNull(); + }); + + test.each(["invalid", "write failure", "canceled", "generation"] as const)( + "refused fresh automatic input preserves settled Stop (%s)", + async (failure) => { + const h = await fixture(); + expect(await h.session.interruptStream()).toEqual(Ok(undefined)); + const stopped = await h.storage.read(); + assert(stopped?.version === 2); + const cancel = new AbortController(); + if (failure === "canceled") cancel.abort(); + if (failure === "write failure") + spyOn(h.historyService, "acceptCompactionReplacement").mockResolvedValueOnce( + Err("write unavailable") + ); + if (failure === "generation") + await h.historyService.getContinuousCompactionJournal(workspaceId).advanceGeneration(); + const result = await h.session.sendMessage( + "refused automatic input", + failure === "invalid" ? { ...options, model: "invalid" } : options, + { + acceptanceOrigin: "automatic", + cancelSignal: cancel.signal, + } + ); + if (failure !== "canceled") expect(result.success).toBe(false); + expect(await h.storage.read()).toEqual(stopped); + expect(await h.rows()).toEqual([]); + expect(await h.session.resumeStream(options, { acceptanceOrigin: "automatic" })).toEqual( + Ok({ started: false }) + ); + expect(h.stream).not.toHaveBeenCalled(); + } + ); + + test.each([false, true])( + "automatic admission cannot replace a foreign Stop first observed after its gate (prior=%s)", + async (prior) => { + const h = await fixture(); + if (prior) expect(await h.session.interruptStream()).toEqual(Ok(undefined)); + const foreignHistory = new HistoryService(h.config); + const foreign = new CompactionCancellation( + foreignHistory.getCompactionCancellationStorage(workspaceId) + ); + const gate = h.session.isAutomaticSendBlocked.bind(h.session); + spyOn(h.session, "isAutomaticSendBlocked").mockImplementationOnce(async () => { + const blocked = await gate(); + expect(blocked).toBe(false); + await foreign.cancel({ settled: Promise.resolve(true) }); + return blocked; + }); + const accepted = mock(() => undefined); + const result = await h.session.sendMessage("already admitted automatic input", options, { + acceptanceOrigin: "automatic", + onAccepted: accepted, + }); + expect(result.success).toBe(false); + expect(await h.storage.read()).toMatchObject({ version: 2 }); + expect(await h.rows()).toEqual([]); + expect(accepted).not.toHaveBeenCalled(); + expect(h.stream).not.toHaveBeenCalled(); + // The same persisted frontier is legitimate for a genuinely later admission. + expect( + await h.session.sendMessage("fresh automatic input", options, { acceptanceOrigin: "automatic", + }) + ).toEqual(Ok(undefined)); + expect(await h.storage.read()).toBeNull(); + } + ); + + test.each([ + ["absent", "after early read"], + ["scoped V1", "after early read"], + ["absent", "before publication lock"], + ["scoped V1", "before publication lock"], + ] as const)( + "ordinary automatic publication fences a late foreign Stop from %s (%s)", + async (frontier, timing) => { + const h = await fixture(); + if (frontier === "scoped V1") { + await h.session.cancelCompaction(); + const stop = await h.storage.read(); + assert(stop); + await h.state.compactionCancellation.narrow(stop.nonce, { + id: "old-summary", + pendingFollowUp: { text: "canceled continuation" }, + }); + } + const foreignHistory = new HistoryService(h.config); + const foreign = new CompactionCancellation( + foreignHistory.getCompactionCancellationStorage(workspaceId) + ); + if (timing === "before publication lock") { + const publish = h.historyService.acceptCompactionReplacement.bind(h.historyService); + spyOn(h.historyService, "acceptCompactionReplacement").mockImplementationOnce( + async (...args) => { + // All request preparation completed against the old frontier; the CAS must recheck it. + await foreign.cancel({ settled: Promise.resolve(true) }); + return publish(...args); + } + ); + } else { + const state = h.session as unknown as { + readCompactionCancellation(): ReturnType; + }; + const read = state.readCompactionCancellation.bind(state); + let reads = 0; + spyOn(state, "readCompactionCancellation").mockImplementation(async () => { + const record = await read(); + if (++reads === 2) await foreign.cancel({ settled: Promise.resolve(true) }); + return record; + }); + } + const accepted = mock(() => undefined); + expect( + ( + await h.session.sendMessage("older automatic input", options, { + acceptanceOrigin: "automatic", + onAccepted: accepted, + }) + ).success + ).toBe(false); + expect(await h.rows()).toEqual([]); + expect(await h.storage.read()).toMatchObject({ version: 2 }); + expect(accepted).not.toHaveBeenCalled(); + expect(h.stream).not.toHaveBeenCalled(); + expect( + await h.session.sendMessage("fresh automatic input", options, { + acceptanceOrigin: "automatic", + }) + ).toEqual(Ok(undefined)); + expect(await h.storage.read()).toBeNull(); + } + ); + + test.each([false, true])( + "ordinary automatic receipt keeps rollback ownership (rollback fails=%s)", + async (rollbackFails) => { + const h = await fixture(); + const cancel = new AbortController(); + let rowsPersisted = false; + let budgetReserved = true; + const accepted = mock(() => undefined); + const canceled = mock(() => { + if (!rowsPersisted) budgetReserved = false; + }); + const publish = h.historyService.acceptCompactionReplacement.bind(h.historyService); + spyOn(h.historyService, "acceptCompactionReplacement").mockImplementationOnce( + async (...args) => { + const result = await publish(...args); + assert(result.success && result.data.kind === "accepted"); + expect(result.data.witness).toBeNull(); + expect(rowsPersisted).toBe(false); + cancel.abort(); + return result; + } + ); + if (rollbackFails) + spyOn(h.historyService, "deleteMessages").mockResolvedValueOnce( + Err("rollback unavailable") + ); + expect( + await h.session.sendMessage("ordinary trigger", options, { + acceptanceOrigin: "automatic", + cancelSignal: cancel.signal, onAccepted: accepted, + onCanceled: canceled, + preTurnMessages: [ + createMuxMessage("ordinary-payload", "assistant", "reserved payload", { + synthetic: true, + }), + ], + onPreTurnRowsPersisted: () => { + rowsPersisted = true; + }, + onAcceptedPreStreamFailure: () => { + if (!rowsPersisted) budgetReserved = false; + }, + }) + ).toEqual(Ok(undefined)); + expect(await h.rows()).toHaveLength(rollbackFails ? 2 : 0); + expect(rowsPersisted).toBe(rollbackFails); + expect(budgetReserved).toBe(rollbackFails); + expect(accepted).toHaveBeenCalledTimes(rollbackFails ? 1 : 0); + expect(canceled).toHaveBeenCalledTimes(rollbackFails ? 0 : 1); + } + ); + + test("generation-mismatched settled Stop blocks idle automatic reconciliation", async () => { + const h = await fixture(); + expect(await h.session.interruptStream()).toEqual(Ok(undefined)); + expect(await h.session.isAutomaticSendBlocked()).toBe(false); + await h.historyService.getContinuousCompactionJournal(workspaceId).advanceGeneration(); + expect(await h.session.isAutomaticSendBlocked()).toBe(true); + expect( + ( + await h.session.sendMessage("deferred input", options, { + acceptanceOrigin: "automatic", }) ).success ).toBe(false); - expect(accepted).not.toHaveBeenCalled(); - expect(stream).not.toHaveBeenCalled(); + expect(await h.session.isAutomaticSendBlocked()).toBe(true); expect(await h.rows()).toEqual([]); - expect(await h.storage.read()).toEqual(stopped); - expect(fresh.session.isBusy()).toBe(false); - expect(await fresh.session.sendMessage("manual replacement", options)).toEqual(Ok(undefined)); - expect(stream).toHaveBeenCalledTimes(1); - expect(await h.storage.read()).toBeNull(); - expect( - (await h.rows()).map((row) => - row.parts.map((part) => (part.type === "text" ? part.text : part.type)) - ) - ).toEqual([["manual replacement"]]); + expect(h.stream).not.toHaveBeenCalled(); }); - test("refused queued automatic ownership settles before idle and the manual successor", async () => { + test("a throwing acceptance observer cannot skip settled Stop retirement", async () => { const h = await fixture(); expect(await h.session.interruptStream()).toEqual(Ok(undefined)); - const entered = Promise.withResolvers(); - const release = Promise.withResolvers(); - const started = Promise.withResolvers(); - const accepted = mock(() => undefined); - let released = false; - const owner: Disposable = { - [Symbol.dispose]: () => { - released = true; - }, - }; - const failed = mock(async () => { - entered.resolve(); - await release.promise; - owner[Symbol.dispose](); + const stopped = await h.storage.read(); + assert(stopped?.version === 2); + let budgetReserved = true; + let rowsPersisted = false; + const accepted = mock(() => { + expect(rowsPersisted).toBe(true); + throw new Error("acceptance observer failed"); }); - h.stream.mockImplementation(() => { - expect(released).toBe(true); - started.resolve(); - return Promise.resolve(Ok(createStartedTurnHandle(h.session.closingSignal))); + const canceled = mock(() => undefined); + const failed = mock(() => { + if (!rowsPersisted) budgetReserved = false; }); - h.session.queueMessage("owned automatic wake", options, { - acceptanceOrigin: "automatic", - synthetic: true, - onAccepted: accepted, - onAcceptedPreStreamFailure: failed, + await nodeAssert.rejects( + h.session.sendMessage("accepted automatic input", options, { + acceptanceOrigin: "automatic", + onAccepted: accepted, + onCanceled: canceled, + onAcceptedPreStreamFailure: failed, + preTurnMessages: [ + createMuxMessage("peer-payload", "assistant", "reserved peer payload", { + synthetic: true, + }), + ], + onPreTurnRowsPersisted: () => { + rowsPersisted = true; + }, + }), + /acceptance observer failed/ + ); + expect(accepted).toHaveBeenCalledTimes(1); + expect(budgetReserved).toBe(true); + expect((await h.rows()).map((row) => row.id)).toContain("peer-payload"); + expect(canceled).not.toHaveBeenCalled(); + expect(failed).toHaveBeenCalledTimes(1); + expect(h.stream).not.toHaveBeenCalled(); + expect((await h.rows()).at(-1)?.metadata?.compactionReplacementNonce).toBe(stopped.nonce); + expect(await h.storage.read()).toBeNull(); + }); + + test("failed Stop retirement still delivers acceptance and retries its exact witness", async () => { + const h = await fixture(); + expect(await h.session.interruptStream()).toEqual(Ok(undefined)); + const stopped = await h.storage.read(); + assert(stopped?.version === 2); + const remove = fs.rmSync; + const failure = spyOn(fs, "rmSync").mockImplementation((file, options) => { + if (file === h.storage.path) throw new Error("retirement unavailable"); + return remove(file, options); }); - h.session.queueMessage("manual successor", options); - h.session.sendQueuedMessages(); - try { - await entered.promise; - expect(h.session.isBusy()).toBe(true); - expect(released).toBe(false); - expect(h.stream).not.toHaveBeenCalled(); - expect(accepted).not.toHaveBeenCalled(); - expect(await h.rows()).toEqual([]); - release.resolve(); - await started.promise; - await h.session.waitForIdle(); - expect(failed).toHaveBeenCalledTimes(1); - expect(released).toBe(true); - expect(h.session.queuedMessageEntryCount()).toBe(0); - expect(h.stream).toHaveBeenCalledTimes(1); - expect(await h.storage.read()).toBeNull(); - expect( - (await h.rows()).map((row) => - row.parts.map((part) => (part.type === "text" ? part.text : part.type)) - ) - ).toEqual([["manual successor"]]); - } finally { - release.resolve(); - await h.session.waitForIdle(); - } + const cancel = new AbortController(); + const accepted = mock(() => cancel.abort()); + const canceled = mock(() => undefined); + const failed = mock(() => undefined); + expect( + await h.session.sendMessage("accepted automatic input", options, { + acceptanceOrigin: "automatic", + cancelSignal: cancel.signal, + withdrawAcceptedOnCancel: true, + onAccepted: accepted, + onCanceled: canceled, + onAcceptedPreStreamFailure: failed, + }) + ).toEqual(Ok(undefined)); + expect(accepted).toHaveBeenCalledTimes(1); + expect(canceled).not.toHaveBeenCalled(); + expect(failed).toHaveBeenCalledTimes(1); + expect(h.stream).not.toHaveBeenCalled(); + expect((await h.rows()).at(-1)?.metadata?.compactionReplacementNonce).toBe(stopped.nonce); + expect(await h.storage.read()).toEqual(stopped); + expect(h.state.compactionCancellation.needsPersistence).toBe(true); + failure.mockRestore(); + expect(await h.session.isAutomaticSendBlocked()).toBe(false); + expect(await h.storage.read()).toBeNull(); + }); + + test("post-receipt automatic cancellation keeps accepted row and callback ownership", async () => { + const h = await fixture(); + expect(await h.session.interruptStream()).toEqual(Ok(undefined)); + const stopped = await h.storage.read(); + assert(stopped?.version === 2); + const cancel = new AbortController(); + const accepted = mock(() => undefined); + const canceled = mock(() => undefined); + const failed = mock(() => undefined); + const publish = h.historyService.acceptCompactionReplacement.bind(h.historyService); + spyOn(h.historyService, "acceptCompactionReplacement").mockImplementationOnce( + async (...args) => { + const result = await publish(...args); + assert(result.success && result.data.kind === "accepted"); + cancel.abort(); + return result; + } + ); + expect( + await h.session.sendMessage("accepted automatic input", options, { + acceptanceOrigin: "automatic", + cancelSignal: cancel.signal, + withdrawAcceptedOnCancel: true, + onAccepted: accepted, + onCanceled: canceled, + onAcceptedPreStreamFailure: failed, + }) + ).toEqual(Ok(undefined)); + expect(accepted).toHaveBeenCalledTimes(1); + expect(canceled).not.toHaveBeenCalled(); + expect(failed).toHaveBeenCalledTimes(1); + expect(h.stream).not.toHaveBeenCalled(); + expect((await h.rows()).at(-1)?.metadata?.compactionReplacementNonce).toBe(stopped.nonce); + expect(await h.storage.read()).toBeNull(); }); test.each([ @@ -389,7 +631,12 @@ describe("compaction cancellation runtime", () => { expect(h.state.compactionCancellation.blocksRecovery).toBe(false); const repaired = await h.storage.read(); assert(repaired); - if (cancellation) expect(repaired).toEqual(cancellation); + const settledGeneration = await h.historyService + .getContinuousCompactionJournal(workspaceId) + .captureGeneration(); + expect(repaired).toMatchObject({ version: 2, settledGeneration }); + if (cancellation) + expect(repaired).toEqual({ ...cancellation, version: 2, settledGeneration }); expect(await h.session.sendMessage("accepted replacement", options)).toEqual(Ok(undefined)); expect(h.stream).toHaveBeenCalledTimes(1); expect((await h.rows()).map((row) => row.id)).toContain("prior"); @@ -510,7 +757,11 @@ describe("compaction cancellation runtime", () => { expect(await h.state.compactionCancellation.retry()).toBe("applied"); const retried = await h.storage.read(); assert(retried); - if (record) expect(retried).toEqual(record); + const settledGeneration = await h.historyService + .getContinuousCompactionJournal(workspaceId) + .captureGeneration(); + expect(retried).toMatchObject({ version: 2, settledGeneration }); + if (record) expect(retried).toEqual({ ...record, version: 2, settledGeneration }); } ); @@ -614,7 +865,6 @@ describe("compaction cancellation runtime", () => { `${stale} during automatic %s append removes only stale rows`, async (kind) => { const h = await fixture(); - const chatPath = path.join(h.config.sessionsDir, workspaceId, "chat.jsonl"); let stopping: ReturnType | undefined; let foreign: ReturnType | undefined; let epochStale = false; @@ -627,26 +877,19 @@ describe("compaction cancellation runtime", () => { createMuxMessage("foreign", "assistant", "new unrelated row") ); }; - if (kind === "batch") { - const atomic = atomicWrite.default; - spyOn(atomicWrite, "default").mockImplementation( - new Proxy(atomic, { - async apply(target, receiver, args: Parameters) { - const result = await Reflect.apply(target, receiver, args); - if (args[0] === chatPath) supersede(); - return result; + const publish = h.historyService.acceptCompactionReplacement.bind(h.historyService); + spyOn(h.historyService, "acceptCompactionReplacement").mockImplementationOnce( + (id, capture, operation, observer) => + publish(id, capture, operation, { + ...observer, + onCommitted: (receipt) => { + observer.onCommitted(receipt); + // The actual journaled append committed; Stop/epoch loss still owns ordinary rollback. + supersede(); + return undefined; }, }) - ); - } else { - const append = fileIO.appendFile; - spyOn(fileIO, "appendFile").mockImplementation( - async (...args: Parameters) => { - await append(...args); - if (args[0] === chatPath) supersede(); - } - ); - } + ); const result = await h.session.sendMessage("stale automatic input", options, { acceptanceOrigin: "automatic", admissionEpochStale: () => epochStale, diff --git a/src/node/services/agentSession.pinnedBudget.test.ts b/src/node/services/agentSession.pinnedBudget.test.ts index 99aebd508b..1bb7f57ce4 100644 --- a/src/node/services/agentSession.pinnedBudget.test.ts +++ b/src/node/services/agentSession.pinnedBudget.test.ts @@ -923,18 +923,23 @@ describe("pinned full-payload rollover admission", () => { } ); test.each([false, true])( - "cancellation during rollover append follows the durable rollback outcome (rollback fails=%s)", + "cancellation after rollover publication follows the durable rollback outcome (rollback fails=%s)", async (rollbackFails) => { const fixture = await setup("small"); const { h, historyService, before, start, assembly, modelCleanup } = fixture; const entered = Promise.withResolvers(); const release = Promise.withResolvers(); - const append = historyService.appendManyToHistory.bind(historyService); - spyOn(historyService, "appendManyToHistory").mockImplementationOnce(async (...args) => { - entered.resolve(); - await release.promise; - return append(...args); - }); + const publish = historyService.acceptCompactionReplacement.bind(historyService); + spyOn(historyService, "acceptCompactionReplacement").mockImplementationOnce( + async (...args) => { + const result = await publish(...args); + expect(result).toEqual(Ok({ kind: "accepted", witness: null })); + // Hold after the real receipt: cancellation must exercise rollback of durable rows. + entered.resolve(); + await release.promise; + return result; + } + ); const rollback = spyOn(historyService, "deleteMessages"); if (rollbackFails) rollback.mockResolvedValueOnce(Err("injected durable rollback failure")); const controller = new AbortController(); diff --git a/src/node/services/agentSession.preTurnMessages.test.ts b/src/node/services/agentSession.preTurnMessages.test.ts index 115ca1222a..0d9ef7e287 100644 --- a/src/node/services/agentSession.preTurnMessages.test.ts +++ b/src/node/services/agentSession.preTurnMessages.test.ts @@ -69,7 +69,7 @@ describe("AgentSession.sendMessage (preTurnMessages)", () => { timestamp: 1, synthetic: true, }); - const appendMany = spyOn(historyService, "appendManyToHistory"); + const publication = spyOn(historyService, "acceptCompactionReplacement"); const appendOne = spyOn(historyService, "appendToHistory"); const result = await session.sendMessage( @@ -86,8 +86,9 @@ describe("AgentSession.sendMessage (preTurnMessages)", () => { // r32: payload + user row land in ONE durable write — separate appends // left a crash window that stranded the payload without its turn. - expect(appendMany).toHaveBeenCalledTimes(1); - expect(appendMany.mock.calls[0]?.[1]).toHaveLength(2); + expect(publication).toHaveBeenCalledTimes(1); + const operation = publication.mock.calls[0]?.[2]; + expect(operation?.kind === "append" && operation.messages).toHaveLength(2); expect(appendOne.mock.calls.filter(([, message]) => message.role === "user")).toHaveLength(0); const history = await historyService.getHistoryFromLatestBoundary(workspaceId); @@ -111,7 +112,7 @@ describe("AgentSession.sendMessage (preTurnMessages)", () => { synthetic: true, }); - spyOn(historyService, "appendManyToHistory").mockImplementation(() => + spyOn(historyService, "acceptCompactionReplacement").mockImplementation(() => Promise.resolve(Err("simulated batch append failure")) ); diff --git a/src/node/services/agentSession.preparationAdmission.test.ts b/src/node/services/agentSession.preparationAdmission.test.ts index c42e29bfbd..14eb2c4bef 100644 --- a/src/node/services/agentSession.preparationAdmission.test.ts +++ b/src/node/services/agentSession.preparationAdmission.test.ts @@ -186,7 +186,7 @@ describe("preparation admission", () => { const entered = Promise.withResolvers(); const release = Promise.withResolvers(); const started = Promise.withResolvers(); - const append = spyOn(h.historyService, "appendToHistory"); + const append = spyOn(h.historyService, "acceptCompactionReplacement"); if (failure === "return") append.mockResolvedValueOnce(Err("disk failure")); else append.mockRejectedValueOnce(new Error("disk failure")); let cleanups = 0; @@ -278,13 +278,15 @@ describe("preparation admission", () => { const releaseAppend = Promise.withResolvers(); const provider = Promise.withResolvers(); const releaseProvider = Promise.withResolvers(); - const append = h.historyService.appendToHistory.bind(h.historyService); - spyOn(h.historyService, "appendToHistory").mockImplementationOnce(async (...args) => { - const result = await append(...args); - appended.resolve(); - await releaseAppend.promise; - return result; - }); + const append = h.historyService.acceptCompactionReplacement.bind(h.historyService); + spyOn(h.historyService, "acceptCompactionReplacement").mockImplementationOnce( + async (...args) => { + const result = await append(...args); + appended.resolve(); + await releaseAppend.promise; + return result; + } + ); const stream = spyOn(h.aiService, "streamMessage").mockImplementation(async () => { provider.resolve(); await releaseProvider.promise; @@ -333,12 +335,14 @@ describe("preparation admission", () => { maxTokens: 100000, }); let stale = false; - const append = h.historyService.appendToHistory.bind(h.historyService); - spyOn(h.historyService, "appendToHistory").mockImplementationOnce(async (...args) => { - const result = await append(...args); - stale = true; - return result; - }); + const append = h.historyService.acceptCompactionReplacement.bind(h.historyService); + spyOn(h.historyService, "acceptCompactionReplacement").mockImplementationOnce( + async (...args) => { + const result = await append(...args); + stale = true; + return result; + } + ); const stream = spyOn(h.aiService, "streamMessage"); const result = await h.session.sendMessage("deferred question", options, { acceptanceOrigin: "automatic", diff --git a/src/node/services/agentSession.preparedHistory.test.ts b/src/node/services/agentSession.preparedHistory.test.ts index c8626ae496..41e9845726 100644 --- a/src/node/services/agentSession.preparedHistory.test.ts +++ b/src/node/services/agentSession.preparedHistory.test.ts @@ -101,76 +101,78 @@ describe("prepared history publication", () => { } ); - test.each( - (["skill", "prompt", "trigger"] as const).flatMap((failure) => - (["result", "rejection"] as const).map((outcome) => ({ failure, outcome })) - ) - )( - "a failed automatic $failure append ($outcome) rolls back only this attempt's previously published rows", - async ({ failure, outcome }) => { - const h = await fixture(); - const foreign = createMuxMessage("foreign", "assistant", "concurrent input"); - const earlier = ["file", "skill", "prompt"].slice( - 0, - ["skill", "prompt", "trigger"].indexOf(failure) + 1 - ); - const append = h.historyService.appendToHistory.bind(h.historyService); - const appends = spyOn(h.historyService, "appendToHistory").mockImplementationOnce( - async (...args) => { - const result = await append(...args); - expect(result).toEqual(Ok(undefined)); - // The real prefix released its lock; a foreign writer now lands before the failure. - expect(await append(workspaceId, foreign)).toEqual(Ok(undefined)); - return result; - } - ); - for (const _row of earlier.slice(1)) appends.mockImplementationOnce(append); - // Disk failures use Result Err; an unexpected service rejection must also retire - // already-published prefixes without deleting a concurrent writer's row. - if (outcome === "result") appends.mockResolvedValueOnce(Err("injected write failure")); - else appends.mockRejectedValueOnce(new Error("injected write failure")); - const start = spyOn(h.aiService, "streamMessage"); - const result = await h.session - .sendMessage("inspect input", options, { acceptanceOrigin: "automatic" }) - .catch((error: unknown) => error); - expect(appends.mock.calls.slice(0, -1).map(([, row]) => row.id)).toEqual(earlier); - expect(foreign.metadata?.historySequence).toBe(1); - expect(appends).toHaveBeenCalledTimes(earlier.length + 1); - expect((await h.rows()).map((row) => row.id)).toEqual([foreign.id]); - expect(result).toMatchObject({ success: false, error: { raw: "injected write failure" } }); - expect(start).not.toHaveBeenCalled(); - } - ); + test("a failed automatic batch leaves foreign history and publishes no owned prefixes", async () => { + const h = await fixture(); + const foreign = createMuxMessage("foreign", "assistant", "concurrent input"); + const publish = spyOn(h.historyService, "acceptCompactionReplacement").mockImplementationOnce( + async (_id, _capture, operation) => { + expect(await h.rows()).toEqual([]); + assert(operation.kind === "append"); + expect(operation.preserveCancellation).toBe(true); + expect(operation.messages.slice(0, -1).map((row) => row.id)).toEqual([ + "file", + "skill", + "prompt", + ]); + expect(await h.historyService.appendToHistory(workspaceId, foreign)).toEqual(Ok(undefined)); + // The batch API reports disk failures as Err; no per-prefix publication exists here. + return Err("injected batch write failure"); + } + ); + expect( + (await h.session.sendMessage("inspect input", options, { acceptanceOrigin: "automatic" })) + .success + ).toBe(false); + expect(publish).toHaveBeenCalledTimes(1); + expect((await h.rows()).map((row) => row.id)).toEqual([foreign.id]); + expect(h.stream).not.toHaveBeenCalled(); + }); - test.each(["file", "skill", "prompt", "trigger"])( - "automatic cancellation after %s publication still runs the existing rollback checkpoint", - async (after) => { - const h = await fixture(); - const controller = new AbortController(); - const canceled = mock(() => undefined); - const accepted = mock(() => undefined); - const append = h.historyService.appendToHistory.bind(h.historyService); - spyOn(h.historyService, "appendToHistory").mockImplementation(async (id, row) => { - const result = await append(id, row); - if (row.id === after || (after === "trigger" && row.metadata?.synthetic !== true)) + test("automatic cancellation after the batch receipt rolls back only its own rows", async () => { + const h = await fixture(); + const foreign = createMuxMessage("foreign", "assistant", "concurrent input"); + expect(await h.historyService.appendToHistory(workspaceId, foreign)).toEqual(Ok(undefined)); + const controller = new AbortController(); + const canceled = mock(() => undefined); + const accepted = mock(() => undefined); + const publish = h.historyService.acceptCompactionReplacement.bind(h.historyService); + const publication = spyOn( + h.historyService, + "acceptCompactionReplacement" + ).mockImplementationOnce(async (id, capture, operation, observer) => { + assert(operation.kind === "append"); + expect(operation.preserveCancellation).toBe(true); + const result = await publish(id, capture, operation, { + ...observer, + onCommitted: (receipt) => { + // Record the actual durable batch before cancellation exercises ordinary rollback. + observer.onCommitted(receipt); controller.abort(); - return result; + }, }); - const start = spyOn(h.aiService, "streamMessage"); - expect( - await h.session.sendMessage("inspect input", options, { - acceptanceOrigin: "automatic", - cancelSignal: controller.signal, - onCanceled: canceled, - onAccepted: accepted, - }) - ).toEqual(Ok(undefined)); - expect(await h.rows()).toEqual([]); - expect(canceled).toHaveBeenCalledTimes(1); - expect(accepted).not.toHaveBeenCalled(); - expect(start).not.toHaveBeenCalled(); - } - ); + expect(result).toEqual(Ok({ kind: "accepted", witness: null })); + const rows = await h.rows(); + expect(rows.map((row) => row.id)).toEqual([ + foreign.id, + ...operation.messages.map((row) => row.id), + ]); + expect(rows.map((row) => row.metadata?.historySequence)).toEqual([0, 1, 2, 3, 4]); + return result; + }); + expect( + await h.session.sendMessage("inspect input", options, { + acceptanceOrigin: "automatic", + cancelSignal: controller.signal, + onCanceled: canceled, + onAccepted: accepted, + }) + ).toEqual(Ok(undefined)); + expect(publication).toHaveBeenCalledTimes(1); + expect((await h.rows()).map((row) => row.id)).toEqual([foreign.id]); + expect(canceled).toHaveBeenCalledTimes(1); + expect(accepted).not.toHaveBeenCalled(); + expect(h.stream).not.toHaveBeenCalled(); + }); test.each(["skill", "prompt"] as const)( "manual %s materialization failure leaves no orphaned prefixes or accepted Stop", diff --git a/src/node/services/agentSession.queueDispatch.test.ts b/src/node/services/agentSession.queueDispatch.test.ts index 1f0542ce03..0a6857f64b 100644 --- a/src/node/services/agentSession.queueDispatch.test.ts +++ b/src/node/services/agentSession.queueDispatch.test.ts @@ -573,7 +573,7 @@ describe("AgentSession queued message tool-call dispatch", () => { workspaceId, aiServiceOverrides: { streamMessage }, }); - const append = spyOn(historyService, "appendToHistory"); + const append = spyOn(historyService, "acceptCompactionReplacement"); if (failureKind === "returned error") { append.mockResolvedValueOnce(Err("disk unavailable")); } else { @@ -1365,7 +1365,7 @@ describe("AgentSession queued message tool-call dispatch", () => { workspaceId, captureEvents: true, }); - const originalAppend = historyService.appendToHistory.bind(historyService); + const originalAppend = historyService.acceptCompactionReplacement.bind(historyService); let markAppendStarted: () => void = () => undefined; const appendStarted = new Promise((resolve) => { markAppendStarted = resolve; @@ -1374,7 +1374,7 @@ describe("AgentSession queued message tool-call dispatch", () => { const appendRelease = new Promise((resolve) => { releaseAppend = resolve; }); - const appendSpy = spyOn(historyService, "appendToHistory").mockImplementation( + const appendSpy = spyOn(historyService, "acceptCompactionReplacement").mockImplementation( async (...args) => { markAppendStarted(); await appendRelease; @@ -1448,7 +1448,7 @@ describe("AgentSession queued message tool-call dispatch", () => { workspaceId, aiServiceOverrides: { streamMessage }, }); - const originalAppend = historyService.appendToHistory.bind(historyService); + const originalAppend = historyService.acceptCompactionReplacement.bind(historyService); let markAppendStarted: () => void = () => undefined; const appendStarted = new Promise((resolve) => { markAppendStarted = resolve; @@ -1457,11 +1457,12 @@ describe("AgentSession queued message tool-call dispatch", () => { const appendRelease = new Promise((resolve) => { releaseAppend = resolve; }); - const appendSpy = spyOn(historyService, "appendToHistory").mockImplementation( + const appendSpy = spyOn(historyService, "acceptCompactionReplacement").mockImplementation( async (...args) => { + const result = await originalAppend(...args); markAppendStarted(); await appendRelease; - return originalAppend(...args); + return result; } ); const deleteMessagesSpy = spyOn(historyService, "deleteMessages").mockResolvedValue( @@ -1529,7 +1530,7 @@ describe("AgentSession queued message tool-call dispatch", () => { test("verifies a committed rollback when batch deletion reports a post-write failure", async () => { const workspaceId = "queue-dispatch-cancel-post-write-failure"; const { session, cleanup, historyService } = await createAgentSessionHarness({ workspaceId }); - const originalAppend = historyService.appendToHistory.bind(historyService); + const originalAppend = historyService.acceptCompactionReplacement.bind(historyService); const originalDeleteMessages = historyService.deleteMessages.bind(historyService); let markAppendStarted: () => void = () => undefined; const appendStarted = new Promise((resolve) => { @@ -1539,11 +1540,12 @@ describe("AgentSession queued message tool-call dispatch", () => { const appendRelease = new Promise((resolve) => { releaseAppend = resolve; }); - const appendSpy = spyOn(historyService, "appendToHistory").mockImplementation( + const appendSpy = spyOn(historyService, "acceptCompactionReplacement").mockImplementation( async (...args) => { + const result = await originalAppend(...args); markAppendStarted(); await appendRelease; - return originalAppend(...args); + return result; } ); const deleteMessagesSpy = spyOn(historyService, "deleteMessages").mockImplementation( diff --git a/src/node/services/agentSession.scopedLifetimes.test.ts b/src/node/services/agentSession.scopedLifetimes.test.ts index 4d5a27ec28..bed5db1aaf 100644 --- a/src/node/services/agentSession.scopedLifetimes.test.ts +++ b/src/node/services/agentSession.scopedLifetimes.test.ts @@ -131,7 +131,9 @@ describe("AgentSession scoped turn lifetimes", () => { const entered = Promise.withResolvers(); const release = Promise.withResolvers(); const h = await createAgentSessionHarness({ workspaceId, appFiberScope }); - spyOn(h.historyService, "appendToHistory").mockResolvedValueOnce(Err("disk unavailable")); + spyOn(h.historyService, "acceptCompactionReplacement").mockResolvedValueOnce( + Err("disk unavailable") + ); let closing: Promise | undefined; let closed = false; const close = (): void => { diff --git a/src/node/services/agentSession.tokenBudget.test.ts b/src/node/services/agentSession.tokenBudget.test.ts index 35634631fa..a41bf73420 100644 --- a/src/node/services/agentSession.tokenBudget.test.ts +++ b/src/node/services/agentSession.tokenBudget.test.ts @@ -463,12 +463,14 @@ describe("AgentSession token-budget lifecycle", () => { const before = await allRows(h); const controller = new AbortController(); const cancelState = { canceledBeforeAcceptance: false }; - const append = h.historyService.appendToHistory.bind(h.historyService); - spyOn(h.historyService, "appendToHistory").mockImplementationOnce(async (id, row) => { - const result = await append(id, row); - controller.abort(); - return result; - }); + const append = h.historyService.acceptCompactionReplacement.bind(h.historyService); + spyOn(h.historyService, "acceptCompactionReplacement").mockImplementationOnce( + async (...args) => { + const result = await append(...args); + controller.abort(); + return result; + } + ); expect( ( await h.session.sendMessage("Cancel after persistence", options, { diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index cd3b6a3368..95fe6a1a0d 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -3516,77 +3516,65 @@ export class AgentSession { const persistedCancelableMessageIds: string[] = []; const stagedPrefixes: MuxMessage[] = []; let replacementCapture: CompactionReplacementCapture | undefined; + let automaticReplacement = false; let replacementCommitted = false; - // Manual prefixes and their trigger form one accepted replacement. Optional context alone - // must never retire Stop, and cancellation after the receipt cannot erase the user's row. + // Prefixes and their trigger publish together against the original Stop frontier. Optional + // context alone cannot replace Stop; only replacement receipts close the rollback horizon. const publishPreparedHistory = async ( publication: | { kind: "prefix"; message: MuxMessage } | { kind: "trigger"; messages: MuxMessage[] } ): Promise> => { const messages = publication.kind === "prefix" ? [publication.message] : publication.messages; - const resetBatch = [...stagedPrefixes, ...messages].some( - (row) => row.metadata?.contextBoundaryKind === "reset" - ); - // Only reset prefixes wait for their actual trigger. Ordinary A prefixes retain their - // existing publication/rollback path; a reset alone cannot authorize replacement. - if (manualReplacement || resetBatch) { - if (publication.kind === "prefix") { - stagedPrefixes.push(...messages); - return Ok(undefined); - } - attempt.inputPublication = messages.at(-1); - const capture = replacementCapture ?? attempt.admissionCapture; - assert(capture, "Publication requires its original admission capture"); - const batch = [...stagedPrefixes, ...messages]; - const accepted = await this.historyService - .acceptCompactionReplacement( - this.workspaceId, - capture, - { - kind: "append", - messages: batch, - ...(!manualReplacement ? { preserveCancellation: true as const } : {}), - }, - { - isCurrent: () => - !isAdmissionStale() && !shutdownRefusesBeforePersist() && !cancelSignal?.aborted, - onContextResetCommitted: (predecessor, successor) => { - this.advanceOwnedCompactionAdmission(predecessor, successor); - if (attempt.admissionCapture) Object.assign(attempt.admissionCapture, successor); - }, - onCommitted: () => { - if (manualReplacement) { - replacementCommitted = true; - attempt.durability = "durable"; - } else persistedCancelableMessageIds.push(...batch.map((row) => row.id)); - return undefined; - }, + if (publication.kind === "prefix") { + stagedPrefixes.push(...messages); + return Ok(undefined); + } + const replacesCancellation = manualReplacement || automaticReplacement; + const batch = [...stagedPrefixes, ...messages]; + attempt.inputPublication = messages.at(-1); + assert(replacementCapture, "Publication requires its admission capture"); + const accepted = await this.historyService.acceptCompactionReplacement( + this.workspaceId, + replacementCapture, + { + kind: "append", + messages: batch, + ...(!replacesCancellation ? { preserveCancellation: true as const } : {}), + }, + { + isCurrent: () => + !isAdmissionStale() && !shutdownRefusesBeforePersist() && !cancelSignal?.aborted, + onCommitted: () => { + if (replacesCancellation) { + replacementCommitted = true; + attempt.durability = manualReplacement ? "durable" : "accepted"; + // Replacement is irrevocable before retirement or fallible acceptance observers. + // Correlated senders must not refund an already accepted prefix. + if ((internal?.preTurnMessages?.length ?? 0) > 0) + internal?.onPreTurnRowsPersisted?.(); + } else { + // Ordinary automatic publication only fences the Stop frontier; cancellation still + // owns rollback until the existing acceptance path closes that horizon. + persistedCancelableMessageIds.push(...batch.map((row) => row.id)); } - ) - .catch((error: unknown) => Err(getErrorMessage(error))); - if (!accepted.success) return accepted; - if (accepted.data.kind !== "accepted") { - if (await cancelBeforeAcceptance()) return Ok(undefined); - return Err(CONTEXT_MUTATION_SEND_BLOCKED_MESSAGE); + return undefined; + }, } + ); + if (!accepted.success) return accepted; + if (accepted.data.kind !== "accepted") { + // A canceled ordinary append can now refuse under the publication lock before writing. + // Its caller still owns cancellation notification and reservation release. + if (await cancelBeforeAcceptance()) return Ok(undefined); + return Err(CONTEXT_MUTATION_SEND_BLOCKED_MESSAGE); + } + if (replacesCancellation) { // Retirement takes the same lock; join it only after publication releases that lock. - if (manualReplacement) { - await this.retireCompactionReplacement(accepted.data.witness, attempt.admissionCapture); - if ((internal?.preTurnMessages?.length ?? 0) > 0) internal?.onPreTurnRowsPersisted?.(); - } - return Ok(undefined); + await this.retireCompactionReplacement(accepted.data.witness); + if (!manualReplacement) await internal?.onAccepted?.(); } - if (await this.isAutomaticSendBlocked()) return Err(CONTEXT_MUTATION_SEND_BLOCKED_MESSAGE); - // Preserve the same rollback checkpoints for unexpected service rejection as for - // ordinary Result failures; earlier prefixes may already be durable. - const result = await ( - messages.length === 1 - ? this.historyService.appendToHistory(this.workspaceId, messages[0]) - : this.historyService.appendManyToHistory(this.workspaceId, messages) - ).catch((error: unknown) => Err(getErrorMessage(error))); - if (result.success) persistedCancelableMessageIds.push(...messages.map((row) => row.id)); - return result; + return Ok(undefined); }; // Roll back synthetic snapshots if the invoking user row fails to persist, or // later provider requests could consume orphaned context. @@ -3672,6 +3660,11 @@ export class AgentSession { return Ok(undefined); } + // Capture before the first automatic gate: a foreign Stop discovered during preparation + // belongs to a later admission and cannot grant this attempt replacement authority. + const automaticCapture = manualReplacement + ? undefined + : await this.historyService.captureCompactionReplacement(this.workspaceId); if (manualReplacement) { await this.readCompactionCancellation("manual"); const stopAdmission = attempt.queuedStopAdmission; @@ -3687,6 +3680,20 @@ export class AgentSession { } else if (await this.isAutomaticSendBlocked()) { return Err(createUnknownSendMessageError(CONTEXT_MUTATION_SEND_BLOCKED_MESSAGE)); } + if (!manualReplacement) { + const record = await this.readCompactionCancellation(); + if (!automaticCapture?.success) + return Err(createUnknownSendMessageError(CONTEXT_MUTATION_SEND_BLOCKED_MESSAGE)); + replacementCapture = automaticCapture.data; + if (record?.version === 2) { + if ( + replacementCapture.nonce !== record.nonce || + replacementCapture.generation !== record.settledGeneration + ) + return Err(createUnknownSendMessageError(CONTEXT_MUTATION_SEND_BLOCKED_MESSAGE)); + automaticReplacement = true; + } + } if (isAdmissionStale()) return refuseBeforeAcceptance( createUnknownSendMessageError(CONTEXT_MUTATION_SEND_BLOCKED_MESSAGE) @@ -7307,7 +7314,7 @@ export class AgentSession { async cancelCompaction( retainUntilReplacement = false, - settled?: Promise, + settled?: Promise, options?: { fullHistoryDeletion?: true; onCaptured?: (capture: CompactionReplacementCapture) => void; @@ -7372,19 +7379,26 @@ export class AgentSession { async isAutomaticSendBlocked(): Promise { const record = await this.readCompactionCancellation(); if (this.compactionCancellation.blocksRecovery || record?.retainUntilReplacement) return true; - // Scoped Stop can distinguish its canceled handoff from later automatic input. - // Keep that handoff identifiable before any fresh request can hide the summary. - if (record?.scope.kind === "unresolved") { + if (record?.version === 2) { + const captured = await this.historyService.captureCompactionReplacement(this.workspaceId); + // A reset can advance the journal before its replacement commits. Keep monitor attention + // deferred in that state instead of repeatedly retrying an admission that must refuse. + return ( + !captured.success || + captured.data.nonce !== record.nonce || + captured.data.generation !== record.settledGeneration || + this.compactionCancellation.blocksRecovery + ); + } + // New monitor/family input remains automatic after ordinary Stop. Keep the canceled + // handoff identifiable when this fresh input hides its summary from tail-only recovery. + if (record?.version === 1 && record.scope.kind === "unresolved") { const history = await this.historyService.getHistoryFromLatestBoundary(this.workspaceId); if (!history.success) throw new Error(history.error); // A Stop admitted during the read can be waiting for this caller's terminal policy. if (this.compactionCancellation.blocksRecovery) return true; const summary = history.data.map(pendingCompactionSummary).findLast((entry) => entry != null); if (summary) await this.compactionCancellation.narrow(record.nonce, summary); - // An unresolved Stop has no durable settlement proof. Refuse fresh automatic input - // before acceptance so it cannot create a compaction continuation that recovery blocks. - // Explicit manual replacement remains the recovery path for this conservative layer. - else return true; } return this.compactionCancellation.blocksRecovery; } @@ -7417,7 +7431,8 @@ export class AgentSession { onCompactionCanceled?: (capture: CompactionReplacementCapture) => void; }): Promise { this.assertNotDisposed("interruptStream"); - const settled = Promise.withResolvers(); + const settled = Promise.withResolvers(); + let physicallyStopped = false; const cancellation = options?.soft || options?.preserveCompactionIntent ? undefined @@ -7451,9 +7466,10 @@ export class AgentSession { }) .then(async (result) => { if (result.success) await interruptedPolicy; + physicallyStopped = result.success; return result; }) - .finally(() => settled.resolve()); + .finally(() => settled.resolve(physicallyStopped)); const canceled = await cancellation; if (!stopResult.success) { return Err(stopResult.error); diff --git a/src/node/services/compactionCancellation.storage.test.ts b/src/node/services/compactionCancellation.storage.test.ts index 07d6d08682..73748a3b0e 100644 --- a/src/node/services/compactionCancellation.storage.test.ts +++ b/src/node/services/compactionCancellation.storage.test.ts @@ -312,7 +312,52 @@ describe("inactive real cancellation storage", () => { } ); - it("settlement still waits and cleans late recovery", async () => { + it.each(["joined", "unjoined", "failed", "retained", "scoped"] as const)( + "settled proof requires successful exact ordinary Stop (%s)", + async (kind) => { + const entered = Promise.withResolvers(); + const settled = Promise.withResolvers(); + const stopping = state.cancel({ + retainUntilReplacement: kind === "retained", + onCaptured: () => entered.resolve(), + ...(kind === "unjoined" ? {} : { settled: settled.promise }), + }); + try { + await entered.promise; + const initial = await storage.read(); + assert(initial); + expect(initial.version).toBe(1); + if (kind === "scoped") { + const other = new CompactionCancellation( + new FileCompactionCancellationStorage(foreign, workspaceId) + ); + await other.read(); + await other.narrow(initial.nonce, summary); + } + settled.resolve(kind !== "failed"); + expect(await stopping).toBe("applied"); + const persisted = await new FileCompactionCancellationStorage(foreign, workspaceId).read(); + expect(persisted?.nonce).toBe(initial.nonce); + expect(persisted?.version).toBe(kind === "joined" ? 2 : 1); + if (kind === "joined") { + expect(persisted?.settledGeneration).toBe( + await foreign.getContinuousCompactionJournal(workspaceId).captureGeneration() + ); + expect( + await new CompactionCancellation( + new FileCompactionCancellationStorage(foreign, workspaceId) + ).read() + ).toEqual(persisted); + } else expect(persisted?.settledGeneration).toBeUndefined(); + if (kind === "scoped") expect(persisted?.scope.kind).toBe("summary"); + } finally { + settled.resolve(false); + await stopping; + } + } + ); + + it("a legacy void settlement still waits and cleans late recovery without granting proof", async () => { const entered = Promise.withResolvers(); const settled = Promise.withResolvers(); const neutralize = h.historyService.neutralizeCompactionRecoveryUnderHistoryLock.bind( @@ -349,12 +394,61 @@ describe("inactive real cancellation storage", () => { type: "compaction-summary", }); expect(await storage.read()).toMatchObject({ version: 1, scope: { kind: "unresolved" } }); + expect((await storage.read())?.settledGeneration).toBeUndefined(); } finally { settled.resolve(); await stopping; } }); + it("a failed settled-proof write retains unresolved Stop debt until retry", async () => { + const rename = nodeFs.renameSync; + let publications = 0; + const failure = spyOn(nodeFs, "renameSync").mockImplementation((source, destination) => { + if (destination === storage.path && ++publications === 2) { + throw new Error("settled proof write failed"); + } + return rename(source, destination); + }); + await assert.rejects( + state.cancel({ settled: Promise.resolve(true) }), + /settled proof write failed/ + ); + expect(publications).toBe(2); + const stopped = await storage.read(); + assert(stopped); + expect(stopped.version).toBe(1); + expect(state.blocksRecovery).toBe(true); + failure.mockRestore(); + expect(await state.retry()).toBe("applied"); + expect(await storage.read()).toEqual({ + ...stopped, + version: 2, + settledGeneration: await h.historyService + .getContinuousCompactionJournal(workspaceId) + .captureGeneration(), + }); + expect(state.blocksRecovery).toBe(false); + }); + + it("a displaced settled-proof write preserves a foreign Stop and generation", async () => { + const successor = { ...record("foreign-successor"), retainUntilReplacement: true }; + const generationPath = path.join(sessionDir, CONTINUOUS_COMPACTION_GENERATION_FILE); + const lockPath = historyWriteLockPath(h.config.rootDir, workspaceId); + let stages = 0; + afterCompactionStaging(storage.path, () => { + if (++stages !== 2) return; + nodeFs.writeFileSync(lockPath, `${process.pid}:foreign-holder`); + nodeFs.writeFileSync(storage.path, JSON.stringify(successor)); + nodeFs.writeFileSync(generationPath, "foreign-generation"); + }); + await assert.rejects(state.cancel({ settled: Promise.resolve(true) }), /no longer owned/); + expect(stages).toBe(2); + expect(state.blocksRecovery).toBe(true); + expect(await storage.read()).toEqual(successor); + expect(await fs.readFile(generationPath, "utf8")).toBe("foreign-generation"); + }); + it.each([ ["nonce", false], ["generation", false], @@ -364,7 +458,7 @@ describe("inactive real cancellation storage", () => { "post-settlement cleanup cannot overwrite a newer %s (retained=%s)", async (successorKind, retainUntilReplacement) => { const firstCleanup = Promise.withResolvers(); - const settled = Promise.withResolvers(); + const settled = Promise.withResolvers(); const neutralize = h.historyService.neutralizeCompactionRecoveryUnderHistoryLock.bind( h.historyService ); @@ -397,12 +491,12 @@ describe("inactive real cancellation storage", () => { }); expect((await foreign.writePartial(workspaceId, partial)).success).toBe(true); const before = await foreign.readPartial(workspaceId); - settled.resolve(); + settled.resolve(true); expect(await stopping).toBe("superseded"); expect(await foreign.readPartial(workspaceId)).toEqual(before); expect(await storage.read()).toEqual(successor); } finally { - settled.resolve(); + settled.resolve(true); await stopping; } } @@ -414,7 +508,7 @@ describe("inactive real cancellation storage", () => { if (phase === "first existing") await state.cancel(); const predecessor = await storage.read(); const initial = Promise.withResolvers(); - const settled = Promise.withResolvers(); + const settled = Promise.withResolvers(); const neutralize = h.historyService.neutralizeCompactionRecoveryUnderHistoryLock.bind( h.historyService ); @@ -450,7 +544,7 @@ describe("inactive real cancellation storage", () => { ) ).success ).toBe(true); - settled.resolve(); + settled.resolve(true); await failed; expect(state.needsPersistence).toBe(true); expect(state.blocksRecovery).toBe(true); @@ -464,7 +558,7 @@ describe("inactive real cancellation storage", () => { type: "compaction-summary", }); } finally { - settled.resolve(); + settled.resolve(true); await failed; } } diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts index 21ce2e37d0..81c2abadfa 100644 --- a/src/node/services/compactionCancellation.ts +++ b/src/node/services/compactionCancellation.ts @@ -36,7 +36,9 @@ export interface CompactionCancellationSummary { } export interface CompactionCancellationRecord { - version: 1; + version: 1 | 2; + /** V2 alone proves a successful physical Stop and final cleanup at this generation. */ + settledGeneration?: string; nonce: string; retainUntilReplacement?: boolean; scope: { kind: "unresolved" } | ({ kind: "summary" } & CompactionCancellationSummary); @@ -62,7 +64,7 @@ export type CompactionCancellationMutation = /** Explicit full deletion owns row removal; never serialized or used by ordinary Stop. */ fullHistoryDeletion?: true; /** In-memory completion of the captured engine and terminal policy, never serialized. */ - settled?: Promise; + settled?: Promise; onCaptured?: (capture: CompactionReplacementCapture) => void; } | { kind: "narrow"; record: CompactionCancellationRecord } @@ -116,7 +118,7 @@ export interface CompactionCancellationStorage { ): Promise; } -const CancellationRecordSchema = z.strictObject({ +const LegacyCancellationRecordSchema = z.strictObject({ version: z.literal(1), nonce: z.string().min(1), retainUntilReplacement: z.boolean().optional(), @@ -131,6 +133,17 @@ const CancellationRecordSchema = z.strictObject({ ]), }); +const SettledCancellationRecordSchema = LegacyCancellationRecordSchema.extend({ + version: z.literal(2), + retainUntilReplacement: z.literal(false).optional(), + scope: z.strictObject({ kind: z.literal("unresolved") }), + settledGeneration: z.string().min(1), +}); +const CancellationRecordSchema = z.union([ + LegacyCancellationRecordSchema, + SettledCancellationRecordSchema, +]); + function assertCancellationSize(contents: string): void { if (Buffer.byteLength(contents, "utf8") > SESSION_HISTORY_MAX_LINE_BYTES) throw new CompactionCancellationReadRefusedError("Cancellation record exceeds supported size"); @@ -191,9 +204,16 @@ export class FileCompactionCancellationStorage implements CompactionCancellation isPlainObject(parsed) && typeof parsed.version === "number" && Number.isInteger(parsed.version) && - parsed.version > 1 + parsed.version > 2 ) throw new CompactionCancellationReadRefusedError("Unsupported cancellation record version"); + if ( + isPlainObject(parsed) && + parsed.version === 2 && + (!SettledCancellationRecordSchema.safeParse(parsed).success || + hasAmbiguousResetKeys(contents)) + ) + throw new CompactionCancellationReadRefusedError("Unsupported settled cancellation record"); if (hasAmbiguousResetKeys(contents)) throw new Error("Duplicate cancellation fields"); return CancellationRecordSchema.parse(parsed); } catch (error) { @@ -334,7 +354,7 @@ export class FileCompactionCancellationStorage implements CompactionCancellation if (mutation.kind !== "publish" || !mutation.settled || outcome !== "applied") return outcome; // Settlement may write a final partial/legacy summary after the first cleanup. // Never hold history locks while joining those writers, or let an old Stop clear a successor. - await mutation.settled; + const physicallyStopped = await mutation.settled; return this.history.withCompactionStorageLock(this.workspaceId, async (_dir, checkLock) => { if (!isCurrent()) return "superseded"; const current = await this.read(); @@ -342,11 +362,36 @@ export class FileCompactionCancellationStorage implements CompactionCancellation .getContinuousCompactionJournal(this.workspaceId) .captureGenerationUnderHistoryLock(); const frontier = mutation.publication.predecessor; - if (current?.nonce !== frontier?.nonce || generation !== frontier?.generation) + if (!current || current.nonce !== frontier?.nonce || generation !== frontier?.generation) return "superseded"; - return (await this.history.neutralizeCompactionRecoveryUnderHistoryLock( - this.workspaceId, + if ( + !(await this.history.neutralizeCompactionRecoveryUnderHistoryLock( + this.workspaceId, + isCurrent, + checkLock + )) + ) + return "superseded"; + if ( + !physicallyStopped || + current.version !== 1 || + current.scope.kind !== "unresolved" || + current.retainUntilReplacement || + generation === undefined + ) + return "applied"; + // The existing second cleanup joins the old producer; only its exact durable frontier + // may qualify a later automatic replacement. Older readers preserve V2 as unsupported. + const { contents, record } = serializeCancellation({ + ...current, + version: 2, + settledGeneration: generation, + }); + return (await publishCompactionFile( + this.path, + contents, isCurrent, + () => onCommitted(record), checkLock )) ? "applied" @@ -506,7 +551,7 @@ export class CompactionCancellation { cancel(options?: { retainUntilReplacement?: boolean; - settled?: Promise; + settled?: Promise; onCaptured?: (capture: CompactionReplacementCapture) => void; fullHistoryDeletion?: true; }): Promise { @@ -626,6 +671,7 @@ export class CompactionCancellation { this.pending !== pending || this.replacementNonce === nonce || this.current?.nonce !== nonce || + this.current.version === 2 || this.current.scope.kind !== "unresolved" || this.current.retainUntilReplacement ) diff --git a/src/node/services/workspaceService.test.ts b/src/node/services/workspaceService.test.ts index c5c9938f56..f73b7996a9 100644 --- a/src/node/services/workspaceService.test.ts +++ b/src/node/services/workspaceService.test.ts @@ -764,29 +764,7 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { } }); - async function expectMonitorDeferredUntilManual( - h: Awaited>, - requestsBeforeReplacement: number - ): Promise { - // V1 cannot prove that an unresolved Stop's producers settled. Preserve attention until - // manual replacement; the V2 layer separately permits fresh automatic replacement. - expect(h.requests).toHaveLength(requestsBeforeReplacement); - expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(2); - expect(await h.session.isAutomaticSendBlocked()).toBe(true); - expect(h.session.isBusy()).toBe(false); - await h.reconciler.reconcile(h.workspaceId); - expect(h.requests).toHaveLength(requestsBeforeReplacement); - expect(h.internal.pendingBashMonitorWakeIdleWaitsByOwner.has(h.workspaceId)).toBe(false); - expect( - await h.session.sendMessage("manual replacement", { model: h.model, agentId: "exec" }) - ).toEqual(Ok(undefined)); - await h.complete(); - await h.reconciler.reconcile(h.workspaceId); - expect(h.requests).toHaveLength(requestsBeforeReplacement + 2); - expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(0); - } - - test("hard Stop retires old attention and defers fresh attention until manual replacement", async () => { + test("hard Stop retires owed attention without disarming future idle wakes", async () => { const h = await createActiveWakeHarness(); try { await h.session.sendMessage("original", { model: h.model, agentId: "exec" }); @@ -806,43 +784,50 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { expect(h.requests).toHaveLength(1); expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(0); await h.addAttention(20); - await expectMonitorDeferredUntilManual(h, 1); + expect(h.requests).toHaveLength(2); } finally { await h.finish(); } }); - test("retained Stop leaves fresh monitor attention owed without spinning and manual replacement re-arms it", async () => { - const h = await createActiveWakeHarness(); - const internal = h.internal as typeof h.internal & { - scheduleBashMonitorWakeReconcileAfterIdle(ownerWorkspaceId: string): void; - }; - // Suppress a broken immediate retry so the refusal is a bounded assertion, not a timeout. - const idleRetry = spyOn( - internal, - "scheduleBashMonitorWakeReconcileAfterIdle" - ).mockImplementation(() => undefined); - try { - await h.session.cancelCompaction(true); - const storage = h.historyService.getCompactionCancellationStorage(h.workspaceId); - const stop = await storage.read(); - await h.addAttention(20); - expect(idleRetry).not.toHaveBeenCalled(); - expect(h.requests).toHaveLength(0); - expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(2); - expect((await storage.read())?.nonce).toBe(stop?.nonce); - const history = await h.historyService.getHistoryFromLatestBoundary(h.workspaceId); - expect(history).toEqual(Ok([])); - idleRetry.mockRestore(); - await h.session.sendMessage("manual replacement", { model: h.model, agentId: "exec" }); - await h.complete(); - await h.reconciler.reconcile(h.workspaceId); - expect(h.requests).toHaveLength(2); - expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(0); - } finally { - await h.finish(); + test.each(["retained", "generation mismatch"] as const)( + "%s Stop leaves fresh monitor attention owed without spinning and manual replacement re-arms it", + async (kind) => { + const h = await createActiveWakeHarness(); + const internal = h.internal as typeof h.internal & { + scheduleBashMonitorWakeReconcileAfterIdle(ownerWorkspaceId: string): void; + }; + // Suppress a broken immediate retry so the refusal is a bounded assertion, not a timeout. + const idleRetry = spyOn( + internal, + "scheduleBashMonitorWakeReconcileAfterIdle" + ).mockImplementation(() => undefined); + try { + if (kind === "retained") await h.session.cancelCompaction(true); + else { + expect(await h.session.interruptStream()).toEqual(Ok(undefined)); + await h.historyService.getContinuousCompactionJournal(h.workspaceId).advanceGeneration(); + } + const storage = h.historyService.getCompactionCancellationStorage(h.workspaceId); + const stop = await storage.read(); + await h.addAttention(20); + expect(idleRetry).not.toHaveBeenCalled(); + expect(h.requests).toHaveLength(0); + expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(2); + expect((await storage.read())?.nonce).toBe(stop?.nonce); + const history = await h.historyService.getHistoryFromLatestBoundary(h.workspaceId); + expect(history).toEqual(Ok([])); + idleRetry.mockRestore(); + await h.session.sendMessage("manual replacement", { model: h.model, agentId: "exec" }); + await h.complete(); + await h.reconciler.reconcile(h.workspaceId); + expect(h.requests).toHaveLength(2); + expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(0); + } finally { + await h.finish(); + } } - }); + ); test("a cancellation write failure still completes successful hard-Stop cleanup", async () => { const h = await createActiveWakeHarness(); @@ -970,13 +955,13 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { expect(h.requests).toHaveLength(1); expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(0); await h.addAttention(20); - await expectMonitorDeferredUntilManual(h, 1); + expect(h.requests).toHaveLength(2); } finally { await h.finish(); } }); - test("a failed hard Stop keeps owed attention until manual replacement", async () => { + test("a failed hard Stop keeps owed attention for the idle wake", async () => { const h = await createActiveWakeHarness(); try { await h.session.sendMessage("original", { model: h.model, agentId: "exec" }); @@ -990,13 +975,13 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { await h.complete(); await h.internal.pendingBashMonitorWakeIdleWaitsByOwner.get(h.workspaceId); await h.reconciler.reconcile(h.workspaceId); - await expectMonitorDeferredUntilManual(h, 1); + expect(h.requests).toHaveLength(2); } finally { await h.finish(); } }); - test("an interrupt without retireBashMonitorAttention preserves owed attention until manual replacement", async () => { + test("an interrupt without retireBashMonitorAttention keeps owed attention for the idle wake", async () => { const h = await createActiveWakeHarness(); try { await h.session.sendMessage("original", { model: h.model, agentId: "exec" }); @@ -1010,7 +995,7 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { expect((await h.service.interruptStream(h.workspaceId)).success).toBe(true); await h.internal.pendingBashMonitorWakeIdleWaitsByOwner.get(h.workspaceId); await h.reconciler.reconcile(h.workspaceId); - await expectMonitorDeferredUntilManual(h, 1); + expect(h.requests).toHaveLength(2); } finally { await h.finish(); } @@ -1058,7 +1043,7 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { expect(h.session.isBusy()).toBe(false); expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(0); await h.addAttention(20); - await expectMonitorDeferredUntilManual(h, 0); + expect(h.requests).toHaveLength(1); } finally { await h.finish(); } @@ -1238,8 +1223,8 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { release.resolve(); expect((await stop).success).toBe(true); await Promise.all([attention, later]); - // Only the frontier the Stop saw was retired; the newer output remains owed. - await expectMonitorDeferredUntilManual(h, 0); + // Only the frontier the Stop saw was retired; the newer output woke the idle agent. + expect(h.requests).toHaveLength(1); } finally { release.resolve(); await h.finish(); @@ -2042,12 +2027,14 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { const h = await createActiveWakeHarness(); const entered = createDeferred(); const release = createDeferred(); - const append = h.historyService.appendToHistory.bind(h.historyService); - spyOn(h.historyService, "appendToHistory").mockImplementationOnce(async (...args) => { - entered.resolve(); - await release.promise; - return append(...args); - }); + const append = h.historyService.acceptCompactionReplacement.bind(h.historyService); + spyOn(h.historyService, "acceptCompactionReplacement").mockImplementationOnce( + async (...args) => { + entered.resolve(); + await release.promise; + return append(...args); + } + ); const controller = new AbortController(); const accepted = mock(() => Promise.resolve()); const deferred = mock(() => Promise.resolve()); From 6b86c974f29937c362cfc9d0fad0c4c970be5bf8 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 11 Sep 2026 15:21:33 +0200 Subject: [PATCH 2/8] =?UTF-8?q?=F0=9F=A4=96=20fix:=20qualify=20Stop=20repl?= =?UTF-8?q?acement=20after=20complete=20owned=20cleanup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wait for the captured producer and exact outer cleanup before V2 replacement authority. Preserve eventual wake after failed physical Stop or retryable monitor retirement, keep unresolved foreign V1 blocked, and retain legacy scoped-V1 ordinary admission. Stamp heartbeat boundary generation for safe fresh follow-up recovery after restart without an optional sidecar; actual input still owns replacement CAS. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ --- src/common/orpc/schemas/message.ts | 1 + src/common/types/message.ts | 2 + ...gentSession.compactionCancellation.test.ts | 158 ++++++++++++++++++ src/node/services/agentSession.ts | 135 ++++++++++++++- .../services/bashMonitorWakeReconciler.ts | 17 +- src/node/services/compactionCancellation.ts | 4 + src/node/services/compactionPendingState.ts | 6 +- src/node/services/turnCoordinator.ts | 4 +- src/node/services/workspaceService.test.ts | 137 ++++++++++++++- src/node/services/workspaceService.ts | 39 ++++- 10 files changed, 488 insertions(+), 15 deletions(-) diff --git a/src/common/orpc/schemas/message.ts b/src/common/orpc/schemas/message.ts index fa8f4ec8b5..eb2fd93519 100644 --- a/src/common/orpc/schemas/message.ts +++ b/src/common/orpc/schemas/message.ts @@ -198,6 +198,7 @@ export const MuxMessageSchema = z.object({ // Durable boundary marker for compaction summaries. compactionBoundary: z.boolean().optional(), compactionPublicationId: z.string().min(1).optional().catch(undefined), + compactionPublicationGeneration: z.string().min(1).nullable().optional().catch(undefined), contextBoundaryKind: z.literal(CONTEXT_BOUNDARY_KINDS.RESET).optional(), toolPolicy: z.any().optional(), disableWorkspaceAgents: z.boolean().optional(), diff --git a/src/common/types/message.ts b/src/common/types/message.ts index 3e729540df..f8561dd2d1 100644 --- a/src/common/types/message.ts +++ b/src/common/types/message.ts @@ -1050,6 +1050,8 @@ export interface MuxMetadata { compactionBoundary?: boolean; /** Exact composed publication occurrence, shared with the pending file writeId. */ compactionPublicationId?: string; + /** Captured generation of this publication; legacy rewrites must not inherit it. */ + compactionPublicationGeneration?: string | null; /** Durable provider-context boundary kind. Existing compaction rows are also boundaries via compactionBoundary. */ contextBoundaryKind?: PersistedContextBoundaryKind; toolPolicy?: ToolPolicy; // Tool policy active when this message was sent (user messages only) diff --git a/src/node/services/agentSession.compactionCancellation.test.ts b/src/node/services/agentSession.compactionCancellation.test.ts index 7e4b2be683..6831d17241 100644 --- a/src/node/services/agentSession.compactionCancellation.test.ts +++ b/src/node/services/agentSession.compactionCancellation.test.ts @@ -212,6 +212,164 @@ describe("compaction cancellation runtime", () => { } ); + test.each( + [false, true].flatMap((restart) => + (["present", "missing", "write failure"] as const).map((sidecar) => ({ restart, sidecar })) + ) + )( + "fresh reset after settled Stop preserves its follow-up ($restart, $sidecar)", + async ({ restart, sidecar }) => { + const h = await fixture(); + expect(await h.session.interruptStream()).toEqual(Ok(undefined)); + const pendingPath = path.join(h.config.sessionsDir, workspaceId, "post-compaction.json"); + if (sidecar === "write failure") { + await fileIO.mkdir(pendingPath); + await fileIO.writeFile(path.join(pendingPath, "unrelated"), "preserve"); + } + const reset = await h.session.appendHeartbeatContextResetBoundary({ + boundaryText: "fresh reset", + pendingFollowUp: { text: "fresh heartbeat", model: options.model, agentId: "exec" }, + }); + expect(reset.success).toBe(true); + assert(reset.success); + if (sidecar === "missing") await fileIO.rm(pendingPath, { force: true }); + let session = h.session; + if (restart) { + await session.dispose(); + const fresh = await createAgentSessionHarness({ + workspaceId, + config: h.config, + historyService: new HistoryService(h.config), + }); + fixtures.push(fresh); + session = fresh.session; + } + expect( + await session.dispatchPendingCompactionFollowUpIfNeeded(reset.data.summaryMessageId) + ).toBe(true); + expect( + (await h.rows()).some( + (row) => + row.role === "user" && + row.parts.some((part) => part.type === "text" && part.text === "fresh heartbeat") + ) + ).toBe(true); + } + ); + + test.each([ + "missing generation", + "old generation", + "local Stop", + "foreign Stop", + "late foreign Stop", + "publication failure", + ] as const)("fresh reset recovery refuses stale authority (%s)", async (change) => { + const h = await fixture(); + expect(await h.session.interruptStream()).toEqual(Ok(undefined)); + const reset = await h.session.appendHeartbeatContextResetBoundary({ + boundaryText: "fresh reset", + pendingFollowUp: { text: "fresh heartbeat", model: options.model, agentId: "exec" }, + }); + assert(reset.success); + const summary = (await h.rows())[0]; + const publicationId = summary.metadata?.compactionPublicationId; + assert(summary.metadata); + if (change === "missing generation" || change === "old generation") { + delete summary.metadata.compactionPublicationId; + summary.metadata.compactionPublicationGeneration = + change === "old generation" ? "old" : undefined; + expect(await h.historyService.updateHistory(workspaceId, summary)).toEqual(Ok(undefined)); + expect((await h.rows())[0].metadata?.compactionPublicationId).toBe(publicationId); + await h.session.dispose(); + } else if (change === "local Stop") await h.session.cancelCompaction(); + else if (change === "foreign Stop") await new CompactionCancellation(h.storage).cancel(); + let session = h.session; + if (change === "missing generation" || change === "old generation") { + const fresh = await createAgentSessionHarness({ + workspaceId, + config: h.config, + historyService: new HistoryService(h.config), + }); + fixtures.push(fresh); + session = fresh.session; + } + if (change === "late foreign Stop") { + const accept = h.historyService.acceptCompactionReplacement.bind(h.historyService); + spyOn(h.historyService, "acceptCompactionReplacement").mockImplementationOnce( + async (...args) => { + await new CompactionCancellation(h.storage).cancel(); + return accept(...args); + } + ); + } else if (change === "publication failure") { + spyOn(h.historyService, "acceptCompactionReplacement").mockResolvedValueOnce( + Err("disk unavailable") + ); + } + const dispatch = session.dispatchPendingCompactionFollowUpIfNeeded(reset.data.summaryMessageId); + if (change === "late foreign Stop" || change === "publication failure") + await nodeAssert.rejects(dispatch); + else expect(await dispatch).toBe(false); + expect((await h.rows()).some((row) => row.role === "user")).toBe(false); + expect(h.stream).not.toHaveBeenCalled(); + }); + + test("abandoned outer Stop completion remains V1 and disposal does not join itself", async () => { + const h = await fixture(); + let finalize: ((success: boolean | Promise) => Promise) | undefined; + expect( + await h.session.interruptStream({ + deferCompactionSettlement: (complete) => { + finalize = complete; + }, + }) + ).toEqual(Ok(undefined)); + expect(await h.storage.read()).toMatchObject({ version: 1 }); + await h.session.dispose(); + expect(await finalize?.(true)).toEqual(Ok(undefined)); + expect(await h.storage.read()).toMatchObject({ version: 1 }); + }); + + test.each(["complete", "supersede", "dispose"] as const)( + "failed Stop supervises captured startup until %s", + async (finish) => { + const h = await fixture(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const qualified = Promise.withResolvers(); + const notify = mock(() => qualified.resolve()); + h.stream.mockImplementationOnce(async () => { + entered.resolve(); + await release.promise; + return Err({ type: "unknown", raw: "startup failed" }); + }); + const sending = h.session.sendMessage("held startup", options); + await entered.promise; + spyOn(h.aiService, "stopStream").mockResolvedValueOnce(Err("stop failed")); + expect(await h.session.interruptStream({ onCompactionSettled: notify })).toEqual( + Err("stop failed") + ); + expect(await h.storage.read()).toMatchObject({ version: 1 }); + expect(await h.session.isAutomaticSendBlocked()).toBe(true); + let disposing: Promise | undefined; + if (finish === "supersede") + expect(await h.session.cancelCompaction(true)).toEqual(Ok(undefined)); + if (finish === "dispose") disposing = h.session.dispose(); + release.resolve(); + await sending; + if (finish === "complete") { + await qualified.promise; + expect(await h.storage.read()).toMatchObject({ version: 2 }); + expect(await h.session.isAutomaticSendBlocked()).toBe(false); + } else { + await disposing; + expect(notify).not.toHaveBeenCalled(); + expect(await h.storage.read()).toMatchObject({ version: 1 }); + } + } + ); + test("settled Stop qualifies only durable fresh automatic input across restart", async () => { const h = await fixture(); expect(await h.session.interruptStream()).toEqual(Ok(undefined)); diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 95fe6a1a0d..2f72f165f3 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -789,6 +789,8 @@ interface CachedMemoryContext { interface SendMessageInternalOptions { readCompactionAdmission?: () => Promise>; + /** Recovery retains its original durable Stop frontier through final trigger publication. */ + recoveryReplacement?: CompactionReplacementCapture; acceptanceOrigin?: TurnAcceptanceOrigin; preparation?: PreparationAttempt; /** A dequeued send keeps its admission owner through acceptance and startup failure. */ @@ -3664,7 +3666,9 @@ export class AgentSession { // belongs to a later admission and cannot grant this attempt replacement authority. const automaticCapture = manualReplacement ? undefined - : await this.historyService.captureCompactionReplacement(this.workspaceId); + : internal?.recoveryReplacement + ? Ok(internal.recoveryReplacement) + : await this.historyService.captureCompactionReplacement(this.workspaceId); if (manualReplacement) { await this.readCompactionCancellation("manual"); const stopAdmission = attempt.queuedStopAdmission; @@ -7312,14 +7316,18 @@ export class AgentSession { }; } + private pendingStopCompletion?: AbortController; + async cancelCompaction( retainUntilReplacement = false, settled?: Promise, options?: { fullHistoryDeletion?: true; onCaptured?: (capture: CompactionReplacementCapture) => void; + onInitialSettlement?: () => void; } ): Promise> { + this.pendingStopCompletion?.abort(); this.compactionStopGeneration++; this.pendingResumeIntent?.abort(); this.coordinator.abandonCompaction(); @@ -7331,6 +7339,7 @@ export class AgentSession { settled, fullHistoryDeletion: options?.fullHistoryDeletion, onCaptured: options?.onCaptured, + onInitialSettlement: options?.onInitialSettlement, }); return Ok(undefined); } catch (error) { @@ -7399,6 +7408,7 @@ export class AgentSession { if (this.compactionCancellation.blocksRecovery) return true; const summary = history.data.map(pendingCompactionSummary).findLast((entry) => entry != null); if (summary) await this.compactionCancellation.narrow(record.nonce, summary); + else return true; } return this.compactionCancellation.blocksRecovery; } @@ -7429,16 +7439,103 @@ export class AgentSession { abandonPartial?: boolean; preserveCompactionIntent?: boolean; onCompactionCanceled?: (capture: CompactionReplacementCapture) => void; + onCompactionSettled?: () => void; + deferCompactionSettlement?: ( + finalize: (cleanupSucceeded: boolean | Promise) => Promise> + ) => void; }): Promise { this.assertNotDisposed("interruptStream"); const settled = Promise.withResolvers(); + const initiallySettled = Promise.withResolvers>(); let physicallyStopped = false; const cancellation = options?.soft || options?.preserveCompactionIntent ? undefined : this.cancelCompaction(false, settled.promise, { onCaptured: options?.onCompactionCanceled, + onInitialSettlement: () => initiallySettled.resolve(Ok(undefined)), }); + const completionController = new AbortController(); + if (cancellation) { + this.pendingStopCompletion = completionController; + const abandon = () => { + completionController.abort(); + settled.resolve(false); + }; + completionController.signal.addEventListener("abort", () => settled.resolve(false), { + once: true, + }); + this.closingSignal.addEventListener("abort", abandon, { once: true }); + if (this.coordinator.closing) abandon(); + const releaseCompletion = () => { + this.closingSignal.removeEventListener("abort", abandon); + if (this.pendingStopCompletion === completionController) + this.pendingStopCompletion = undefined; + }; + cancellation.then(releaseCompletion, releaseCompletion); + } + const deferred = cancellation && options?.deferCompactionSettlement; + const generation = this.compactionStopGeneration; + // Failed startup must retain the exact registered producer, even before stream-start. + const producerCompletion = this.coordinator.captureInterruptSettlement(options?.soft, true); + let finalized: Promise> | undefined; + const finalize = (cleanupSucceeded: boolean | Promise): Promise> => { + if (finalized) return finalized; + if (!cancellation) return Promise.resolve(Ok(undefined)); + if ((physicallyStopped && cleanupSucceeded === true) || cleanupSucceeded === false) { + settled.resolve(physicallyStopped && cleanupSucceeded === true); + return (finalized = cancellation); + } + // A failed interrupt still returns its error immediately. Its captured producer can + // finish naturally; that exact completion restores automatic admission without losing + // owed monitor attention. The guardian joins this lease, and close/supersession abort it. + const controller = completionController; + const execution = this.coordinator.enterExecution(); + const abandoned = Promise.withResolvers(); + const abandon = () => abandoned.resolve(false); + controller.signal.addEventListener("abort", abandon, { once: true }); + this.closingSignal.addEventListener("abort", abandon, { once: true }); + if ( + controller.signal.aborted || + this.coordinator.closing || + generation !== this.compactionStopGeneration + ) + abandon(); + Promise.race([ + Promise.all([physicallyStopped ? undefined : producerCompletion, cleanupSucceeded]).then( + ([, complete]) => complete + ), + abandoned.promise, + ]) + .then(async (completed) => { + const current = + completed && !this.coordinator.closing && generation === this.compactionStopGeneration; + settled.resolve(current); + const result = await cancellation; + if ( + current && + result.success && + !this.coordinator.closing && + generation === this.compactionStopGeneration + ) + options?.onCompactionSettled?.(); + }) + .catch((error: unknown) => { + settled.resolve(false); + log.warn("Deferred Stop completion failed", { error }); + }) + .finally(() => { + controller.signal.removeEventListener("abort", abandon); + this.closingSignal.removeEventListener("abort", abandon); + if (this.pendingStopCompletion === controller) this.pendingStopCompletion = undefined; + execution[Symbol.dispose](); + }); + return (finalized = Promise.resolve(Ok(undefined))); + }; + deferred?.(finalize); + cancellation?.then(initiallySettled.resolve, (error: unknown) => + initiallySettled.resolve(Err(getErrorMessage(error))) + ); // Send-now callers may replace the turn immediately after this returns. Capture // its settlement before any await so the old abort reaches accounting and the // renderer before replacement PREPARING invalidates its operation identity. @@ -7469,8 +7566,16 @@ export class AgentSession { physicallyStopped = result.success; return result; }) - .finally(() => settled.resolve(physicallyStopped)); - const canceled = await cancellation; + .catch((error: unknown) => { + settled.resolve(false); + throw error; + }); + let canceled = cancellation ? await initiallySettled.promise : undefined; + if (!deferred) { + // Cancellation I/O debt owns its retry; it does not undo physical completion. + const finalResult = await finalize(true); + if (!finalResult.success) canceled = finalResult; + } if (!stopResult.success) { return Err(stopResult.error); } @@ -10467,6 +10572,9 @@ export class AgentSession { const resumeCanceled = () => stopGeneration !== this.compactionStopGeneration || cancelResume?.() === true; + const recoveryCapture = await this.historyService.captureCompactionReplacement( + this.workspaceId + ); const canceled = await this.readCompactionCancellation(); // Stop can be waiting for this policy to settle before its final cleanup. // Do not join that same mutation from automatic continuation dispatch. @@ -10569,7 +10677,24 @@ export class AgentSession { } const summary = pendingCompactionSummary(lastMessage); - if (canceled && summary && matchesCompactionCancellation(canceled, summary)) { + // V2 cleanup removed every older handoff before settlement. Only an explicitly stamped + // heartbeat publication in that exact generation can be newer, including without a sidecar. + // Marker-preserving legacy rewrites omit this generation and gain no recovery authority. + const freshHeartbeat = + canceled?.version === 2 && + !canceled.retainUntilReplacement && + lastMessage.metadata?.compacted === "heartbeat" && + lastMessage.metadata.compactionPublicationId !== undefined && + lastMessage.metadata.compactionPublicationGeneration === canceled.settledGeneration && + recoveryCapture.success && + recoveryCapture.data.nonce === canceled.nonce && + recoveryCapture.data.generation === canceled.settledGeneration; + if ( + canceled && + summary && + !freshHeartbeat && + matchesCompactionCancellation(canceled, summary) + ) { // The history read may have admitted a newer Stop that is waiting for this policy. if (this.compactionCancellation.blocksRecovery) return false; await this.compactionCancellation.narrow(canceled.nonce, summary); @@ -10810,6 +10935,8 @@ export class AgentSession { // re-enable auto-retry after a user explicitly opted out. const sendResult = await this.sendMessage(finalText, options, { startStreamInBackground, + recoveryReplacement: + freshHeartbeat && recoveryCapture.success ? recoveryCapture.data : undefined, acceptanceOrigin: "automatic", synthetic: true, agentInitiated: followUp.agentInitiated, diff --git a/src/node/services/bashMonitorWakeReconciler.ts b/src/node/services/bashMonitorWakeReconciler.ts index beca2c6e34..8b91a621c2 100644 --- a/src/node/services/bashMonitorWakeReconciler.ts +++ b/src/node/services/bashMonitorWakeReconciler.ts @@ -172,6 +172,7 @@ interface ReconcileState { owedAcceptance?: readonly DerivedSignal[]; /** Frontier a committed stop still has to retire; applied before any dispatch. */ owedRetirement?: readonly BashMonitorProcessSnapshot[]; + retirementCompletions?: Set<() => void>; /** Outstanding wake keys already looked up in the transcript (see deliveredSignals). */ transcriptChecked?: ReadonlySet; } @@ -717,7 +718,11 @@ export class BashMonitorWakeReconciler { * given, the durable consumption waits for it under the lock and is skipped when it resolves * false, leaving the withdrawn signals owed to the next reconcile. */ - async consumeCurrent(ownerWorkspaceId: string, commit?: () => Promise): Promise { + async consumeCurrent( + ownerWorkspaceId: string, + commit?: () => Promise, + onSettled?: () => void + ): Promise { // Withdraw before taking the lock: an acceptance in progress holds it across watermark, // registry, and process-acknowledgement I/O, and a hard Stop must cancel the admission // without waiting behind that. The lock slot is reserved synchronously too, ahead of any @@ -729,13 +734,17 @@ export class BashMonitorWakeReconciler { const frontier = this.args.processManager.pullMonitorWakeSignals(ownerWorkspaceId); const committed = await this.locks.withLock(ownerWorkspaceId, async () => { this.abortDispatch(ownerWorkspaceId); - if (commit != null && !(await commit())) return false; + if (commit != null && !(await commit())) { + onSettled?.(); + return false; + } const state = this.state(ownerWorkspaceId); const owed = new Map( (state.owedRetirement ?? []).map((s) => [signalKey(s.processId, s.createdAt), s] as const) ); for (const s of frontier) owed.set(signalKey(s.processId, s.createdAt), s); state.owedRetirement = [...owed.values()]; + if (onSettled) (state.retirementCompletions ??= new Set()).add(onSettled); await this.retireOwed(ownerWorkspaceId, state); return true; }); @@ -749,6 +758,10 @@ export class BashMonitorWakeReconciler { await this.advanceWatermarks(ownerWorkspaceId, collected.watermarks, consumed); await this.cleanup(consumed); state.owedRetirement = undefined; + // Failed I/O retains the original obligation and its receipts for the existing retry. + const completions = state.retirementCompletions; + state.retirementCompletions = undefined; + for (const complete of completions ?? []) complete(); } /** diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts index 81c2abadfa..afe1479218 100644 --- a/src/node/services/compactionCancellation.ts +++ b/src/node/services/compactionCancellation.ts @@ -66,6 +66,7 @@ export type CompactionCancellationMutation = /** In-memory completion of the captured engine and terminal policy, never serialized. */ settled?: Promise; onCaptured?: (capture: CompactionReplacementCapture) => void; + onInitialSettlement?: () => void; } | { kind: "narrow"; record: CompactionCancellationRecord } | { @@ -351,6 +352,7 @@ export class FileCompactionCancellationStorage implements CompactionCancellation return "applied"; }); const outcome = await write; + if (mutation.kind === "publish") mutation.onInitialSettlement?.(); if (mutation.kind !== "publish" || !mutation.settled || outcome !== "applied") return outcome; // Settlement may write a final partial/legacy summary after the first cleanup. // Never hold history locks while joining those writers, or let an old Stop clear a successor. @@ -553,6 +555,7 @@ export class CompactionCancellation { retainUntilReplacement?: boolean; settled?: Promise; onCaptured?: (capture: CompactionReplacementCapture) => void; + onInitialSettlement?: () => void; fullHistoryDeletion?: true; }): Promise { this.current = { @@ -571,6 +574,7 @@ export class CompactionCancellation { publication: { attempts: 0 }, fullHistoryDeletion: options?.fullHistoryDeletion, onCaptured: options?.onCaptured, + onInitialSettlement: options?.onInitialSettlement, ...(options?.settled ? { settled: options.settled } : {}), }); } diff --git a/src/node/services/compactionPendingState.ts b/src/node/services/compactionPendingState.ts index 5d951e6da9..52ca85b8ea 100644 --- a/src/node/services/compactionPendingState.ts +++ b/src/node/services/compactionPendingState.ts @@ -605,8 +605,12 @@ export class CompactionPendingState { // Caller metadata changes only with the synchronous history receipt, never while staging. const summaryMessage = structuredClone(input.summaryMessage); const writeId = randomUUID(); - summaryMessage.metadata = { ...summaryMessage.metadata, compactionPublicationId: writeId }; const publication = structuredClone(input.publication); + summaryMessage.metadata = { + ...summaryMessage.metadata, + compactionPublicationId: writeId, + compactionPublicationGeneration: publication.generation ?? null, + }; const preparation = { attachments: structuredClone(input.attachments), boundaryMessageId: summaryMessage.id, diff --git a/src/node/services/turnCoordinator.ts b/src/node/services/turnCoordinator.ts index 308ee13fee..60291957f0 100644 --- a/src/node/services/turnCoordinator.ts +++ b/src/node/services/turnCoordinator.ts @@ -1159,9 +1159,9 @@ export class TurnCoordinator { return notify && !this.disposed; } - captureInterruptSettlement(soft?: boolean): Promise | undefined { + captureInterruptSettlement(soft?: boolean, includeStartup = false): Promise | undefined { const operation = this.state.turn.operation; - return !soft && operation?.stage === "started" + return !soft && operation && (includeStartup || operation.stage === "started") ? this.settlements.get(operation.id)?.promise : undefined; } diff --git a/src/node/services/workspaceService.test.ts b/src/node/services/workspaceService.test.ts index f73b7996a9..dba1fb0ff2 100644 --- a/src/node/services/workspaceService.test.ts +++ b/src/node/services/workspaceService.test.ts @@ -1,5 +1,8 @@ import type { TurnCompletion } from "./streamManager"; -import { FileCompactionCancellationStorage } from "./compactionCancellation"; +import { + FileCompactionCancellationStorage, + type CompactionCancellation, +} from "./compactionCancellation"; import { CompactionPendingState } from "./compactionPendingState"; import * as historyScanner from "./historyScanner"; import type { TurnCoordinator } from "./turnCoordinator"; @@ -764,6 +767,134 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { } }); + test("outer Stop preserves physical completion through exact cleanup retry", async () => { + const h = await createActiveWakeHarness(); + try { + spyOn(h.historyService, "neutralizeCompactionRecoveryUnderHistoryLock").mockRejectedValueOnce( + new Error("cleanup unavailable") + ); + expect(await h.service.interruptStream(h.workspaceId)).toEqual(Err(STOP_UNRECORDED_MESSAGE)); + const storage = h.historyService.getCompactionCancellationStorage(h.workspaceId); + const failed = await storage.read(); + expect(failed).toMatchObject({ version: 1 }); + const cancellation = ( + h.session as unknown as { compactionCancellation: CompactionCancellation } + ).compactionCancellation; + expect(await cancellation.retry()).toBe("applied"); + expect(await storage.read()).toMatchObject({ version: 2, nonce: failed?.nonce }); + expect(await h.session.isAutomaticSendBlocked()).toBe(false); + } finally { + await h.finish(); + } + }); + + test("failed descendant Stop cleanup stays V1 across restart", async () => { + const h = await createActiveWakeHarness(); + h.service.setAgentTaskIntegration( + makeAgentTaskIntegrationFake({ + terminateAllDescendantAgentTasks: () => Promise.reject(new Error("descendant unavailable")), + }) + ); + const foreign = await createAgentSessionHarness({ + workspaceId: h.workspaceId, + config: h.config, + historyService: new HistoryService(h.config), + }); + try { + // Preserve the existing API result; swallowed cleanup errors confer no settlement proof. + expect(await h.service.interruptStream(h.workspaceId)).toEqual(Ok(undefined)); + expect( + await h.historyService.getCompactionCancellationStorage(h.workspaceId).read() + ).toMatchObject({ version: 1 }); + expect(await foreign.session.isAutomaticSendBlocked()).toBe(true); + } finally { + await foreign.session.dispose(); + await foreign.cleanup(); + await h.finish(); + } + }); + + test.each( + (["retirement", "descendants"] as const).flatMap((phase) => + [false, true].map((superseded) => ({ phase, superseded })) + ) + )( + "hard Stop remains V1 until outer $phase finishes across instances (superseded=$superseded)", + async ({ phase, superseded }) => { + const h = await createActiveWakeHarness(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + if (phase === "retirement") { + const consume = h.reconciler.consumeCurrent.bind(h.reconciler); + spyOn(h.reconciler, "consumeCurrent").mockImplementationOnce(async (...args) => { + const result = await consume(...args); + entered.resolve(); + await release.promise; + return result; + }); + } else { + h.service.setAgentTaskIntegration( + makeAgentTaskIntegrationFake({ + terminateAllDescendantAgentTasks: async () => { + entered.resolve(); + await release.promise; + return []; + }, + }) + ); + } + const foreign = await createAgentSessionHarness({ + workspaceId: h.workspaceId, + config: h.config, + historyService: new HistoryService(h.config), + }); + let stopping: Promise | undefined; + try { + stopping = h.service.interruptStream(h.workspaceId, { retireBashMonitorAttention: true }); + await entered.promise; + expect( + await h.historyService.getCompactionCancellationStorage(h.workspaceId).read() + ).toMatchObject({ version: 1, scope: { kind: "unresolved" } }); + expect(await foreign.session.isAutomaticSendBlocked()).toBe(true); + expect( + ( + await foreign.session.sendMessage( + "too early", + { model: h.model, agentId: "exec" }, + { acceptanceOrigin: "automatic" } + ) + ).success + ).toBe(false); + if (superseded) expect(await foreign.session.cancelCompaction()).toEqual(Ok(undefined)); + const successor = superseded + ? await h.historyService.getCompactionCancellationStorage(h.workspaceId).read() + : null; + release.resolve(); + expect(await stopping).toEqual(Ok(undefined)); + const completed = await h.historyService + .getCompactionCancellationStorage(h.workspaceId) + .read(); + if (superseded) expect(completed).toEqual(successor); + else expect(completed).toMatchObject({ version: 2 }); + expect( + ( + await foreign.session.sendMessage( + "fresh after cleanup", + { model: h.model, agentId: "exec" }, + { acceptanceOrigin: "automatic" } + ) + ).success + ).toBe(!superseded); + } finally { + release.resolve(); + await stopping; + await foreign.session.dispose(); + await foreign.cleanup(); + await h.finish(); + } + } + ); + test("hard Stop retires owed attention without disarming future idle wakes", async () => { const h = await createActiveWakeHarness(); try { @@ -954,7 +1085,9 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { // The stop's idle reconcile retried the retirement instead of re-dispatching the output. expect(h.requests).toHaveLength(1); expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(0); + const woke = new Promise((resolve) => h.launched.once("start", resolve)); await h.addAttention(20); + await woke; expect(h.requests).toHaveLength(2); } finally { await h.finish(); @@ -972,7 +1105,9 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { .success ).toBe(false); expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(2); + const woke = new Promise((resolve) => h.launched.once("start", resolve)); await h.complete(); + await woke; await h.internal.pendingBashMonitorWakeIdleWaitsByOwner.get(h.workspaceId); await h.reconciler.reconcile(h.workspaceId); expect(h.requests).toHaveLength(2); diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index 7d668beae8..413a7dcaf1 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -12022,6 +12022,9 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { } ): Promise> { let releaseHardStopLatch: (() => void) | undefined; + let finalizeCompactionStop: + | ((cleanupSucceeded: boolean | Promise) => Promise>) + | undefined; try { this.agentTaskIntegration?.resetAutoResumeCount(workspaceId); if (!options?.soft) { @@ -12059,9 +12062,15 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { settleStop = resolve; }); let retirementRecorded = true; + const retirementSettled = Promise.withResolvers(); + if (!retiring) retirementSettled.resolve(); const retirement = retiring ? this.bashMonitorWakeReconciler - .consumeCurrent(workspaceId, () => stopSettled) + .consumeCurrent( + workspaceId, + () => stopSettled, + () => retirementSettled.resolve() + ) .catch((error: unknown) => { retirementRecorded = false; log.warn("Failed to retire bash monitor attention before Stop", { @@ -12087,6 +12096,10 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { try { const stopping = session.interruptStream({ ...options, + onCompactionSettled: () => this.scheduleBashMonitorWakeReconcile(workspaceId), + deferCompactionSettlement: (finalize) => { + finalizeCompactionStop = finalize; + }, onCompactionCanceled: (capture) => { stopCapture = capture; }, @@ -12108,10 +12121,17 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { // unrecorded after the session retried the write fails the Stop below, on this and every later // Stop, so the obligation is not lost with the joined send. await withdrawnWakeSend?.catch(() => undefined); - const stopRecorded = - !(retiring || disabling) || - ((await session.recordPendingAutoRetryState()) && retirementRecorded); + const retryStateRecorded = + !(retiring || disabling) || (await session.recordPendingAutoRetryState()); + const stopRecorded = retryStateRecorded && retirementRecorded; + const cleanupQualification = () => + retryStateRecorded + ? retirementRecorded + ? true + : retirementSettled.promise.then(() => true) + : false; if (!stopResult.success && !stopResult.streamStopped) { + await finalizeCompactionStop?.(cleanupQualification()); // Interrupt failed, so clear hard-interrupt suppression we set above. if (!options?.soft) { this.agentTaskIntegration?.resetAutoResumeCount(workspaceId); @@ -12127,6 +12147,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { await this.historyService.deletePartial(workspaceId); } + let descendantsSettled = true; // Rationale: user-initiated hard interrupts should stop the entire task tree so // descendant sub-agents cannot finish later and auto-resume this workspace. if (!options?.soft) { @@ -12140,6 +12161,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { }); } } catch (error: unknown) { + descendantsSettled = false; log.error("Failed to cascade-interrupt descendant tasks on interrupt", { workspaceId, error, @@ -12167,7 +12189,12 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { session.restoreQueueToInput(); } - if (!stopRecorded || !stopResult.success) { + const finalized = await finalizeCompactionStop?.( + (stopResult.success || stopResult.streamStopped) && descendantsSettled + ? cleanupQualification() + : false + ); + if (!stopRecorded || !stopResult.success || finalized?.success === false) { log.error("Stop left stopped work eligible to resume on restart", { workspaceId }); return Err(STOP_UNRECORDED_MESSAGE); } @@ -12181,6 +12208,8 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { log.error("Unexpected error in interruptStream handler:", error); return Err(`Failed to interrupt stream: ${errorMessage}`); } finally { + // Every early return/throw abandons V2 qualification while still joining the exact Stop. + await finalizeCompactionStop?.(false); releaseHardStopLatch?.(); } } From 6a090ca07eb5706a7dd37adc994337b3aebb8a74 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 11 Sep 2026 16:08:13 +0200 Subject: [PATCH 3/8] =?UTF-8?q?=F0=9F=A4=96=20fix:=20settle=20captured=20s?= =?UTF-8?q?tartup=20before=20restoring=20monitor=20wakes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Ib96dfe46e536ed66daacef4347f6aca478a9102b --- ...gentSession.compactionCancellation.test.ts | 172 +++++++++++++++--- src/node/services/agentSession.ts | 66 +++++-- src/node/services/compactionCancellation.ts | 6 + src/node/services/workspaceService.test.ts | 35 +++- 4 files changed, 239 insertions(+), 40 deletions(-) diff --git a/src/node/services/agentSession.compactionCancellation.test.ts b/src/node/services/agentSession.compactionCancellation.test.ts index 6831d17241..56fb41c757 100644 --- a/src/node/services/agentSession.compactionCancellation.test.ts +++ b/src/node/services/agentSession.compactionCancellation.test.ts @@ -331,9 +331,13 @@ describe("compaction cancellation runtime", () => { expect(await h.storage.read()).toMatchObject({ version: 1 }); }); - test.each(["complete", "supersede", "dispose"] as const)( - "failed Stop supervises captured startup until %s", - async (finish) => { + test.each( + (["complete", "supersede", "dispose"] as const).flatMap((finish) => + [false, true].map((physicalSuccess) => ({ finish, physicalSuccess })) + ) + )( + "Stop supervises captured startup until $finish (physicalSuccess=$physicalSuccess)", + async ({ finish, physicalSuccess }) => { const h = await fixture(); const entered = Promise.withResolvers(); const release = Promise.withResolvers(); @@ -345,28 +349,152 @@ describe("compaction cancellation runtime", () => { return Err({ type: "unknown", raw: "startup failed" }); }); const sending = h.session.sendMessage("held startup", options); - await entered.promise; - spyOn(h.aiService, "stopStream").mockResolvedValueOnce(Err("stop failed")); + try { + await entered.promise; + spyOn(h.aiService, "stopStream").mockResolvedValueOnce( + physicalSuccess ? Ok(undefined) : Err("stop failed") + ); + expect(await h.session.interruptStream({ onCompactionSettled: notify })).toEqual( + physicalSuccess ? Ok(undefined) : Err("stop failed") + ); + expect(await h.storage.read()).toMatchObject({ version: 1 }); + expect(await h.session.isAutomaticSendBlocked()).toBe(true); + let disposing: Promise | undefined; + if (finish === "supersede") + expect(await h.session.cancelCompaction(true)).toEqual(Ok(undefined)); + if (finish === "dispose") disposing = h.session.dispose(); + release.resolve(); + await sending; + if (finish === "complete") { + await qualified.promise; + expect(await h.storage.read()).toMatchObject({ version: 2 }); + expect(await h.session.isAutomaticSendBlocked()).toBe(false); + } else { + await disposing; + expect(notify).not.toHaveBeenCalled(); + expect(await h.storage.read()).toMatchObject({ version: 1 }); + } + } finally { + release.resolve(); + await sending; + } + } + ); + + test.each(["foreign Stop", "generation"] as const)( + "a superseded deferred Stop cannot notify (%s)", + async (superseded) => { + const h = await fixture(); + const notify = mock(() => undefined); + let finalize: ((complete: boolean | Promise) => Promise) | undefined; + expect( + await h.session.interruptStream({ + onCompactionSettled: notify, + deferCompactionSettlement: (complete) => { + finalize = complete; + }, + }) + ).toEqual(Ok(undefined)); + if (superseded === "foreign Stop") await new CompactionCancellation(h.storage).cancel(); + else await h.historyService.getContinuousCompactionJournal(workspaceId).advanceGeneration(); + const successor = await h.storage.read(); + const finished = Promise.withResolvers(); + const enter = h.state.coordinator.enterExecution.bind(h.state.coordinator); + spyOn(h.state.coordinator, "enterExecution").mockImplementationOnce(() => { + const lease = enter(); + return { + [Symbol.dispose]: () => { + lease[Symbol.dispose](); + finished.resolve(); + }, + }; + }); + await finalize?.(Promise.resolve(true)); + await finished.promise; + expect(notify).not.toHaveBeenCalled(); + expect(await h.storage.read()).toEqual(successor); + } + ); + + test.each(["foreign Stop", "generation"] as const)( + "a committed Stop cannot notify after persisted ownership changes (%s)", + async (superseded) => { + const h = await fixture(); + const notify = mock(() => undefined); + const capture = h.historyService.captureCompactionReplacement.bind(h.historyService); + spyOn(h.historyService, "captureCompactionReplacement").mockImplementationOnce( + async (...args) => { + expect(await h.storage.read()).toMatchObject({ version: 2 }); + if (superseded === "foreign Stop") await new CompactionCancellation(h.storage).cancel(); + else + await h.historyService.getContinuousCompactionJournal(workspaceId).advanceGeneration(); + return capture(...args); + } + ); expect(await h.session.interruptStream({ onCompactionSettled: notify })).toEqual( - Err("stop failed") + Ok(undefined) ); - expect(await h.storage.read()).toMatchObject({ version: 1 }); - expect(await h.session.isAutomaticSendBlocked()).toBe(true); - let disposing: Promise | undefined; - if (finish === "supersede") - expect(await h.session.cancelCompaction(true)).toEqual(Ok(undefined)); - if (finish === "dispose") disposing = h.session.dispose(); - release.resolve(); - await sending; - if (finish === "complete") { - await qualified.promise; - expect(await h.storage.read()).toMatchObject({ version: 2 }); - expect(await h.session.isAutomaticSendBlocked()).toBe(false); - } else { - await disposing; - expect(notify).not.toHaveBeenCalled(); - expect(await h.storage.read()).toMatchObject({ version: 1 }); + expect(notify).not.toHaveBeenCalled(); + } + ); + + test.each([false, true])( + "failed V2 durability cannot notify (after receipt=%s)", + async (afterReceipt) => { + const h = await fixture(); + const notify = mock(() => undefined); + const lock = h.historyService.withCompactionStorageLock.bind(h.historyService); + let calls = 0; + spyOn(h.historyService, "withCompactionStorageLock").mockImplementation(async (...args) => { + if (++calls !== 2) return lock(...args); + if (afterReceipt) await lock(...args); + throw new Error("settlement durability unavailable"); + }); + expect(await h.session.interruptStream({ onCompactionSettled: notify })).toMatchObject({ + success: false, + streamStopped: true, + }); + expect(await h.storage.read()).toMatchObject({ version: afterReceipt ? 2 : 1 }); + expect(notify).not.toHaveBeenCalled(); + } + ); + + test.each([false, true])( + "throwing settlement observer preserves Stop and releases ownership (deferred=%s)", + async (deferred) => { + const h = await fixture(); + const notify = mock(() => { + throw new Error("observer unavailable"); + }); + let finalize: ((complete: boolean | Promise) => Promise) | undefined; + expect( + await h.session.interruptStream({ + onCompactionSettled: notify, + deferCompactionSettlement: deferred + ? (complete) => { + finalize = complete; + } + : undefined, + }) + ).toEqual(Ok(undefined)); + if (deferred) { + const finished = Promise.withResolvers(); + const enter = h.state.coordinator.enterExecution.bind(h.state.coordinator); + spyOn(h.state.coordinator, "enterExecution").mockImplementationOnce(() => { + const lease = enter(); + return { + [Symbol.dispose]: () => { + lease[Symbol.dispose](); + finished.resolve(); + }, + }; + }); + expect(await finalize?.(Promise.resolve(true))).toEqual(Ok(undefined)); + await finished.promise; } + expect(notify).toHaveBeenCalledTimes(1); + expect(await h.storage.read()).toMatchObject({ version: 2 }); + await h.session.dispose(); } ); diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 2f72f165f3..666b7e05e2 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -7325,6 +7325,7 @@ export class AgentSession { fullHistoryDeletion?: true; onCaptured?: (capture: CompactionReplacementCapture) => void; onInitialSettlement?: () => void; + onSettled?: (capture: CompactionReplacementCapture) => void; } ): Promise> { this.pendingStopCompletion?.abort(); @@ -7340,6 +7341,7 @@ export class AgentSession { fullHistoryDeletion: options?.fullHistoryDeletion, onCaptured: options?.onCaptured, onInitialSettlement: options?.onInitialSettlement, + onSettled: options?.onSettled, }); return Ok(undefined); } catch (error) { @@ -7448,12 +7450,16 @@ export class AgentSession { const settled = Promise.withResolvers(); const initiallySettled = Promise.withResolvers>(); let physicallyStopped = false; + let settledCapture: CompactionReplacementCapture | undefined; const cancellation = options?.soft || options?.preserveCompactionIntent ? undefined : this.cancelCompaction(false, settled.promise, { onCaptured: options?.onCompactionCanceled, onInitialSettlement: () => initiallySettled.resolve(Ok(undefined)), + onSettled: (capture) => { + settledCapture = capture; + }, }); const completionController = new AbortController(); if (cancellation) { @@ -7476,19 +7482,53 @@ export class AgentSession { } const deferred = cancellation && options?.deferCompactionSettlement; const generation = this.compactionStopGeneration; - // Failed startup must retain the exact registered producer, even before stream-start. + // Physical Stop can acknowledge startup abort delivery before that original producer exits. const producerCompletion = this.coordinator.captureInterruptSettlement(options?.soft, true); + let producerSettled = producerCompletion === undefined; + producerCompletion?.then( + () => { + producerSettled = true; + }, + () => undefined + ); + const isCurrent = () => + !completionController.signal.aborted && + !this.coordinator.closing && + generation === this.compactionStopGeneration; + const notifySettlement = async (result: Result, completed: boolean) => { + if (!completed || !result.success || !settledCapture || !isCurrent()) return; + try { + const current = await this.historyService.captureCompactionReplacement(this.workspaceId); + if ( + current.success && + current.data.nonce === settledCapture.nonce && + current.data.generation === settledCapture.generation && + isCurrent() + ) + options?.onCompactionSettled?.(); + } catch (error) { + // Notification is ancillary; observers cannot change the original physical Stop result. + log.warn("Stop settlement observer failed", { error }); + } + }; let finalized: Promise> | undefined; const finalize = (cleanupSucceeded: boolean | Promise): Promise> => { if (finalized) return finalized; if (!cancellation) return Promise.resolve(Ok(undefined)); - if ((physicallyStopped && cleanupSucceeded === true) || cleanupSucceeded === false) { - settled.resolve(physicallyStopped && cleanupSucceeded === true); - return (finalized = cancellation); + if ( + (physicallyStopped && producerSettled && cleanupSucceeded === true) || + cleanupSucceeded === false + ) { + const completed = physicallyStopped && producerSettled && cleanupSucceeded === true; + settled.resolve(completed); + return (finalized = cancellation.then(async (result) => { + await notifySettlement(result, completed); + return result; + })); } - // A failed interrupt still returns its error immediately. Its captured producer can - // finish naturally; that exact completion restores automatic admission without losing - // owed monitor attention. The guardian joins this lease, and close/supersession abort it. + // Physical Stop can return before startup unwinds, or fail while its producer is live. + // Preserve that result promptly, but qualify only the captured producer's completion. + // The guardian joins this lease, and close/supersession abort it. const controller = completionController; const execution = this.coordinator.enterExecution(); const abandoned = Promise.withResolvers(); @@ -7502,9 +7542,7 @@ export class AgentSession { ) abandon(); Promise.race([ - Promise.all([physicallyStopped ? undefined : producerCompletion, cleanupSucceeded]).then( - ([, complete]) => complete - ), + Promise.all([producerCompletion, cleanupSucceeded]).then(([, complete]) => complete), abandoned.promise, ]) .then(async (completed) => { @@ -7512,13 +7550,7 @@ export class AgentSession { completed && !this.coordinator.closing && generation === this.compactionStopGeneration; settled.resolve(current); const result = await cancellation; - if ( - current && - result.success && - !this.coordinator.closing && - generation === this.compactionStopGeneration - ) - options?.onCompactionSettled?.(); + await notifySettlement(result, current); }) .catch((error: unknown) => { settled.resolve(false); diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts index afe1479218..12931da028 100644 --- a/src/node/services/compactionCancellation.ts +++ b/src/node/services/compactionCancellation.ts @@ -67,6 +67,8 @@ export type CompactionCancellationMutation = settled?: Promise; onCaptured?: (capture: CompactionReplacementCapture) => void; onInitialSettlement?: () => void; + /** Exact V2 commit receipt, distinct from an applied mutation retaining V1. */ + onSettled?: (capture: CompactionReplacementCapture) => void; } | { kind: "narrow"; record: CompactionCancellationRecord } | { @@ -556,6 +558,7 @@ export class CompactionCancellation { settled?: Promise; onCaptured?: (capture: CompactionReplacementCapture) => void; onInitialSettlement?: () => void; + onSettled?: (capture: CompactionReplacementCapture) => void; fullHistoryDeletion?: true; }): Promise { this.current = { @@ -575,6 +578,7 @@ export class CompactionCancellation { fullHistoryDeletion: options?.fullHistoryDeletion, onCaptured: options?.onCaptured, onInitialSettlement: options?.onInitialSettlement, + onSettled: options?.onSettled, ...(options?.settled ? { settled: options.settled } : {}), }); } @@ -757,6 +761,8 @@ export class CompactionCancellation { nonce: record.nonce, generation: mutation.publication.predecessor.generation, }); + if (record.version === 2) + mutation.onSettled?.({ nonce: record.nonce, generation: record.settledGeneration }); } // Commit invalidates pre-deletion reads before lock release. A later foreign // read must survive acknowledgment delayed by adapter cleanup. diff --git a/src/node/services/workspaceService.test.ts b/src/node/services/workspaceService.test.ts index dba1fb0ff2..9c22765e12 100644 --- a/src/node/services/workspaceService.test.ts +++ b/src/node/services/workspaceService.test.ts @@ -23,7 +23,7 @@ import type { AutoCompactionUsageState } from "@/common/utils/compaction/autoCom import { createDisplayUsage } from "@/common/utils/tokens/displayUsage"; import { askUserQuestionManager } from "./askUserQuestionManager"; import { WorkspaceLifecycleHooks } from "./workspaceLifecycleHooks"; -import { EventEmitter } from "events"; +import { EventEmitter, once } from "events"; import { existsSync } from "fs"; import * as fsPromises from "fs/promises"; import { tmpdir } from "os"; @@ -767,6 +767,39 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { } }); + test("output after retirement wakes when fast Stop settlement completes", async () => { + const h = await createActiveWakeHarness(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + let stopping: ReturnType | undefined; + try { + h.service.setAgentTaskIntegration( + makeAgentTaskIntegrationFake({ + terminateAllDescendantAgentTasks: async () => { + entered.resolve(); + await release.promise; + return []; + }, + }) + ); + stopping = h.service.interruptStream(h.workspaceId, { retireBashMonitorAttention: true }); + await entered.promise; + await h.addAttention(20); + expect(h.requests).toHaveLength(0); + expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(2); + const launched = once(h.launched, "start"); + release.resolve(); + expect(await stopping).toEqual(Ok(undefined)); + await launched; + expect(h.requests).toHaveLength(1); + expect((await h.reconciler.snapshot(h.workspaceId)).pendingWakeKinds.size).toBe(0); + } finally { + release.resolve(); + await stopping; + await h.finish(); + } + }); + test("outer Stop preserves physical completion through exact cleanup retry", async () => { const h = await createActiveWakeHarness(); try { From c35df268be9eda14c305fb6b442e29079628edde Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 11 Sep 2026 16:21:07 +0200 Subject: [PATCH 4/8] =?UTF-8?q?=F0=9F=A4=96=20tests:=20settle=20fake=20pin?= =?UTF-8?q?ned-budget=20turns=20on=20Stop?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Change-Id: I654ec5bc5ef9b25c54bc9d294e1db0714ed0713f Signed-off-by: Thomas Kosiewski --- .../agentSession.pinnedBudget.test.ts | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/src/node/services/agentSession.pinnedBudget.test.ts b/src/node/services/agentSession.pinnedBudget.test.ts index 1bb7f57ce4..fc6228e305 100644 --- a/src/node/services/agentSession.pinnedBudget.test.ts +++ b/src/node/services/agentSession.pinnedBudget.test.ts @@ -214,6 +214,26 @@ async function setup( }; } +// These controls explicitly Stop before disposal, so their fake engine must complete that +// started turn on Stop. The closing signal still retires the subsequent ordinary control turn. +function completeStartedTurnsOnStop(fixture: Awaited>) { + let stopping = new AbortController(); + fixture.start.mockImplementation(async (options) => { + stopping = new AbortController(); + await options.onStreamConstructed?.(); + return Ok( + createStartedTurnHandle( + AbortSignal.any([fixture.h.session.closingSignal, stopping.signal]), + options.messageId + ) + ); + }); + spyOn(fixture.manager, "stopStream").mockImplementation(() => { + stopping.abort(); + return Promise.resolve(Ok(undefined)); + }); +} + describe("pinned full-payload rollover admission", () => { test.each( (["system", "advertised-schema", "deferred-schema"] as const).flatMap((kind) => @@ -627,6 +647,7 @@ describe("pinned full-payload rollover admission", () => { test("a final-flush turn never starts MCP servers", async () => { const fixture = await setup("small"); + completeStartedTurnsOnStop(fixture); const { h, service, start } = fixture; const mcpServerManager = service.turnRequestBuilderBindings.mcpServerManager!; const startServers = spyOn(mcpServerManager, "getToolsForWorkspace"); @@ -687,6 +708,7 @@ describe("pinned full-payload rollover admission", () => { test("a final-flush fallback runs at its own inherent thinking minimum with a matching cap", async () => { const fixture = await setup("small"); + completeStartedTurnsOnStop(fixture); const { h, config, start } = fixture; // gpt-5.2 cannot go below medium thinking, and the user floor for it is higher still; the // flush must ignore the floor (housekeeping) but size its cap for the model's own minimum. From 25aec548208164b9c38deda02b355c19298ec413 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 11 Sep 2026 16:42:22 +0200 Subject: [PATCH 5/8] =?UTF-8?q?=F0=9F=A4=96=20fix:=20preserve=20publicatio?= =?UTF-8?q?n=20rejection=20handling=20after=20activation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Change-Id: I572da59a863113ee84d1d1ed6416c697f7dae02c Signed-off-by: Thomas Kosiewski --- .../agentSession.preparedHistory.test.ts | 26 +++++++++++++++++++ src/node/services/agentSession.ts | 5 +++- 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/src/node/services/agentSession.preparedHistory.test.ts b/src/node/services/agentSession.preparedHistory.test.ts index 41e9845726..ff5bbea87a 100644 --- a/src/node/services/agentSession.preparedHistory.test.ts +++ b/src/node/services/agentSession.preparedHistory.test.ts @@ -128,6 +128,32 @@ describe("prepared history publication", () => { expect(h.stream).not.toHaveBeenCalled(); }); + test("a rejected automatic batch leaves foreign history and publishes no owned prefixes", async () => { + const h = await fixture(); + const foreign = createMuxMessage("foreign", "assistant", "concurrent input"); + expect(await h.historyService.appendToHistory(workspaceId, foreign)).toEqual(Ok(undefined)); + const publish = spyOn(h.historyService, "acceptCompactionReplacement").mockRejectedValueOnce( + new Error("injected batch rejection") + ); + const result = await h.session + .sendMessage("inspect input", options, { + acceptanceOrigin: "automatic", + }) + .catch((error: unknown) => error); + expect(publish).toHaveBeenCalledTimes(1); + const operation = publish.mock.calls[0][2]; + assert(operation.kind === "append"); + expect(operation.preserveCancellation).toBe(true); + expect(operation.messages.slice(0, -1).map((row) => row.id)).toEqual([ + "file", + "skill", + "prompt", + ]); + expect((await h.rows()).map((row) => row.id)).toEqual([foreign.id]); + expect(result).toMatchObject({ success: false, error: { raw: "injected batch rejection" } }); + expect(h.stream).not.toHaveBeenCalled(); + }); + test("automatic cancellation after the batch receipt rolls back only its own rows", async () => { const h = await fixture(); const foreign = createMuxMessage("foreign", "assistant", "concurrent input"); diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 666b7e05e2..8906a92221 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -3536,7 +3536,7 @@ export class AgentSession { const batch = [...stagedPrefixes, ...messages]; attempt.inputPublication = messages.at(-1); assert(replacementCapture, "Publication requires its admission capture"); - const accepted = await this.historyService.acceptCompactionReplacement( + const publishing = this.historyService.acceptCompactionReplacement( this.workspaceId, replacementCapture, { @@ -3564,6 +3564,9 @@ export class AgentSession { }, } ); + // Unexpected rejection follows Result Err's rollback path; the synchronous receipt + // still decides whether input is irrevocable. Retirement and observers remain outside. + const accepted = await publishing.catch((error: unknown) => Err(getErrorMessage(error))); if (!accepted.success) return accepted; if (accepted.data.kind !== "accepted") { // A canceled ordinary append can now refuse under the publication lock before writing. From d63ed25a1a7bd84508918234d5eabdc1a1ce4439 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 11 Sep 2026 17:23:04 +0200 Subject: [PATCH 6/8] =?UTF-8?q?=F0=9F=A4=96=20fix:=20fence=20queued=20auto?= =?UTF-8?q?matic=20input=20with=20its=20original=20Stop=20frontier?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Preserve recovery captures and reject automatic payloads admitted before a foreign Stop, while allowing fresh post-Stop input. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: I0328700c20a8f24b704e3ec0e90b1e349fa0bdcb --- src/node/services/agentSession.ts | 12 ++- src/node/services/workspaceService.test.ts | 104 +++++++++++++++++++++ 2 files changed, 111 insertions(+), 5 deletions(-) diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 8906a92221..f0b0154d64 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -3377,10 +3377,12 @@ export class AgentSession { if (internal?.preparation) return this.prepareMessage(message, options, internal, internal.preparation); if (!internal?.readCompactionAdmission) { - const admission = this.historyService.captureCompactionReplacement(this.workspaceId, { - onRepaired: () => this.clearUsageState(), - replaceUnreadable: (internal?.acceptanceOrigin ?? "manual") === "manual", - }); + const admission = internal?.recoveryReplacement + ? Promise.resolve(Ok(internal.recoveryReplacement)) + : this.historyService.captureCompactionReplacement(this.workspaceId, { + onRepaired: () => this.clearUsageState(), + replaceUnreadable: (internal?.acceptanceOrigin ?? "manual") === "manual", + }); internal = { ...internal, readCompactionAdmission: () => admission }; } const attempt: PreparationAttempt = { @@ -3671,7 +3673,7 @@ export class AgentSession { ? undefined : internal?.recoveryReplacement ? Ok(internal.recoveryReplacement) - : await this.historyService.captureCompactionReplacement(this.workspaceId); + : frontier; if (manualReplacement) { await this.readCompactionCancellation("manual"); const stopAdmission = attempt.queuedStopAdmission; diff --git a/src/node/services/workspaceService.test.ts b/src/node/services/workspaceService.test.ts index 9c22765e12..c13c2a2c04 100644 --- a/src/node/services/workspaceService.test.ts +++ b/src/node/services/workspaceService.test.ts @@ -7809,6 +7809,110 @@ describe("WorkspaceService truncateHistory goal acknowledgment", () => { } ); + test.each(["pricing", "queue"] as const)( + "automatic family work admitted before a foreign Stop stays fenced through %s", + async (stage) => { + const { config, historyService, workspaceService, goalService, cleanup } = + await createServices(); + const workspaceId = `foreign-family-${stage}`; + await config.addWorkspace("/tmp/foreign-family-project", { + id: workspaceId, + name: workspaceId, + projectName: "foreign-family-project", + projectPath: "/tmp/foreign-family-project", + runtimeConfig: { type: "local" }, + }); + const h = await createAgentSessionHarness({ + workspaceId, + config, + historyService, + workspaceGoalService: goalService, + }); + const foreign = await createAgentSessionHarness({ + workspaceId, + config, + historyService: new HistoryService(config), + }); + workspaceService.registerSession(workspaceId, h.session); + const streamStarted = Promise.withResolvers(); + const stream = spyOn(h.aiService, "streamMessage").mockImplementation(() => { + streamStarted.resolve(); + return Promise.resolve(Ok(createStartedTurnHandle(h.session.closingSignal))); + }); + await historyService.appendToHistory( + workspaceId, + createMuxMessage("prior", "user", "old request") + ); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const failed = Promise.withResolvers(); + const price = goalService.assertPricedModelForBudgetedGoal.bind(goalService); + const busy = stage === "queue" ? spyOn(h.session, "isBusy").mockReturnValue(true) : undefined; + if (stage === "pricing") + spyOn(goalService, "assertPricedModelForBudgetedGoal").mockImplementationOnce( + async (...args) => { + entered.resolve(); + await release.promise; + return price(...args); + } + ); + const options = { model: "openai:gpt-4o", agentId: "exec" }; + const dispatched = workspaceService.sendMessage(workspaceId, "stale child trigger", options, { + acceptanceOrigin: "automatic", + synthetic: true, + agentInitiated: true, + preTurnMessages: [ + createMuxMessage("child-payload", "assistant", "stale child payload", { + synthetic: true, + }), + ], + onAcceptedPreStreamFailure: () => { + failed.resolve(); + }, + }); + try { + if (stage === "pricing") await entered.promise; + else { + expect(await dispatched).toEqual(Ok(undefined)); + expect(h.session.hasQueuedMessages()).toBe(true); + } + expect(await foreign.session.interruptStream()).toEqual(Ok(undefined)); + const stopped = await historyService.getCompactionCancellationStorage(workspaceId).read(); + expect(stopped?.version).toBe(2); + release.resolve(); + busy?.mockRestore(); + if (stage === "queue") { + h.session.drainQueuedMessagesIfIdle(); + await Promise.race([failed.promise, streamStarted.promise]); + expect(stream).not.toHaveBeenCalled(); + await h.session.waitForIdle(); + } else expect((await dispatched).success).toBe(false); + const rows = await historyService.getLastMessages(workspaceId, 10); + expect(rows.success && rows.data.map((row) => row.id)).toEqual(["prior"]); + expect(stream).not.toHaveBeenCalled(); + expect(await historyService.getCompactionCancellationStorage(workspaceId).read()).toEqual( + stopped + ); + // The fence belongs to the old admission, not to the automatic origin itself. + expect( + await workspaceService.sendMessage(workspaceId, "fresh child trigger", options, { + acceptanceOrigin: "automatic", + synthetic: true, + agentInitiated: true, + }) + ).toEqual(Ok(undefined)); + expect(stream).toHaveBeenCalledTimes(1); + } finally { + release.resolve(); + busy?.mockRestore(); + await dispatched; + await workspaceService.disposeSession(workspaceId); + await foreign.session.dispose(); + await cleanup(); + } + } + ); + test("requireIdle sends carry a live idle-admission probe re-evaluated at session gates", async () => { // Codex P1 (PRRT_kwDOPxxmWM6cJ6NI): the preflight count check at // sendMessage entry is a one-shot snapshot — a manual send can enter From 0ef43369c2cd8e6b7f700fc109cd948c1dd0e98a Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 11 Sep 2026 18:00:26 +0200 Subject: [PATCH 7/8] =?UTF-8?q?=F0=9F=A4=96=20fix:=20advance=20queued=20au?= =?UTF-8?q?tomatic=20input=20after=20owned=20Stop=20retirement?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Preserve the exact preparation capture through committed reset and Stop deletion, including two separate automatic messages queued after settled Stop. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Ic14d44e4f23e1a66e9987ab4642b8d461ebc6ac9 --- .../services/agentSession.tokenBudget.test.ts | 27 +++++++++++++++++++ src/node/services/agentSession.ts | 5 +++- 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/src/node/services/agentSession.tokenBudget.test.ts b/src/node/services/agentSession.tokenBudget.test.ts index a41bf73420..b2cb88b5c2 100644 --- a/src/node/services/agentSession.tokenBudget.test.ts +++ b/src/node/services/agentSession.tokenBudget.test.ts @@ -1924,6 +1924,33 @@ describe("AgentSession token-budget lifecycle", () => { } ); + test.each([false, true])( + "separate automatic inputs survive settled Stop retirement (reset=%s)", + async (reset) => { + const h = await setup(); + await seedHistory(h, reset ? 110_000 : 20_000); + const foreign = new CompactionCancellation( + new HistoryService(h.config).getCompactionCancellationStorage(workspaceId) + ); + await foreign.cancel({ settled: Promise.resolve(true) }); + for (const message of ["First after Stop", "Second after Stop"]) + h.session.queueMessage(message, options, { + acceptanceOrigin: "automatic", + synthetic: true, + onAccepted: () => undefined, + }); + h.session.sendQueuedMessages(); + await h.waitForRequest(1); + h.settleStream(0, { finishReason: "stop" }); + await h.waitForRequest(2); + const rows = await allRows(h); + expect( + rows.filter((row) => ["First after Stop", "Second after Stop"].includes(text(row))) + ).toHaveLength(2); + expect(rolloverRows(rows)).toHaveLength(reset ? 1 : 0); + } + ); + test("cancellation before an owned reset receipt releases its automatic caller without publication", async () => { const h = await setup(); await seedHistory(h, 110_000); diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index f0b0154d64..6a75fec6aa 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -3549,6 +3549,9 @@ export class AgentSession { { isCurrent: () => !isAdmissionStale() && !shutdownRefusesBeforePersist() && !cancelSignal?.aborted, + onContextResetCommitted: (predecessor, successor) => { + this.advanceOwnedCompactionAdmission(predecessor, successor, attempt.admissionCapture); + }, onCommitted: () => { if (replacesCancellation) { replacementCommitted = true; @@ -3578,7 +3581,7 @@ export class AgentSession { } if (replacesCancellation) { // Retirement takes the same lock; join it only after publication releases that lock. - await this.retireCompactionReplacement(accepted.data.witness); + await this.retireCompactionReplacement(accepted.data.witness, attempt.admissionCapture); if (!manualReplacement) await internal?.onAccepted?.(); } return Ok(undefined); From d00a1ed323440da22cfdcce70aad9c194a9428d4 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 11 Sep 2026 18:08:54 +0200 Subject: [PATCH 8/8] =?UTF-8?q?=F0=9F=A4=96=20tests:=20assert=20local=20St?= =?UTF-8?q?op=20retry=20before=20sidecar=20publication?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Preserve the failed cleanup API result, absent disk state, exact pending nonce and successful durable settlement after retry. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Ia9ca58cfba4cf60793de6ff75556a51cc10922cd --- src/node/services/workspaceService.test.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/node/services/workspaceService.test.ts b/src/node/services/workspaceService.test.ts index c13c2a2c04..347677f5cb 100644 --- a/src/node/services/workspaceService.test.ts +++ b/src/node/services/workspaceService.test.ts @@ -808,11 +808,14 @@ describe("WorkspaceService bash monitor wake reconciler wiring", () => { ); expect(await h.service.interruptStream(h.workspaceId)).toEqual(Err(STOP_UNRECORDED_MESSAGE)); const storage = h.historyService.getCompactionCancellationStorage(h.workspaceId); - const failed = await storage.read(); - expect(failed).toMatchObject({ version: 1 }); const cancellation = ( h.session as unknown as { compactionCancellation: CompactionCancellation } ).compactionCancellation; + // Downgrade cleanup fails before publication; the local Stop still owns its exact retry. + expect(await storage.read()).toBeNull(); + const failed = await cancellation.read(); + expect(failed).toMatchObject({ version: 1 }); + expect(cancellation.needsPersistence).toBe(true); expect(await cancellation.retry()).toBe("applied"); expect(await storage.read()).toMatchObject({ version: 2, nonce: failed?.nonce }); expect(await h.session.isAutomaticSendBlocked()).toBe(false);