diff --git a/apps/x/packages/core/docs/turn-runtime-design.md b/apps/x/packages/core/docs/turn-runtime-design.md index 5f1e4402..bda228ad 100644 --- a/apps/x/packages/core/docs/turn-runtime-design.md +++ b/apps/x/packages/core/docs/turn-runtime-design.md @@ -1325,6 +1325,10 @@ Missing or mismatched runtime dependencies: - Do not append `turn_failed`. - Leave the turn unchanged so the caller can fix its environment and retry. +Cancel inputs are exempt from this validation: they are applied before +dependency materialization (section 22), because cancellation is the +terminal exit for environments that never come back. + Provider failures after actual execution begins are modeled turn failures. ## 16. Hot TurnExecution stream @@ -1478,10 +1482,13 @@ At a high level, `advanceTurn` performs: 3. Publish turn-processing-start. 4. Read and validate JSONL. 5. Reduce events to TurnState. -6. Materialize context through the injected context resolver. -7. Validate terminal state, input, and runtime dependencies. -8. Apply the optional single external input. -9. Repeatedly advance deterministic work: +6. Validate terminal state and the optional input. +7. If the input is a cancel, cancel immediately — live dependencies are + never materialized for cancellation (section 22). +8. Materialize context through the injected context resolver and validate + runtime dependencies. +9. Apply the optional single external input. +10. Repeatedly advance deterministic work: a. Recover from the current durable boundary. b. Resolve permissions. c. Execute eligible sync tools. @@ -1494,9 +1501,9 @@ At a high level, `advanceTurn` performs: j. Persist normalized non-delta events and stream deltas. k. Append model_call_completed or model_call_failed. l. Complete when the response has no tool calls. -10. Publish turn-processing-end in finalization. -11. Release the lock. -12. Resolve/reject outcome and close/error event stream. +11. Publish turn-processing-end in finalization. +12. Release the lock. +13. Resolve/reject outcome and close/error event stream. ``` The loop does not read session queues, accept new user messages, switch agents, @@ -1592,6 +1599,11 @@ Rules: - Append `turn_cancelled`. - Never initiate another model call. - Reject late permission decisions, progress, and results after cancellation. +- A cancel input is applied before live-dependency materialization: + cancellation never requires the context, agent snapshot, model, or tools + to resolve, so a turn whose environment is no longer resolvable (provider + removed from config, builtin renamed, context chain unreadable) can + always be cancelled — the terminal exit that keeps its session usable. Sync tools are cooperatively cancellable. A tool that ignores the signal may not settle immediately. The runtime cannot guarantee rollback of external side @@ -1791,6 +1803,9 @@ Assertions: - Turn becomes terminal cancelled. - No subsequent model request occurs. - Late external inputs are rejected. +- Cancellation succeeds when live dependencies can no longer be resolved, + while non-cancel inputs against the same broken environment reject with + the turn file unchanged. ### 26.6 Failures diff --git a/apps/x/packages/core/src/runtime/turns/runtime.test.ts b/apps/x/packages/core/src/runtime/turns/runtime.test.ts index 47e51c35..6deb6b64 100644 --- a/apps/x/packages/core/src/runtime/turns/runtime.test.ts +++ b/apps/x/packages/core/src/runtime/turns/runtime.test.ts @@ -320,6 +320,8 @@ function makeRuntime(opts: { agentError?: string; repo?: InMemoryTurnRepo; idStart?: number; + modelRegistry?: IModelRegistry; + toolRegistry?: IToolRegistry; } = {}) { const repo = opts.repo ?? new InMemoryTurnRepo(); const models = new FakeModelRegistry(opts.models ?? []); @@ -333,8 +335,8 @@ function makeRuntime(opts: { idGenerator: new FakeIdGen(opts.idStart ?? 0), clock: new FakeClock(), agentResolver: new FakeAgentResolver(opts.agent ?? defaultAgent, opts.agentError), - modelRegistry: models, - toolRegistry: new FakeToolRegistry(opts.tools ?? defaultTools), + modelRegistry: opts.modelRegistry ?? models, + toolRegistry: opts.toolRegistry ?? new FakeToolRegistry(opts.tools ?? defaultTools), contextResolver: new TurnRepoContextResolver({ turnRepo: repo }), permissionChecker: checker, permissionClassifier: classifier, @@ -2232,6 +2234,169 @@ describe("concurrent sync tool execution (10.5)", () => { }); }); +// Registries that can be broken mid-test: resolution throws once `error` +// is set, simulating a provider removed from config or a renamed builtin. +class BreakableToolRegistry implements IToolRegistry { + error?: Error; + constructor(private readonly inner: IToolRegistry) {} + async resolve( + descriptor: z.infer, + ): Promise { + if (this.error) { + throw this.error; + } + return this.inner.resolve(descriptor); + } +} + +class BreakableModelRegistry implements IModelRegistry { + error?: Error; + constructor(private readonly inner: IModelRegistry) {} + async resolve( + descriptor: ResolvedModel["descriptor"], + ): Promise { + if (this.error) { + throw this.error; + } + return this.inner.resolve(descriptor); + } +} + +describe("cancellation without live dependencies (22)", () => { + it("cancels a suspended turn whose tool registry can no longer resolve", async () => { + const toolRegistry = new BreakableToolRegistry( + new FakeToolRegistry(defaultTools), + ); + const { runtime, repo } = makeRuntime({ + toolRegistry, + models: [ + respond( + completedResp(assistantCalls(toolCallPart("B", "fetch"))), + ), + ], + }); + const turnId = await newTurn(runtime); + const first = await advanceAndSettle(runtime, turnId); + expect(first.outcome?.status).toBe("suspended"); + + toolRegistry.error = new TurnDependencyError( + "no live tool for tool.fetch", + ); + const before = await persisted(repo, turnId); + + // Continuing the turn is impossible — and must not touch the file. + const blocked = await advanceAndSettle(runtime, turnId, { + type: "async_tool_result", + toolCallId: "B", + result: { output: "late", isError: false }, + }); + expect(blocked.outcome).toBeUndefined(); + expect(String(blocked.error)).toContain("no live tool"); + expect(await persisted(repo, turnId)).toEqual(before); + + // Cancelling must still work: synthetic results + turn_cancelled. + const cancelled = await advanceAndSettle(runtime, turnId, { + type: "cancel", + reason: "environment broken", + }); + expect(cancelled.outcome?.status).toBe("cancelled"); + const log = await persisted(repo, turnId); + expect(typesOf(log)).toEqual([ + "turn_created", + "model_call_requested", + "model_call_completed", + "tool_invocation_requested", + "turn_suspended", + "tool_result", + "turn_cancelled", + ]); + expect(log.find((e) => e.type === "tool_result")).toMatchObject({ + source: "runtime", + result: { isError: true }, + }); + expect(log.find((e) => e.type === "turn_cancelled")).toMatchObject({ + reason: "environment broken", + }); + + // Terminal short-circuit works with the environment still broken. + const again = await advanceAndSettle(runtime, turnId); + expect(again.outcome?.status).toBe("cancelled"); + }); + + it("cancels a suspended turn whose model can no longer resolve", async () => { + const inner = new FakeModelRegistry([ + respond(completedResp(assistantCalls(toolCallPart("B", "fetch")))), + ]); + const modelRegistry = new BreakableModelRegistry(inner); + const { runtime, repo } = makeRuntime({ modelRegistry }); + const turnId = await newTurn(runtime); + const first = await advanceAndSettle(runtime, turnId); + expect(first.outcome?.status).toBe("suspended"); + + modelRegistry.error = new Error("provider not configured"); + const cancelled = await advanceAndSettle(runtime, turnId, { + type: "cancel", + }); + expect(cancelled.outcome?.status).toBe("cancelled"); + const log = await persisted(repo, turnId); + expect(typesOf(log)).toContain("turn_cancelled"); + }); + + it("cancels a turn whose context chain is unreadable", async () => { + const { runtime, repo } = makeRuntime(); + const turnId = await runtime.createTurn({ + agent: { agentId: "copilot" }, + context: { previousTurnId: "2026-07-02T10-00-00Z-9999999-000" }, + input: user("hello"), + config: { humanAvailable: true }, + }); + + // Resuming rejects: the referenced predecessor does not exist. + const blocked = await advanceAndSettle(runtime, turnId); + expect(blocked.outcome).toBeUndefined(); + expect(blocked.error).toBeTruthy(); + + const cancelled = await advanceAndSettle(runtime, turnId, { + type: "cancel", + }); + expect(cancelled.outcome?.status).toBe("cancelled"); + expect(typesOf(await persisted(repo, turnId))).toEqual([ + "turn_created", + "turn_cancelled", + ]); + }); + + it("healthy-environment cancellation is byte-identical to before", async () => { + const { runtime, repo } = makeRuntime({ + models: [ + respond( + completedResp(assistantCalls(toolCallPart("B", "fetch"))), + ), + ], + }); + const turnId = await newTurn(runtime); + await advanceAndSettle(runtime, turnId); + const cancelled = await advanceAndSettle(runtime, turnId, { + type: "cancel", + reason: "user stop", + }); + expect(cancelled.outcome).toEqual({ + status: "cancelled", + reason: "user stop", + usage: { inputTokens: 10, outputTokens: 5, totalTokens: 15 }, + }); + expect(typesOf(await persisted(repo, turnId))).toEqual([ + "turn_created", + "model_call_requested", + "model_call_completed", + "tool_invocation_requested", + "turn_suspended", + "tool_result", + "turn_cancelled", + ]); + }); +}); + // Fails appends whose batch matches the predicate; counts the failures. class FailingAppendRepo extends InMemoryTurnRepo { failWhen?: (events: TEvent[]) => boolean; diff --git a/apps/x/packages/core/src/runtime/turns/runtime.ts b/apps/x/packages/core/src/runtime/turns/runtime.ts index 06b8195b..bca8a857 100644 --- a/apps/x/packages/core/src/runtime/turns/runtime.ts +++ b/apps/x/packages/core/src/runtime/turns/runtime.ts @@ -217,10 +217,12 @@ export class TurnRuntime implements ITurnRuntime { } } - // §18 steps 4–7: read, reduce, short-circuit terminal turns, materialize - // context, and validate live dependencies — all before any mutation. - // Failures here are infrastructure errors: the execution rejects and the - // turn is left unchanged. + // §18 steps 4–7: read, reduce, short-circuit terminal turns, and prepare + // live-dependency materialization. Materialization itself is deferred + // into run(): non-cancel work awaits it first, so failures there remain + // infrastructure errors that reject the execution with the turn + // unchanged — but a cancel input never needs it (§22), so a turn whose + // environment can no longer be resolved can always be cancelled. private async advance( turnId: string, input: TurnExternalInput | undefined, @@ -238,32 +240,36 @@ export class TurnRuntime implements ITurnRuntime { } const definition = state.definition; - const resolvedContext = await this.contextResolver.resolve( - definition.context, - ); - const resolvedAgent = await this.contextResolver.resolveAgent( - definition.agent.resolved, - ); - const model = await this.modelRegistry.resolve(resolvedAgent.model); - // Base snapshot plus every durable mid-turn extension already in the - // log — rebuilding from durable state IS the crash-recovery path. - const toolsByName = new Map(); - for (const descriptor of effectiveTools( - state, - state.modelCalls.length, - resolvedAgent.tools, - )) { - const tool = await this.toolRegistry.resolve(descriptor); - if ( - tool.descriptor.toolId !== descriptor.toolId || - tool.descriptor.execution !== descriptor.execution - ) { - throw new TurnDependencyError( - `resolved tool ${descriptor.toolId} does not match its persisted descriptor`, - ); + const materialize = async (): Promise => { + const resolvedContext = await this.contextResolver.resolve( + definition.context, + ); + const resolvedAgent = await this.contextResolver.resolveAgent( + definition.agent.resolved, + ); + const model = await this.modelRegistry.resolve(resolvedAgent.model); + // Base snapshot plus every durable mid-turn extension already in + // the log — rebuilding from durable state IS the crash-recovery + // path. + const toolsByName = new Map(); + for (const descriptor of effectiveTools( + state, + state.modelCalls.length, + resolvedAgent.tools, + )) { + const tool = await this.toolRegistry.resolve(descriptor); + if ( + tool.descriptor.toolId !== descriptor.toolId || + tool.descriptor.execution !== descriptor.execution + ) { + throw new TurnDependencyError( + `resolved tool ${descriptor.toolId} does not match its persisted descriptor`, + ); + } + toolsByName.set(descriptor.name, tool); } - toolsByName.set(descriptor.name, tool); - } + return { resolvedContext, resolvedAgent, model, toolsByName }; + }; const controller = new AbortController(); const forwardAbort = () => controller.abort(); @@ -282,11 +288,8 @@ export class TurnRuntime implements ITurnRuntime { events, state, stream, - resolvedContext, - resolvedAgent, - model, + materialize, usageReporter: this.usageReporter, - toolsByName, resolveTool: (descriptor) => this.toolRegistry.resolve(descriptor), signal: controller.signal, turnRepo: this.turnRepo, @@ -305,6 +308,16 @@ export class TurnRuntime implements ITurnRuntime { } } +// The live execution environment resolved from a turn's durable snapshot. +// Materialized lazily by run(): every phase that continues the turn needs +// it; cancellation deliberately does not (§22). +interface MaterializedEnv { + resolvedContext: Array>; + resolvedAgent: z.infer; + model: ResolvedModel; + toolsByName: Map; +} + // One advanceTurn invocation. Owns the per-invocation context and implements // the §18 main loop as one method per phase. All state it acts on is derived // from the durable log via the shared reducer after every append. @@ -313,11 +326,14 @@ class TurnAdvance { private readonly events: TEvent[]; private state: TurnState; private readonly stream: HotStream; - private readonly resolvedContext: Array>; - private readonly resolvedAgent: z.infer; - private readonly model: ResolvedModel; + private readonly materialize: () => Promise; + // Assigned by run() immediately after the cancel fast-path, before any + // loop phase can touch them. cancel() must never read these. + private resolvedContext!: Array>; + private resolvedAgent!: z.infer; + private model!: ResolvedModel; + private toolsByName!: Map; private readonly usageReporter: IUsageReporter; - private readonly toolsByName: Map; private readonly resolveTool: ( descriptor: z.infer, ) => Promise; @@ -339,11 +355,8 @@ class TurnAdvance { events: TEvent[]; state: TurnState; stream: HotStream; - resolvedContext: Array>; - resolvedAgent: z.infer; - model: ResolvedModel; + materialize: () => Promise; usageReporter: IUsageReporter; - toolsByName: Map; resolveTool: ( descriptor: z.infer, ) => Promise; @@ -358,11 +371,8 @@ class TurnAdvance { this.events = init.events; this.state = init.state; this.stream = init.stream; - this.resolvedContext = init.resolvedContext; - this.resolvedAgent = init.resolvedAgent; - this.model = init.model; + this.materialize = init.materialize; this.usageReporter = init.usageReporter; - this.toolsByName = init.toolsByName; this.resolveTool = init.resolveTool; this.signal = init.signal; this.turnRepo = init.turnRepo; @@ -457,11 +467,24 @@ class TurnAdvance { // appends durable facts and lets the loop continue, or produces the // invocation's outcome. async run(input: TurnExternalInput | undefined): Promise { + // §22: cancel is the one input that must keep working when the + // environment is gone (provider unconfigured, builtin renamed, + // context chain unreadable) — it needs no live dependencies, so it + // runs before materialization. + if (input?.type === "cancel") { + this.cancelReason = input.reason; + return this.cancel(); + } + // §15.3: everything that continues the turn materializes first, so + // a broken environment rejects as infrastructure before any input + // is persisted and the turn is left unchanged. + const env = await this.materialize(); + this.resolvedContext = env.resolvedContext; + this.resolvedAgent = env.resolvedAgent; + this.model = env.model; + this.toolsByName = env.toolsByName; if (input) { - const cancelled = await this.applyInput(input); - if (cancelled) { - return cancelled; - } + await this.applyInput(input); } for (;;) { if (this.signal.aborted) { @@ -506,14 +529,12 @@ class TurnAdvance { } // §11.2: exactly one input, validated against durable pending state. - // Returns an outcome only for cancel inputs. + // Cancel inputs never reach here — run() short-circuits them before + // materialization. private async applyInput( - input: TurnExternalInput, - ): Promise { + input: Exclude, + ): Promise { switch (input.type) { - case "cancel": - this.cancelReason = input.reason; - return this.cancel(); case "permission_decision": { const tc = this.state.toolCalls.find( (t) => t.toolCallId === input.toolCallId,