apunkt/rowboat
1
0
Fork 0
mirror of https://github.com/rowboatlabs/rowboat.git synced 2026-07-12 21:02:17 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(x): cancellation no longer requires live dependency materialization

Browse source 
advance() resolved the context chain, agent snapshot, live model, and
every snapshotted tool before applying ANY input — including cancel,
which uses none of them. A permanently broken environment (provider
removed from models.json, builtin renamed by an update, context
predecessor unreadable) therefore rejected every advance forever: the
turn could not be cancelled, stopTurn's fallback failed, and the
session rejected sendMessage with TurnNotSettledError permanently — a
bricked session with valid bytes on disk.

Materialization is now a thunk the invocation awaits only for work that
continues the turn: run() short-circuits a cancel input straight to the
cancel path (same commit ritual, same synthetic results, byte-identical
events for healthy turns), while every other input still materializes
first, preserving §15.3's reject-before-touching-the-file ordering.
applyInput narrows to non-cancel inputs.

Tests: cancel succeeds against a broken tool registry (with the
non-cancel input asserted to reject file-unchanged), a broken model
registry, and an unreadable context chain; plus a healthy-cancel
regression pinning the exact event sequence and outcome. All three
broken-environment tests fail against the old code. Design doc §15.3,
§18, §22, and §26.5 updated.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Ramnique Singh 2026-07-11 10:59:42 +05:30
parent 30def374af
commit 49cb64d8ba
3 changed files with 265 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

29
apps/x/packages/core/docs/turn-runtime-design.md
View file

@ -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

169
apps/x/packages/core/src/runtime/turns/runtime.test.ts
View file

@ -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<typeof ToolDescriptor>,
): Promise<RuntimeTool> {
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<ResolvedModel> {
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;

131
apps/x/packages/core/src/runtime/turns/runtime.ts
View file

@ -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<string, RuntimeTool>();
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<MaterializedEnv> => {
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<string, RuntimeTool>();
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<z.infer<typeof ConversationMessage>>;
resolvedAgent: z.infer<typeof ResolvedAgent>;
model: ResolvedModel;
toolsByName: Map<string, RuntimeTool>;
}
// 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<TurnStreamEvent, TurnOutcome>;
private readonly resolvedContext: Array<z.infer<typeof ConversationMessage>>;
private readonly resolvedAgent: z.infer<typeof ResolvedAgent>;
private readonly model: ResolvedModel;
private readonly materialize: () => Promise<MaterializedEnv>;
// Assigned by run() immediately after the cancel fast-path, before any
// loop phase can touch them. cancel() must never read these.
private resolvedContext!: Array<z.infer<typeof ConversationMessage>>;
private resolvedAgent!: z.infer<typeof ResolvedAgent>;
private model!: ResolvedModel;
private toolsByName!: Map<string, RuntimeTool>;
private readonly usageReporter: IUsageReporter;
private readonly toolsByName: Map<string, RuntimeTool>;
private readonly resolveTool: (
descriptor: z.infer<typeof ToolDescriptor>,
) => Promise<RuntimeTool>;
@ -339,11 +355,8 @@ class TurnAdvance {
events: TEvent[];
state: TurnState;
stream: HotStream<TurnStreamEvent, TurnOutcome>;
resolvedContext: Array<z.infer<typeof ConversationMessage>>;
resolvedAgent: z.infer<typeof ResolvedAgent>;
model: ResolvedModel;
materialize: () => Promise<MaterializedEnv>;
usageReporter: IUsageReporter;
toolsByName: Map<string, RuntimeTool>;
resolveTool: (
descriptor: z.infer<typeof ToolDescriptor>,
) => Promise<RuntimeTool>;
@ -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<TurnOutcome> {
// §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<TurnOutcome | undefined> {
input: Exclude<TurnExternalInput, { type: "cancel" }>,
): Promise<void> {
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,