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.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..56fb41c757 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,297 @@ describe("compaction cancellation runtime", () => { } ); - test("unresolved Stop refuses fresh automatic input across restart until manual replacement", async () => { + 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).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(); + 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); + 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( + Ok(undefined) + ); + 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(); + } + ); + + 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 +510,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 +917,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 +1043,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 +1151,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 +1163,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..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. @@ -923,18 +945,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..ff5bbea87a 100644 --- a/src/node/services/agentSession.preparedHistory.test.ts +++ b/src/node/services/agentSession.preparedHistory.test.ts @@ -101,76 +101,104 @@ 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("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"); + 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..b2cb88b5c2 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, { @@ -1922,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 cd3b6a3368..6a75fec6aa 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. */ @@ -3375,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 = { @@ -3516,77 +3520,71 @@ 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 publishing = this.historyService.acceptCompactionReplacement( + this.workspaceId, + replacementCapture, + { + kind: "append", + messages: batch, + ...(!replacesCancellation ? { preserveCancellation: true as const } : {}), + }, + { + isCurrent: () => + !isAdmissionStale() && !shutdownRefusesBeforePersist() && !cancelSignal?.aborted, + onContextResetCommitted: (predecessor, successor) => { + this.advanceOwnedCompactionAdmission(predecessor, successor, attempt.admissionCapture); + }, + 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; + }, } + ); + // 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. + // 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, attempt.admissionCapture); + 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 +3670,13 @@ 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 + : internal?.recoveryReplacement + ? Ok(internal.recoveryReplacement) + : frontier; if (manualReplacement) { await this.readCompactionCancellation("manual"); const stopAdmission = attempt.queuedStopAdmission; @@ -3687,6 +3692,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) @@ -7305,14 +7324,19 @@ export class AgentSession { }; } + private pendingStopCompletion?: AbortController; + async cancelCompaction( retainUntilReplacement = false, - settled?: Promise, + settled?: Promise, options?: { fullHistoryDeletion?: true; onCaptured?: (capture: CompactionReplacementCapture) => void; + onInitialSettlement?: () => void; + onSettled?: (capture: CompactionReplacementCapture) => void; } ): Promise> { + this.pendingStopCompletion?.abort(); this.compactionStopGeneration++; this.pendingResumeIntent?.abort(); this.coordinator.abandonCompaction(); @@ -7324,6 +7348,8 @@ export class AgentSession { settled, fullHistoryDeletion: options?.fullHistoryDeletion, onCaptured: options?.onCaptured, + onInitialSettlement: options?.onInitialSettlement, + onSettled: options?.onSettled, }); return Ok(undefined); } catch (error) { @@ -7372,18 +7398,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; @@ -7415,15 +7449,133 @@ 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 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) { + 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; + // 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 && 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; + })); + } + // 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(); + 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([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; + await notifySettlement(result, current); + }) + .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. @@ -7451,10 +7603,19 @@ export class AgentSession { }) .then(async (result) => { if (result.success) await interruptedPolicy; + physicallyStopped = result.success; return result; }) - .finally(() => settled.resolve()); - 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); } @@ -10451,6 +10612,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. @@ -10553,7 +10717,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); @@ -10794,6 +10975,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.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..12931da028 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,8 +64,11 @@ 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; + onInitialSettlement?: () => void; + /** Exact V2 commit receipt, distinct from an applied mutation retaining V1. */ + onSettled?: (capture: CompactionReplacementCapture) => void; } | { kind: "narrow"; record: CompactionCancellationRecord } | { @@ -116,7 +121,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 +136,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 +207,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) { @@ -331,10 +354,11 @@ 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. - 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 +366,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,8 +555,10 @@ export class CompactionCancellation { cancel(options?: { retainUntilReplacement?: boolean; - settled?: Promise; + settled?: Promise; onCaptured?: (capture: CompactionReplacementCapture) => void; + onInitialSettlement?: () => void; + onSettled?: (capture: CompactionReplacementCapture) => void; fullHistoryDeletion?: true; }): Promise { this.current = { @@ -526,6 +577,8 @@ export class CompactionCancellation { publication: { attempts: 0 }, fullHistoryDeletion: options?.fullHistoryDeletion, onCaptured: options?.onCaptured, + onInitialSettlement: options?.onInitialSettlement, + onSettled: options?.onSettled, ...(options?.settled ? { settled: options.settled } : {}), }); } @@ -626,6 +679,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 ) @@ -707,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/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 c5c9938f56..347677f5cb 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"; @@ -20,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"; @@ -764,29 +767,171 @@ 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("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 { + 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 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); + } 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("hard Stop retires old attention and defers fresh attention until manual replacement", async () => { + 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 { await h.session.sendMessage("original", { model: h.model, agentId: "exec" }); @@ -806,43 +951,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(); @@ -969,14 +1121,16 @@ 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 expectMonitorDeferredUntilManual(h, 1); + await woke; + 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" }); @@ -987,16 +1141,18 @@ 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); - 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 +1166,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 +1214,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 +1394,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 +2198,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()); @@ -7654,6 +7812,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 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?.(); } }