From 65b3a2fe450ce372067ceaa6c321be234a533a37 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 11:56:07 -0500 Subject: [PATCH 01/11] feat(testing): checkCrashRecovery; fix three crash-recovery bugs it found - checkCrashRecovery kills a durable run after every log append and while every effect is in flight, resumes from the store, and checks the resume against an uninterrupted run. - A child that finished just before a crash now reports to its parent on resume, at any depth. - Journal writes are batched per settled step, so a parent's sendTo into a child (and the child's reply) land with their cause. - @agent.usage carries callKey; a re-executed call's usage is counted once. --- .changeset/crash-recovery.md | 11 + docs/event-log.md | 3 +- docs/persistence.md | 66 +++ docs/usage-and-budgets.md | 2 + docs/verify.md | 1 + skills/build-machine/SKILL.md | 5 +- src/agent-usage-event.test.ts | 3 +- src/crash-recovery.test.ts | 296 ++++++++++++ src/crash-recovery.ts | 747 ++++++++++++++++++++++++++++++ src/durable-child-state.test.ts | 83 ++++ src/run-agent.ts | 128 ++++- src/serialize-trace-event.test.ts | 2 +- src/testing/index.ts | 9 + src/usage.ts | 5 + src/verify-round9.test.ts | 4 +- 15 files changed, 1349 insertions(+), 16 deletions(-) create mode 100644 .changeset/crash-recovery.md create mode 100644 src/crash-recovery.test.ts create mode 100644 src/crash-recovery.ts diff --git a/.changeset/crash-recovery.md b/.changeset/crash-recovery.md new file mode 100644 index 00000000..9a186168 --- /dev/null +++ b/.changeset/crash-recovery.md @@ -0,0 +1,11 @@ +--- +"@statelyai/agent": patch +--- + +Add `checkCrashRecovery` to `@statelyai/agent/testing`: it kills a durable run after every log append and while every effect is in flight, resumes each from the store's log, and checks the resume against an uninterrupted run (no durable effect re-runs, none skipped, same `callKey` and request, same outcome, the log replays). Pass `store` to check your own store adapter. + +Fix: a child machine that finished just before a crash (its last entry journaled, its parent's `xstate.done.actor` not yet) no longer leaves the resumed parent waiting forever. The restored child's completion is now delivered to its parent, at any depth. + +Fix: a call re-executed after a crash no longer reports its usage twice. `@agent.usage` now carries the call's `callKey`; when the log already holds usage for that key, the repeat is dropped (`usage.dropped` with `reason: "repeat"`), so a token budget counts each call once. + +Fix: a parent's `sendTo` into a journaled child machine, and the child's reply, are no longer lost when the run dies between appends. The store now receives every entry since the loop last waited in one atomic append (`expectedIndex` is the batch's first index), so an event and the child inputs it caused land together. diff --git a/docs/event-log.md b/docs/event-log.md index 629d1d6d..e09b42c4 100644 --- a/docs/event-log.md +++ b/docs/event-log.md @@ -74,7 +74,8 @@ const result = await runToQuiescence( ); ``` -- Each entry is written at its own `index` as `expectedIndex`, so a concurrent writer conflicts instead of interleaving. +- Entries are written in batches, each at its first entry's `index` as `expectedIndex`, so a concurrent writer conflicts instead of interleaving. +- A batch holds every entry since the loop last waited: an event, the child inputs its transition sent, and the events those children sent back land in one atomic append. A crash never keeps a cause without its effect. - No model call starts until every entry before it is durable. Pure transitions never wait. - A rejected write (an `AgentEventLogConflictError`, or any storage failure) aborts in-flight work and settles the run `{ status: "error", cause: "journal" }`. No further calls run. - A `@agent.usage` entry that arrives after the run settled is still written, but the result does not wait for it. Await `result.drain()` before terminating the process if you care about those straggler entries; it resolves once every write issued so far has landed, and never rejects (a failed write already settled the run with `cause: "journal"`). diff --git a/docs/persistence.md b/docs/persistence.md index 730335bb..87c5d877 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -70,6 +70,72 @@ if (result.ignored) { - A log that already reached a final state settles immediately with the recorded output. - The resumed result's `events` extends the same log, so the whole history stays replayable. +## Crash testing + +`checkCrashRecovery` from `@statelyai/agent/testing` kills a run at every point it could die and checks that the resume matches a run that never died. + +```ts +import { createAsyncLogic } from "xstate"; +import { z } from "zod"; +import { setupAgent, createTextLogic } from "@statelyai/agent"; +import { checkCrashRecovery } from "@statelyai/agent/testing"; + +const machine = setupAgent({ + context: z.object({ order: z.string() }), + events: {}, + actors: { + lookup: createAsyncLogic({ run: async () => "order-7" }), + reply: createTextLogic({ model: "m", prompt: "Reply." }), + }, +}).createMachine({ + context: { order: "" }, + initial: "lookup", + states: { + lookup: { + invoke: { + src: "lookup", + onDone: ({ event }) => ({ target: "reply", context: { order: event.output } }), + }, + }, + reply: { invoke: { src: "reply", onDone: { target: "done" } } }, + done: { type: "final" }, + }, +}); + +const report = await checkCrashRecovery(machine, { + executors: { generateText: async () => ({ result: "Your order shipped." }) }, +}); +report.failures; // [] when every crash point recovers +``` + +It runs the machine once uninterrupted, then once per crash point, each resumed from the store's log by a fresh runtime: + +- **Log points:** before the first append, and after each append. The next write never lands. +- **Effect points:** while each model call or plain actor is in flight. It ran, but its result was never recorded. + +Each resumed run is checked against the uninterrupted one. Each failure names its crash point and one of these problems: + +| `problem` | Meaning | +| ----------------- | ----------------------------------------------------------------------------------------- | +| `reran` | An effect ran again although its completion was durable before the crash | +| `skipped` | An effect of the uninterrupted run never ran | +| `unexpected` | An effect ran that the uninterrupted run never made, such as a call under a new `callKey` | +| `request-changed` | A model call re-executed under its `callKey` with a different request | +| `outcome` | The resume ended with a different status, output, state, or context | +| `replay` | The final log does not replay to the resumed state | +| `stuck` | The crashed run or its resume did not settle within `timeout` (default 5000 ms) | +| `threw` | The resume threw | + +- An effect in flight at the crash may run twice. That is the at-least-once contract (see [Idempotency keys](hosts.md#idempotency-keys)), and it is not reported. +- Model calls are keyed by `callKey` without the execution id, and plain actors by invoke site. Only actors bound through string-keyed sources are seen. +- Model calls run live only in the uninterrupted run. After a crash, a call it made gets its recorded answer, after the same delay, so a fake may answer by call order. +- Plain actors run live every time, so they must give the same result however often they run. A fake that keeps state passes `reset`, which runs before every run. +- The outcome check compares final states, so a machine whose result depends on which concurrent call finishes first can fail it after a crash. Make the result order-independent, or test that part without concurrency. +- `events` are host events, delivered one per idle settle. A resume skips the ones its log already holds. +- `store` takes a factory for your own store adapter, one fresh store per run. A store that acknowledges a write it then loses shows up as `reran`. + +Run it in CI next to the machine's other tests. It needs no API key. + ## Snapshot as cache `persist()` output carries `agentMeta: { machineId, version, logId, logIndex }` — the lineage and position of the log it caches. diff --git a/docs/usage-and-budgets.md b/docs/usage-and-budgets.md index fbd83f9c..9cc98554 100644 --- a/docs/usage-and-budgets.md +++ b/docs/usage-and-budgets.md @@ -190,6 +190,7 @@ A wrapper is still host-side. The machine cannot read its counts or react to the - Delivery is opt-in. The event is delivered only when an active state declares a transition that matches it, either `'@agent.usage'` or the wildcard `'*'`. Otherwise no usage event is sent to the machine. Wildcard matching follows XState semantics, so a catch-all `on: { '*': … }` does receive `@agent.usage`. Narrow the handler on `event.type` when the catch-all is meant for other events. - Declare the handler at machine level. A root `on: { '@agent.usage': … }` catches every call regardless of which state made it. A state-scoped handler drops calls made in other states. - Delivery goes to the run's root machine, including usage from requests inside an invoked child machine. Attribute those calls with `id`, `src`, and `model`. +- Each call is counted once across a crash. A call in flight at a crash re-executes on resume, and when the log already holds its usage, matched by `callKey`, the repeat is not delivered or journaled. `onTrace` reports it as `usage.dropped` with `reason: "repeat"`. The payload has this shape. @@ -202,6 +203,7 @@ The payload has this shape. src?: string, // the request's invoke src model?: string, // the model ref it targeted name?: string, // a text request's registered name + callKey?: string, // the call's idempotency key, with an event log budget?: { modelCalls, maxModelCalls }, // counted against maxModelCalls so far, this call included } ``` diff --git a/docs/verify.md b/docs/verify.md index 436de0a7..fc830487 100644 --- a/docs/verify.md +++ b/docs/verify.md @@ -480,6 +480,7 @@ For machines authored as data, validate the config with `validateAgentConfig(con ## Related +- [Crash testing](persistence.md#crash-testing): `checkCrashRecovery` kills a run at every log append and in-flight effect, and checks each resume. - [Debugging](debugging.md): scripted reproduction and the diagnostic codes in context. - [Evals](evals.md): scoring runs on output, trajectory, and budget. - [Quickstart](quickstart.md#define-and-run-one-artifact): a plain-function executor for `runToQuiescence`. diff --git a/skills/build-machine/SKILL.md b/skills/build-machine/SKILL.md index 4ca5bd7c..2759ae48 100644 --- a/skills/build-machine/SKILL.md +++ b/skills/build-machine/SKILL.md @@ -62,7 +62,10 @@ expect(result).toMatchObject({ reachable: false, complete: true }); ## 5. Crash and resume -Run with a store and a thread id, kill the run mid-flight (throw from an executor, or exit the process), then call `runAgent` again with the same store and `threadId`. +Call `checkCrashRecovery(machine, { input, events, executors })` from `@statelyai/agent/testing` for each scenario, and assert `report.failures` is `[]`. It kills the run after every log append and while every effect is in flight, resumes each from the log, and checks the resume against an uninterrupted run. See `docs/persistence.md#crash-testing`. + +- Executors and fake tools must answer from the request, not call order. A fake that keeps state passes `reset`. +- A failure names the crash point and the problem (`reran`, `skipped`, `outcome`, `stuck`, …). Fix the machine or the effect, never the check. - A completed model call never runs again. An in-flight one re-runs with the same `info.callKey`. - A plain actor (`createAsyncLogic`) in flight at the crash re-runs. It gets no call key, so derive an idempotency key from the domain (the ticket id, the booking key) and dedupe in the effect. diff --git a/src/agent-usage-event.test.ts b/src/agent-usage-event.test.ts index e3abdf10..49ed93a7 100644 --- a/src/agent-usage-event.test.ts +++ b/src/agent-usage-event.test.ts @@ -105,7 +105,7 @@ describe("@agent.usage (reserved per-call usage event)", () => { expect(result.usage).toEqual({ totalTokens: 1200, inputTokens: 900, modelCalls: 3 }); }); - test("carries attribution: kind, invoke id, src, model, and the request name", async () => { + test("carries attribution: kind, invoke id, src, model, the request name, and the call key", async () => { const delivered: AgentUsageEvent[] = []; await runToQuiescence( createAgentRuntime(budgetMachine, { @@ -129,6 +129,7 @@ describe("@agent.usage (reserved per-call usage event)", () => { src: "researchStep", model: "test-model", name: "researchStep", + callKey: expect.stringMatching(/:research#1$/), usage: { totalTokens: 400, inputTokens: 300 }, budget: { modelCalls: 1, maxModelCalls: 100 }, }); diff --git a/src/crash-recovery.test.ts b/src/crash-recovery.test.ts new file mode 100644 index 00000000..86b76d8b --- /dev/null +++ b/src/crash-recovery.test.ts @@ -0,0 +1,296 @@ +import { describe, expect, test } from "vitest"; +import { createAsyncLogic } from "xstate"; +import { z } from "zod"; +import { checkCrashRecovery } from "./crash-recovery.js"; +import { createInMemoryEventLogStore, type AgentEventLogStore } from "./event-log-store.js"; +import { + AGENT_USAGE_EVENT_TYPE, + createAgentSchemas, + createDecisionLogic, + createTextLogic, + setupAgent, + type AgentDecisionExecutor, +} from "./index.js"; + +const executors = { + generateText: async (request: { prompt?: unknown }) => ({ + result: `re: ${String(request.prompt)}`, + }), + decide: (async () => ({ event: { type: "APPROVE" } as never })) as AgentDecisionExecutor, +}; + +// tool → draft → decide → (idle for REVIEW) → publish +function createPipeline() { + const draft = createTextLogic({ model: "m", prompt: "Draft." }); + const choose = createDecisionLogic({ + model: "m", + prompt: "Approve?", + allowedEvents: ["APPROVE"] as const, + }); + return setupAgent({ + schemas: createAgentSchemas({ + context: z.object({ fetched: z.number(), draft: z.string(), published: z.boolean() }), + events: { APPROVE: z.object({}), REVIEW: z.object({ ok: z.boolean() }) }, + }), + actors: { + draft, + choose, + fetch: createAsyncLogic({ run: async () => 42 }), + publish: createAsyncLogic({ run: async () => true }), + }, + }).createMachine({ + context: { fetched: 0, draft: "", published: false }, + initial: "fetching", + states: { + fetching: { + invoke: { + id: "fetch", + src: "fetch", + onDone: ({ context, event }) => ({ + target: "drafting", + context: { ...context, fetched: event.output as number }, + }), + }, + }, + drafting: { + invoke: { + id: "draft", + src: "draft", + onDone: ({ context, event }) => ({ + target: "deciding", + context: { ...context, draft: JSON.stringify(event.output) }, + }), + }, + }, + deciding: { + invoke: { id: "choose", src: "choose", input: {} }, + on: { APPROVE: { target: "review" } }, + }, + review: { on: { REVIEW: { target: "publishing" } } }, + publishing: { + invoke: { + id: "publish", + src: "publish", + onDone: ({ context }) => ({ target: "done", context: { ...context, published: true } }), + }, + }, + done: { type: "final", output: ({ context }) => context }, + }, + }); +} + +describe("checkCrashRecovery", () => { + test("a pipeline recovers from a crash at every log point and in-flight effect", async () => { + const report = await checkCrashRecovery(createPipeline(), { + executors, + events: [{ type: "REVIEW", ok: true }], + }); + expect(report.failures).toEqual([]); + expect(report.ok).toBe(true); + expect(report.baseline.status).toBe("done"); + expect(report.baseline.output).toMatchObject({ fetched: 42, published: true }); + expect(report.baseline.effects.map((effect) => effect.key)).toEqual([ + "actor:fetch", + "draft#1", + "choose#1.0", + "actor:publish", + ]); + const kinds = report.points.map((point) => point.kind); + expect(kinds.filter((kind) => kind === "effect")).toHaveLength(4); + expect(kinds.filter((kind) => kind === "log").length).toBeGreaterThan(2); + }); + + test("a child machine's tool does not re-run once its completion is durable", async () => { + const worker = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { + tool: createAsyncLogic({ run: async () => 1 }), + summarize: createTextLogic({ model: "m", prompt: "Sum." }), + }, + }).createMachine({ + context: {}, + initial: "tool", + states: { + tool: { invoke: { id: "tool", src: "tool", onDone: { target: "work" } } }, + work: { invoke: { id: "summarize", src: "summarize", onDone: { target: "done" } } }, + done: { type: "final" }, + }, + }); + const parent = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { worker }, + }).createMachine({ + context: {}, + initial: "running", + states: { + running: { invoke: { id: "worker", src: "worker", onDone: { target: "done" } } }, + done: { type: "final" }, + }, + }); + const report = await checkCrashRecovery(parent, { executors }); + expect(report.failures).toEqual([]); + expect(report.baseline.effects.map((effect) => effect.key)).toEqual([ + "actor:worker/tool", + "worker/summarize#1", + ]); + }); + + test("a grandchild's completion reaches its parent after a crash between them", async () => { + const inner = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { summarize: createTextLogic({ model: "m", prompt: "Sum." }) }, + }).createMachine({ + context: {}, + initial: "work", + states: { + work: { invoke: { id: "summarize", src: "summarize", onDone: { target: "done" } } }, + done: { type: "final", output: () => "inner done" }, + }, + }); + const middle = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { inner, tool: createAsyncLogic({ run: async () => 1 }) }, + }).createMachine({ + context: {}, + initial: "inner", + states: { + inner: { invoke: { id: "inner", src: "inner", onDone: { target: "tool" } } }, + tool: { invoke: { id: "tool", src: "tool", onDone: { target: "done" } } }, + done: { type: "final" }, + }, + }); + const root = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { middle }, + }).createMachine({ + context: {}, + initial: "running", + states: { + running: { invoke: { id: "middle", src: "middle", onDone: { target: "done" } } }, + done: { type: "final" }, + }, + }); + const report = await checkCrashRecovery(root, { executors }); + expect(report.failures).toEqual([]); + expect(report.baseline.effects.map((effect) => effect.key)).toEqual([ + "middle/inner/summarize#1", + "actor:middle/tool", + ]); + }); + + test("reports effects that re-run when the store loses an acknowledged write", async () => { + // Write-behind: an append is acknowledged before it is stored, so the + // last one dies with the process. + const writeBehind = (): AgentEventLogStore => { + const store = createInMemoryEventLogStore(); + let pending: Parameters[0] | undefined; + return { + read: (threadId, options) => store.read(threadId, options), + length: (threadId) => store.length(threadId), + fork: (input) => store.fork(input), + async append(input) { + if (pending && input.expectedIndex === pending.expectedIndex + pending.entries.length) { + await store.append(pending); + } + pending = input; + }, + }; + }; + const report = await checkCrashRecovery(createPipeline(), { + executors, + events: [{ type: "REVIEW", ok: true }], + store: writeBehind, + }); + expect(report.ok).toBe(false); + expect(report.failures.some((failure) => failure.problem === "reran")).toBe(true); + }); + + test("an executor that answers by call order gets its recorded answer after a crash", async () => { + let calls = 0; + const report = await checkCrashRecovery(createPipeline(), { + executors: { ...executors, generateText: async () => ({ result: `draft ${++calls}` }) }, + events: [{ type: "REVIEW", ok: true }], + }); + expect(report.failures).toEqual([]); + expect(calls).toBe(1); + }); +}); + +describe("usage across a crash", () => { + test("a re-executed call's usage is counted once", async () => { + const machine = setupAgent({ + schemas: createAgentSchemas({ context: z.object({ tokens: z.number() }), events: {} }), + actors: { ask: createTextLogic({ model: "m", prompt: "q" }) }, + }).createMachine({ + context: { tokens: 0 }, + initial: "asking", + on: { + [AGENT_USAGE_EVENT_TYPE]: ({ context, event }) => ({ + context: { tokens: context.tokens + (event.usage.totalTokens ?? 0) }, + }), + }, + states: { + asking: { invoke: { id: "ask", src: "ask", onDone: { target: "done" } } }, + done: { type: "final", output: ({ context }) => context }, + }, + }); + const report = await checkCrashRecovery(machine, { + executors: { generateText: async () => ({ result: "x", usage: { totalTokens: 100 } }) }, + }); + expect(report.failures).toEqual([]); + expect(report.baseline.output).toEqual({ tokens: 100 }); + }); +}); + +describe("child inputs across a crash", () => { + test("a parent's send to its child and the child's reply land with the event that caused them", async () => { + const child = setupAgent({ + schemas: createAgentSchemas({ + context: z.object({ notes: z.number() }), + events: { NOTE: z.object({}), REPORT: z.object({}) }, + }), + }).createMachine({ + context: { notes: 0 }, + on: { + NOTE: ({ context }) => ({ context: { notes: context.notes + 1 } }), + REPORT: ({ context, parent }, enq) => { + if (parent) enq.sendTo(parent, { type: "REPORTED", notes: context.notes }); + }, + }, + }); + const parent = setupAgent({ + schemas: createAgentSchemas({ + context: z.object({ notes: z.number() }), + events: { + PING: z.object({}), + DONE: z.object({}), + REPORTED: z.object({ notes: z.number() }), + }, + }), + actors: { child }, + }).createMachine({ + context: { notes: -1 }, + invoke: { id: "child", src: "child" }, + initial: "on", + states: { + on: { + on: { + PING: ({ children }, enq) => { + enq.sendTo(children.child, { type: "NOTE" }); + }, + DONE: ({ children }, enq) => { + enq.sendTo(children.child, { type: "REPORT" }); + }, + REPORTED: ({ event }) => ({ target: "done", context: { notes: event.notes } }), + }, + }, + done: { type: "final", output: ({ context }) => context }, + }, + }); + const report = await checkCrashRecovery(parent, { + events: [{ type: "PING" }, { type: "DONE" }], + }); + expect(report.baseline.output).toEqual({ notes: 1 }); + expect(report.failures).toEqual([]); + }); +}); diff --git a/src/crash-recovery.ts b/src/crash-recovery.ts new file mode 100644 index 00000000..0031c3f3 --- /dev/null +++ b/src/crash-recovery.ts @@ -0,0 +1,747 @@ +/** + * `checkCrashRecovery` — kills a durable run at every point it could die, + * resumes it from its log, and checks the resumed run against an + * uninterrupted one. + * @module + */ +import type { + AnyActorLogic, + AnyActorRef, + AnyStateMachine, + EventFromLogic, + EventObject, +} from "xstate"; +import { isActLogic } from "./act.js"; +import { isDecisionLogic } from "./decision.js"; +import { stableStringify } from "./equivalence.js"; +import { agentCallSite, canonicalEventJson, replay, type AgentLogEntry } from "./event-log.js"; +import { createInMemoryEventLogStore, type AgentEventLogStore } from "./event-log-store.js"; +import { + createAgentRuntime, + isStateMachineLogic, + runToQuiescence, + toJsonValue, + type AgentRunResult, + type AgentRuntimeOptions, +} from "./run-agent.js"; +import { isTextLogic, type AgentRequestExecutors } from "./text-logic.js"; + +/** Options for {@link checkCrashRecovery}. */ +export interface CheckCrashRecoveryOptions { + /** Machine input for the run. */ + input?: unknown; + /** + * Host events, delivered one per turn: each time the run settles `idle`, + * the next one resumes it. The run ends when it settles `done` or `error`, + * or `idle` with none left. + */ + events?: readonly EventFromLogic[]; + /** + * Model executors. Each call is recorded by its `callKey`. They run in the + * uninterrupted run only: after a crash, a call it made gets its recorded + * answer, so a fake may answer by call order. + */ + executors?: Partial; + /** Actor sources, as `createAgentRuntime`'s `actors`. */ + actors?: Record; + /** + * A fresh, empty store for one run: the baseline, or one crash and its + * resume. Default: an in-memory store. Pass your own adapter to check that + * an entry it acknowledged is read back after the crash. + */ + store?: () => AgentEventLogStore; + /** Called before every run (the baseline and each crash), to reset fakes the executors or actors keep. */ + reset?: () => void; + /** Other runtime options (`maxModelCalls`, `verification`, …), shared by every run. */ + runtime?: Omit< + AgentRuntimeOptions, + "executors" | "actors" | "store" | "threadId" | "signal" | "timers" | "abortWith" + >; + /** Milliseconds one run (the baseline, a crashed run, or its resume) may take. Default 5000. */ + timeout?: number; +} + +/** + * One effect the run started: a model call (keyed by its `callKey` without the + * execution id, which differs per run) or a plain actor (keyed by its invoke + * site, as `actor:`). + */ +export interface CrashEffect { + kind: "model" | "actor"; + key: string; +} + +/** Where a run was killed. */ +export type CrashPoint = + | { + /** After the store accepted `entries` entries: the next append never lands. */ + kind: "log"; + entries: number; + } + | { + /** While `effect` was in flight: it ran, but its result was never recorded. */ + kind: "effect"; + effect: CrashEffect; + /** The effect's position among the baseline's effects, from 0. */ + index: number; + }; + +/** What went wrong after a crash. */ +export type CrashRecoveryProblem = + /** An effect whose completion was durable ran again. */ + | "reran" + /** An effect the uninterrupted run made never ran. */ + | "skipped" + /** An effect ran that the uninterrupted run never made (a model call under a new key). */ + | "unexpected" + /** A model call re-executed under its key with a different request. */ + | "request-changed" + /** The resumed run ended with a different status, output, state, or context. */ + | "outcome" + /** The final log does not replay to the resumed run's state. */ + | "replay" + /** The crashed run or its resume never settled within `timeout`. */ + | "stuck" + /** The resume threw. */ + | "threw"; + +/** One problem at one crash point. */ +export interface CrashRecoveryFailure { + point: CrashPoint; + problem: CrashRecoveryProblem; + message: string; + /** The effect key, for effect problems. */ + key?: string; +} + +/** What {@link checkCrashRecovery} found. */ +export interface CrashRecoveryReport { + /** True when every crash point recovered. */ + ok: boolean; + /** The uninterrupted run. */ + baseline: { + status: AgentRunResult["status"]; + output?: unknown; + /** Entries in its log. */ + entries: number; + /** Effects it started, in start order. */ + effects: CrashEffect[]; + }; + /** Every point a run was killed at, in order. */ + points: CrashPoint[]; + failures: CrashRecoveryFailure[]; +} + +class CrashError extends Error { + constructor() { + super("checkCrashRecovery: simulated crash"); + this.name = "CrashError"; + } +} + +const DONE_ACTOR = "xstate.done.actor"; +const ERROR_ACTOR = "xstate.error.actor"; + +type RecordedAnswer = { result: unknown; ms: number }; + +type StartedEffect = CrashEffect & { site: string; completed: boolean; request?: string }; + +/** One run's effects and how it decides to crash. */ +interface RunWorld { + effects: StartedEffect[]; + /** Each model call's result and how long it took, by key. */ + answers: Map; + /** The uninterrupted run's answers: a call it made gets the same result again. */ + recorded?: ReadonlyMap; + /** Called as each effect starts; returns true when the run crashes there. */ + onEffect(effect: StartedEffect): boolean; + crashed: boolean; +} + +/** + * Runs `machine` once uninterrupted, then once per crash point: killed after + * each log append and while each effect is in flight, then resumed from the + * store's log by a fresh runtime. Each resumed run is checked against the + * uninterrupted one: + * + * - An effect whose completion was durable before the crash never runs again. + * Only an effect in flight at the crash may (execution is at-least-once). + * - Every effect of the uninterrupted run runs, and no other: a re-executed + * model call keeps its `callKey` and its request. + * - The run ends with the same status, output, state value, and context. + * - The resumed log replays to that state. + * + * Model calls the uninterrupted run made are answered from its recording. + * Plain actors run live, so they must give the same result however many times + * they run (pass `reset` for fakes that keep state). + */ +export async function checkCrashRecovery( + machine: TMachine, + options: CheckCrashRecoveryOptions = {}, +): Promise { + const timeout = options.timeout ?? 5000; + + // ─── Baseline ─── + options.reset?.(); + const baselineWorld = newWorld(() => false); + const newStore = options.store ?? createInMemoryEventLogStore; + const baselineStore = observedStore(newStore(), baselineWorld, () => false); + const baseline = await withTimeout( + runTurns(machine, options, baselineWorld, baselineStore.store, "baseline"), + timeout, + ); + if (baseline === TIMED_OUT) { + throw new Error(`checkCrashRecovery: the uninterrupted run did not settle within ${timeout}ms`); + } + const baselineEntries = await baselineStore.store.read(THREAD); + const baselineCounts = countBy(baselineWorld.effects); + const expected = outcomeOf(baseline); + + const points: CrashPoint[] = []; + // Before the first append, and after each one but the last. + for (const entries of new Set([0, ...baselineStore.appends.slice(0, -1)])) { + points.push({ kind: "log", entries }); + } + baselineWorld.effects.forEach((effect, index) => { + points.push({ kind: "effect", effect: { kind: effect.kind, key: effect.key }, index }); + }); + + const failures: CrashRecoveryFailure[] = []; + for (const point of points) { + failures.push( + ...(await crashAt(point, { + machine, + options, + timeout, + baselineCounts, + recorded: baselineWorld.answers, + expected, + })), + ); + } + + return { + ok: failures.length === 0, + baseline: { + status: baseline.status, + ...(baseline.status === "done" ? { output: baseline.output } : {}), + entries: baselineEntries.length, + effects: baselineWorld.effects.map(({ kind, key }) => ({ kind, key })), + }, + points, + failures, + }; +} + +const THREAD = "crash-recovery"; +const TIMED_OUT = Symbol("timed out"); + +interface CrashContext { + machine: TMachine; + options: CheckCrashRecoveryOptions; + timeout: number; + baselineCounts: Map; + recorded: ReadonlyMap; + expected: string; +} + +async function crashAt( + point: CrashPoint, + ctx: CrashContext, +): Promise { + const { machine, options, timeout } = ctx; + const fail = (problem: CrashRecoveryProblem, message: string, key?: string) => ({ + point, + problem, + message, + ...(key !== undefined ? { key } : {}), + }); + options.reset?.(); + const backing = (options.store ?? createInMemoryEventLogStore)(); + + // ─── The run that dies ─── + const controller = new AbortController(); + let started = 0; + const crashed = newWorld( + () => point.kind === "effect" && started++ === point.index, + ctx.recorded, + ); + const crash = () => { + if (crashed.crashed) return; + crashed.crashed = true; + controller.abort(new CrashError()); + }; + const onEffect = crashed.onEffect; + crashed.onEffect = (effect) => { + const dies = onEffect(effect); + // The effect has started; the process dies before its result is recorded. + if (dies) queueMicrotask(crash); + return dies; + }; + const dying = observedStore(backing, crashed, (accepted) => { + if (point.kind !== "log" || accepted < point.entries) return false; + crash(); + return true; + }); + if (point.kind === "log" && point.entries === 0) dying.freeze(); + const died = await withTimeout( + runTurns(machine, options, crashed, dying.store, "crash", controller.signal), + timeout, + ); + crash(); + if (died === TIMED_OUT) { + return [fail("stuck", `the run did not stop within ${timeout}ms after the crash`)]; + } + + // ─── The resume ─── + const resumed = newWorld(() => false, ctx.recorded); + let result: AgentRunResult | typeof TIMED_OUT; + try { + result = await withTimeout(runTurns(machine, options, resumed, backing, "resume"), timeout); + } catch (error) { + return [fail("threw", `the resume threw: ${String(error)}`)]; + } + if (result === TIMED_OUT) { + return [fail("stuck", `the resume did not settle within ${timeout}ms`)]; + } + + const failures: CrashRecoveryFailure[] = []; + + // Effects: every baseline effect ran, nothing else ran, and only what was + // in flight at the crash ran twice. + const total = countBy([...crashed.effects, ...resumed.effects]); + const inFlight = countBy(crashed.effects.filter((effect) => !effect.completed)); + for (const [key, count] of total) { + const expected = ctx.baselineCounts.get(key) ?? 0; + const allowed = expected + (inFlight.get(key) ?? 0); + if (expected === 0) { + failures.push( + fail("unexpected", `'${key}' ran ${count}x; the uninterrupted run never made it`, key), + ); + } else if (count > allowed) { + failures.push( + fail( + "reran", + `'${key}' ran ${count}x, at most ${allowed} expected: its completion was durable before the crash`, + key, + ), + ); + } + } + for (const [key, expected] of ctx.baselineCounts) { + const count = total.get(key) ?? 0; + if (count < expected) { + failures.push(fail("skipped", `'${key}' ran ${count}x, ${expected} expected`, key)); + } + } + const crashedRequests = new Map( + crashed.effects + .filter((effect) => effect.kind === "model") + .map((effect) => [effect.key, effect.request]), + ); + for (const effect of resumed.effects) { + if (effect.kind !== "model" || !crashedRequests.has(effect.key)) continue; + if (crashedRequests.get(effect.key) !== effect.request) { + failures.push( + fail("request-changed", `'${effect.key}' re-executed with a different request`, effect.key), + ); + } + } + + const actual = outcomeOf(result); + if (actual !== ctx.expected) { + failures.push(fail("outcome", `resumed run ended ${actual}, expected ${ctx.expected}`)); + } + + const log = await backing.read(THREAD); + try { + const replayed = replay(machine, log); + const fromLog = stableStringify( + toJsonValue({ + value: (replayed.snapshot as { value?: unknown }).value, + context: (replayed.snapshot as { context?: unknown }).context, + }), + ); + const live = stableStringify( + toJsonValue({ + value: (result.snapshot as { value?: unknown }).value, + context: (result.snapshot as { context?: unknown }).context, + }), + ); + if (fromLog !== live) { + failures.push( + fail("replay", `the log replays to ${fromLog}, the resumed run ended at ${live}`), + ); + } + } catch (error) { + failures.push(fail("replay", `the log does not replay: ${String(error)}`)); + } + + return failures; +} + +/** + * Runs the turns: start from `input`, then one host event per idle settle. + * Resuming, the events already in the log are skipped. + */ +async function runTurns( + machine: TMachine, + options: CheckCrashRecoveryOptions, + world: RunWorld, + store: AgentEventLogStore, + phase: "baseline" | "crash" | "resume", + signal?: AbortSignal, +): Promise> { + const tracked = trackActors(machine, world); + const executors = trackExecutors(options.executors ?? {}, world); + const actors = + options.actors !== undefined + ? Object.fromEntries( + Object.entries(options.actors).map(([key, logic]) => [ + key, + isStateMachineLogic(logic) ? trackActors(logic, world) : trackLogic(logic, world), + ]), + ) + : undefined; + const turn = (event?: EventObject) => + runToQuiescence( + createAgentRuntime(tracked, { + ...options.runtime, + executors, + ...(actors !== undefined ? { actors } : {}), + store, + threadId: THREAD, + ...(signal !== undefined ? { signal } : {}), + }), + { + ...(options.input !== undefined ? { input: options.input as never } : {}), + ...(event !== undefined ? { event: event as never } : {}), + }, + ); + + const events = options.events ?? []; + let next = phase === "resume" ? delivered(await store.read(THREAD), events) : 0; + let result = await turn(); + while (result.status === "idle" && next < events.length && !world.crashed) { + result = await turn(events[next++]); + await result.drain?.(); + } + await result.drain?.(); + return result; +} + +/** How many of `events` the log already holds, matched in order on the root. */ +function delivered(log: readonly AgentLogEntry[], events: readonly EventObject[]): number { + let next = 0; + for (const entry of log) { + if (next >= events.length) break; + if (entry.actorPath !== undefined) continue; + if (canonicalEventJson(entry.event) === canonicalEventJson(events[next])) next++; + } + return next; +} + +function newWorld( + crashesAt: (effect: StartedEffect) => boolean, + recorded?: ReadonlyMap, +): RunWorld { + return { + effects: [], + answers: new Map(), + ...(recorded !== undefined ? { recorded } : {}), + crashed: false, + onEffect: (effect) => crashesAt(effect), + }; +} + +/** + * `store`, recording each append's resulting length and marking the effects + * whose completion an accepted entry carries. `afterAppend` returns true to + * freeze it: every later append rejects, as if the process died. + */ +function observedStore( + store: AgentEventLogStore, + world: RunWorld, + afterAppend: (accepted: number) => boolean, +) { + const appends: number[] = []; + let frozen = false; + const observed: AgentEventLogStore = { + ...store, + read: (threadId, options) => store.read(threadId, options), + length: (threadId) => store.length(threadId), + fork: (input) => store.fork(input), + async append(input) { + if (frozen || world.crashed) throw new CrashError(); + await store.append(input); + for (const entry of input.entries) markCompleted(entry, world.effects); + const accepted = input.expectedIndex + input.entries.length; + appends.push(accepted); + if (afterAppend(accepted)) frozen = true; + }, + }; + return { + store: observed, + appends, + freeze: () => { + frozen = true; + }, + }; +} + +/** + * Marks the call an accepted entry completes: a decision's chosen event names + * its call (`site#n`); a completion completes the oldest open effect at its + * invoke site, with the other attempts of the same model call. + */ +function markCompleted(entry: AgentLogEntry, effects: StartedEffect[]): void { + if (entry.decidedBy !== undefined) { + const call = entry.decidedBy; + for (const effect of effects) { + if (effect.key === call || effect.key.startsWith(`${call}.`)) effect.completed = true; + } + return; + } + if (entry.event.type !== DONE_ACTOR && entry.event.type !== ERROR_ACTOR) return; + const actorId = (entry.event as { actorId?: unknown }).actorId; + if (typeof actorId !== "string") return; + const site = agentCallSite(entry.actorPath ?? [], actorId); + const oldest = effects.find((effect) => effect.site === site && !effect.completed); + if (oldest === undefined) return; + const call = oldest.kind === "model" ? callOfModelKey(oldest.key) : undefined; + for (const effect of effects) { + if ( + effect === oldest || + (call !== undefined && effect.site === site && callOfModelKey(effect.key) === call) + ) { + effect.completed = true; + } + } +} + +function countBy(effects: readonly CrashEffect[]): Map { + const counts = new Map(); + for (const { key } of effects) counts.set(key, (counts.get(key) ?? 0) + 1); + return counts; +} + +function outcomeOf(result: AgentRunResult): string { + const snapshot = result.snapshot as { value?: unknown; context?: unknown }; + return stableStringify( + toJsonValue({ + status: result.status, + ...(result.status === "done" ? { output: result.output } : {}), + ...(result.status === "error" ? { cause: result.cause } : {}), + value: snapshot.value, + context: snapshot.context, + }), + ); +} + +async function withTimeout(promise: Promise, ms: number): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((resolve) => { + timer = setTimeout(() => resolve(TIMED_OUT), ms); + }), + ]); + } finally { + clearTimeout(timer); + } +} + +// ─── Effect tracking ─── + +/** `callKey` without its execution id: the same call has the same key in every run. */ +function modelKey(callKey: string): string { + return callKey.slice(callKey.indexOf(":") + 1); +} + +function siteOfModelKey(key: string): string { + const hash = key.lastIndexOf("#"); + return hash === -1 ? key : key.slice(0, hash); +} + +/** `site#n` of a model key: the call, without its attempt or strategy suffix. */ +function callOfModelKey(key: string): string { + const hash = key.lastIndexOf("#"); + if (hash === -1) return key; + const dot = key.indexOf(".", hash); + return dot === -1 ? key : key.slice(0, dot); +} + +function trackExecutors( + executors: Partial, + world: RunWorld, +): Partial { + const tracked: Record = {}; + for (const [slot, executor] of Object.entries(executors)) { + if (typeof executor !== "function") continue; + tracked[slot] = async (request: unknown, info?: { callKey?: string; signal?: AbortSignal }) => { + if (world.crashed) return never(info?.signal); + const key = info?.callKey !== undefined ? modelKey(info.callKey) : `${slot}:?`; + const effect: StartedEffect = { + kind: "model", + key, + site: siteOfModelKey(key), + completed: false, + request: requestFingerprint(request), + }; + world.effects.push(effect); + const dies = world.onEffect(effect); + const recorded = world.recorded?.get(key); + let result: unknown; + if (recorded !== undefined) { + // As long as it took the first time, so concurrent calls settle in + // the same order. + await new Promise((resolve) => setTimeout(resolve, recorded.ms)); + result = recorded.result; + } else { + const started = Date.now(); + result = await (executor as (...args: unknown[]) => unknown)(request, info); + if (info?.callKey !== undefined) { + world.answers.set(key, { result, ms: Date.now() - started }); + } + } + // Died while the call was out: it ran, but its result never comes back. + if (dies || world.crashed) return never(info?.signal); + return result; + }; + } + return tracked as Partial; +} + +// Fields that name the run, not the request. +const VOLATILE_REQUEST_KEYS = new Set(["runId"]); + +function requestFingerprint(request: unknown): string | undefined { + try { + const json = toJsonValue(request); + if (json && typeof json === "object" && !Array.isArray(json)) { + for (const key of VOLATILE_REQUEST_KEYS) delete (json as Record)[key]; + } + return stableStringify(json); + } catch { + return undefined; + } +} + +function never(signal: AbortSignal | undefined): Promise { + return new Promise((_, reject) => { + if (signal?.aborted) reject(signal.reason); + signal?.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); +} + +/** `machine` with every plain actor source, at any depth, tracked. */ +function trackActors( + machine: TMachine, + world: RunWorld, + visited = new Set(), +): TMachine { + if (visited.has(machine)) return machine; + visited.add(machine); + const sources = (machine.sources?.actors ?? {}) as Record; + const wrapped: Record = {}; + for (const [key, logic] of Object.entries(sources)) { + if (isTextLogic(logic) || isDecisionLogic(logic) || isActLogic(logic)) continue; + if (isStateMachineLogic(logic)) { + const inner = trackActors(logic, world, visited); + if (inner !== logic) wrapped[key] = inner; + continue; + } + if (typeof logic?.transition === "function") wrapped[key] = trackLogic(logic, world); + } + return Object.keys(wrapped).length > 0 + ? (machine.provide({ actors: wrapped as never }) as TMachine) + : machine; +} + +type ActorSelf = AnyActorRef & { id?: string; _parent?: ActorSelf }; + +/** The invoke site of an actor: its path's last id, under its parents' path. */ +function actorSite(self: ActorSelf): string { + const path: string[] = []; + for (let at: ActorSelf | undefined = self; at?._parent !== undefined; at = at._parent) { + path.unshift(String(at.id)); + } + return agentCallSite(path.slice(0, -1), path.at(-1) ?? ""); +} + +/** + * `logic`, recording each effect it starts (`@xstate.effect`: an async + * actor's run, a callback's body). After the crash, effects do not start. + */ +function trackLogic(logic: AnyActorLogic, world: RunWorld): AnyActorLogic { + if (isTextLogic(logic) || isDecisionLogic(logic) || isActLogic(logic)) return logic; + type Result = [unknown, unknown[]]; + type Effect = { type?: string; exec: (...args: unknown[]) => unknown }; + const track = (result: Result, self: ActorSelf): Result => { + const [snapshot, effects] = result; + if (!Array.isArray(effects) || !effects.some((e) => (e as Effect)?.type === "@xstate.effect")) { + return result; + } + return [ + snapshot, + effects.map((effect) => { + if ((effect as Effect)?.type !== "@xstate.effect") return effect; + const exec = (effect as Effect).exec; + const tracked = (...args: unknown[]) => { + if (world.crashed) return; + const site = actorSite(self); + const started: StartedEffect = { + kind: "actor", + key: `actor:${site}`, + site, + completed: false, + }; + world.effects.push(started); + world.onEffect(started); + return exec(...args); + }; + return { ...(effect as object), exec: tracked, action: tracked }; + }), + ]; + }; + const overrides: PropertyDescriptorMap = { + transition: { + value: (...args: Parameters) => + track(logic.transition(...args) as unknown as Result, args[2].self as ActorSelf), + }, + }; + if (typeof logic.initialTransition === "function") { + overrides["initialTransition"] = { + value: (...args: Parameters>) => + track(logic.initialTransition!(...args) as unknown as Result, args[1]!.self as ActorSelf), + }; + } + if (typeof logic.start === "function") { + // A restored actor restarts its effects from `start`, through its own + // `transition`, not this one: track what it hands the executor. + overrides["start"] = { + value: ( + snapshot: unknown, + scope: { actionExecutor: (effect: unknown) => void; self: AnyActorRef }, + startOptions: unknown, + ) => + logic.start!( + snapshot as never, + new Proxy(scope, { + get(target, property) { + if (property === "actionExecutor") { + return (effect: unknown) => { + const [, [tracked]] = track([undefined, [effect]], target.self as ActorSelf); + target.actionExecutor(tracked); + }; + } + return Reflect.get(target, property, target); + }, + }) as never, + startOptions as never, + ), + }; + } + return Object.create(logic, overrides) as AnyActorLogic; +} diff --git a/src/durable-child-state.test.ts b/src/durable-child-state.test.ts index 27b62ebd..36ec5b86 100644 --- a/src/durable-child-state.test.ts +++ b/src/durable-child-state.test.ts @@ -527,3 +527,86 @@ describe("a plain actor inside a child machine", () => { expect(seenAtB[0]).toContain("xstate.done.actor@child"); }); }); + +describe("a child that finished just before a crash", () => { + // root → middle → inner; the crash cuts the log after a child's last entry, + // before its parent journaled the child's completion. + function createNested() { + const inner = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { summarize: createTextLogic({ model: "test-model", prompt: "Sum." }) }, + }).createMachine({ + context: {}, + initial: "work", + states: { + work: { invoke: { id: "summarize", src: "summarize", onDone: { target: "done" } } }, + done: { type: "final", output: () => "inner done" }, + }, + }); + const middle = setupAgent({ + schemas: createAgentSchemas({ context: z.object({ got: z.unknown() }), events: {} }), + actors: { inner }, + }).createMachine({ + context: { got: undefined }, + initial: "inner", + states: { + inner: { + invoke: { + id: "inner", + src: "inner", + onDone: ({ event }) => ({ target: "done", context: { got: event.output } }), + }, + }, + done: { type: "final", output: ({ context }) => context }, + }, + }); + return setupAgent({ + schemas: createAgentSchemas({ context: z.object({ got: z.unknown() }), events: {} }), + actors: { middle }, + }).createMachine({ + context: { got: undefined }, + initial: "running", + states: { + running: { + invoke: { + id: "middle", + src: "middle", + onDone: ({ event }) => ({ target: "done", context: { got: event.output } }), + }, + }, + done: { type: "final", output: ({ context }) => context }, + }, + }); + } + + test.each([ + ["the root", "middle"], + ["a child machine", "inner"], + ])("%s still gets its child's completion on resume", async (_, childId) => { + const machine = createNested(); + const generateText = async () => ({ result: "sum" }); + const full = await runToQuiescence( + createAgentRuntime(machine, { executors: { generateText } }), + {}, + ); + expect(full.status).toBe("done"); + const cut = full.events.findIndex( + (entry) => + entry.event.type === "xstate.done.actor" && + (entry.event as { actorId?: string }).actorId === childId, + ); + expect(cut).toBeGreaterThan(0); + const calls: unknown[] = []; + const resumed = await runToQuiescence( + createAgentRuntime(machine, { + executors: { generateText: async () => (calls.push(1), { result: "sum" }) }, + }), + { events: full.events.slice(0, cut) }, + ); + expect(resumed.status).toBe("done"); + expect(resumed.status === "done" && resumed.output).toEqual( + full.status === "done" && full.output, + ); + expect(calls).toEqual([]); + }); +}); diff --git a/src/run-agent.ts b/src/run-agent.ts index f0a9d640..fec25fed 100644 --- a/src/run-agent.ts +++ b/src/run-agent.ts @@ -254,8 +254,11 @@ export type AgentTraceEvent * event is dropped. */ type: "usage.dropped"; event: AgentUsageEvent; - /** `'settled'`: the call settled after the run's cycle had resolved. */ - reason: "settled"; + /** + * `'settled'`: the call settled after the run's cycle had resolved. + * `'repeat'`: a call re-executed after a crash; the log already holds its usage. + */ + reason: "settled" | "repeat"; } | ( | { @@ -1383,6 +1386,7 @@ async function runTextRequest( ...(src !== "" ? { src } : {}), model: request.model, ...(request.name !== undefined ? { name: request.name } : {}), + ...(callKey !== undefined ? { callKey } : {}), }, self, ); @@ -1598,6 +1602,7 @@ function createCountingDecide( ...(attemptRequest.id ? { id: attemptRequest.id } : {}), ...(src !== "" ? { src } : {}), model: attemptRequest.model, + ...(callKey !== undefined ? { callKey } : {}), }, self, ), @@ -1662,6 +1667,7 @@ function createCountingFill( ...(id !== "" ? { id } : {}), ...(src !== "" ? { src } : {}), model: request.model, + ...(info?.callKey !== undefined ? { callKey: info.callKey } : {}), }, self, ), @@ -2984,8 +2990,9 @@ export function createAgentRuntime( // ─── Write-ahead journaling (`options.store`) ─── // - // Every appended entry is queued onto ONE chain, so writes reach the store in - // log order and each carries its own index as `expectedIndex`. The chain is + // Entries are written in batches on ONE chain, so writes reach the store in + // log order and each batch starts at its first entry's index as + // `expectedIndex`. The chain is // awaited at two points only: before a model call (`awaitJournal`, the // barrier the executors call) and at finish. Pure transitions never wait. // The first rejection wins, stops the run (`cause: 'journal'`) and blocks @@ -3002,10 +3009,20 @@ export function createAgentRuntime( let journalPending = 0; let journalError: unknown; - const journalEntry = (entry: AgentLogEntry): void => { - if (store === undefined) { + // Entries not yet handed to the store. They are written together, in one + // append, whenever the loop waits (for an effect, or for nothing) and before + // a barrier, so the entries one input causes land atomically: a root event + // and the child inputs its transition sends, a child input and the event + // it sends back. A crash between them would otherwise keep the cause and + // lose the effect, which a fold cannot re-derive. + let unflushed: AgentLogEntry[] = []; + + const flushJournal = (): void => { + if (store === undefined || unflushed.length === 0) { return; } + const entries = unflushed; + unflushed = []; journalPending++; journalChain = journalChain .then(async () => { @@ -3014,8 +3031,8 @@ export function createAgentRuntime( } await store.append({ threadId: threadId!, - expectedIndex: entry.index, - entries: [entry], + expectedIndex: entries[0]!.index, + entries, }); }) .catch((error: unknown) => { @@ -3027,6 +3044,15 @@ export function createAgentRuntime( }); }; + const journalEntry = (entry: AgentLogEntry): void => { + if (store === undefined) { + return; + } + unflushed.push(entry); + // A straggler after the run settled has no loop left to flush it. + if (finished) flushJournal(); + }; + /** * Every journal write ISSUED SO FAR, including the stragglers appended after * the run settled. Re-reads the chain tail until it stops moving, so a write @@ -3037,6 +3063,7 @@ export function createAgentRuntime( if (store === undefined) { return; } + flushJournal(); let tail = journalChain; for (;;) { await tail; @@ -3052,6 +3079,7 @@ export function createAgentRuntime( if (store === undefined) { return; } + flushJournal(); while (journalPending > 0) { await journalChain; } @@ -3203,6 +3231,19 @@ export function createAgentRuntime( usage, budget: { modelCalls: modelCallsSpent + modelCallCount, maxModelCalls }, }; + // At-least-once execution, exactly-once accounting: the log already + // holds this call's spend from before the crash. + if ( + source.callKey !== undefined && + logEntries.some( + (entry) => + entry.event.type === AGENT_USAGE_EVENT_TYPE && + (entry.event as AgentUsageEvent).callKey === source.callKey, + ) + ) { + onTrace({ type: "usage.dropped", event, reason: "repeat" }); + return; + } let delivered = false; deliverUsageEvent( event, @@ -3332,8 +3373,10 @@ export function createAgentRuntime( // happen, but a child runs live, and a fold reproduces its state only from // its own inputs. Stopping a child is the root's doing, which the root's // own entries already fold. - pendingJournal: () => - store !== undefined && journalPending > 0 ? awaitJournal().catch(() => {}) : undefined, + pendingJournal: () => { + flushJournal(); + return store !== undefined && journalPending > 0 ? awaitJournal().catch(() => {}) : undefined; + }, onChildEvent: (self, event) => { if (finished || event.type === "xstate.stop") return; const actorPath = actorPathOf(self); @@ -3593,6 +3636,66 @@ export function createAgentRuntime( } }); + // A child restored already finished never tells its parent: restoring does + // not replay the transition that finished it. That happens when the run + // died after the child's last entry was journaled but before its parent's + // completion was, so the parent still waits on it. Deliver the completion + // the child would have sent, at any depth. + type RestoredChild = AnyActorRef & { + id: string; + _restored?: boolean; + _parent?: AnyActorRef; + sessionId?: string; + system?: { sendEvent?: (source: object, target: object, event: object) => void }; + }; + type ChildSnapshot = { + status?: string; + output?: unknown; + error?: unknown; + children?: Record; + }; + const completionOf = (child: RestoredChild): EventObject | undefined => { + const snapshot = child.getSnapshot() as ChildSnapshot; + if (snapshot.status === "done") { + return { + type: "xstate.done.actor", + output: snapshot.output, + actorId: child.id, + sessionId: child.sessionId, + } as EventObject; + } + if (snapshot.status === "error") { + return { + type: "xstate.error.actor", + error: snapshot.error, + actorId: child.id, + sessionId: child.sessionId, + } as EventObject; + } + return undefined; + }; + // The finished children of a restored root: their completions go to the mailbox. + const restoredRootCompletions = (snapshot: AnyMachineSnapshot): EventObject[] => + Object.values((snapshot.children ?? {}) as Record) + .map((child) => (child ? completionOf(child) : undefined)) + .filter((event): event is EventObject => event !== undefined); + // The restored, finished children inside a child machine: their completions + // go to that machine, which journals them as its own inputs. A child that + // finished live has already told its parent. + const reportRestoredCompletions = (actor: AnyActorRef): void => { + const snapshot = actor.getSnapshot() as ChildSnapshot; + if (snapshot.status !== "active") return; + for (const child of Object.values(snapshot.children ?? {}) as (RestoredChild | undefined)[]) { + if (!child) continue; + const event = completionOf(child); + if (event !== undefined && child._restored) { + child.system?.sendEvent?.(child, actor, event); + } else { + reportRestoredCompletions(child); + } + } + }; + let execution!: DurableExecution; const durableAdapter = { startActor: (actor: AnyActorRef) => { @@ -3601,6 +3704,7 @@ export function createAgentRuntime( // reporting a child error the machine already handles as unhandled. actor.subscribe({ next: recheck, complete: recheck, error: recheck }); (actor as unknown as { start(): void }).start(); + reportRestoredCompletions(actor); }, enqueueRootEvent: (_source: unknown, event: EventObject) => enqueue(event), executeAction: ( @@ -3613,6 +3717,8 @@ export function createAgentRuntime( if (queued !== undefined) { return queued; } + // Nothing queued: the entries so far are one settled step. Write them. + flushJournal(); if (stopReason !== undefined || finished || !workInFlight()) { return QUIESCENT as unknown as EventObject; } @@ -4046,6 +4152,7 @@ export function createAgentRuntime( AnyMachineSnapshot, AgentEffects, ]; + for (const event of restoredRootCompletions(snapshot)) enqueue(event); } catch (error) { // A snapshot this machine cannot restore fails the run on its own // terms rather than throwing out of the host. @@ -4283,6 +4390,7 @@ export function createAgentRuntime( }; finished = true; + flushJournal(); releaseAllDeliveries(); // Every write queued so far: the result waits for these. A straggler // entry appended after this point is still written, but not awaited. diff --git a/src/serialize-trace-event.test.ts b/src/serialize-trace-event.test.ts index d0cf0ce7..a16ad46f 100644 --- a/src/serialize-trace-event.test.ts +++ b/src/serialize-trace-event.test.ts @@ -42,7 +42,7 @@ type HandWrittenJsonSerializableTraceEvent = { | { type: "stream.chunk"; request: JsonValue; chunk: string } | { type: "machine.transition"; from: JsonValue; snapshot: JsonValue; event: JsonValue } | { type: "emit"; event: JsonValue } - | { type: "usage.dropped"; event: JsonValue; reason: "settled" } + | { type: "usage.dropped"; event: JsonValue; reason: "settled" | "repeat" } | { type: "run.end"; status: "done"; output: JsonValue; snapshot: JsonValue } | { type: "run.end"; diff --git a/src/testing/index.ts b/src/testing/index.ts index d130dcaa..ac62ffb5 100644 --- a/src/testing/index.ts +++ b/src/testing/index.ts @@ -32,6 +32,15 @@ export type { SimulationScript, SimulationTrailEntry, } from "../verify.js"; +export { checkCrashRecovery } from "../crash-recovery.js"; +export type { + CheckCrashRecoveryOptions, + CrashEffect, + CrashPoint, + CrashRecoveryFailure, + CrashRecoveryProblem, + CrashRecoveryReport, +} from "../crash-recovery.js"; export { matchesTrajectory } from "../trajectory.js"; export type { MatchTrajectoryOptions, diff --git a/src/usage.ts b/src/usage.ts index e5ec4a5b..8a22629b 100644 --- a/src/usage.ts +++ b/src/usage.ts @@ -13,6 +13,11 @@ export interface AgentUsageEvent extends EventObject { src?: string; model?: string; name?: string; + /** + * The call's `info.callKey`, when the run has an event log. A call re-executed + * after a crash, whose usage the log already holds, is not reported again. + */ + callKey?: string; /** * The run's model-call budget after this call: `modelCalls` counted against * `maxModelCalls` so far, including `modelCallsSpent`. Guard on diff --git a/src/verify-round9.test.ts b/src/verify-round9.test.ts index b9f06881..70a2efd2 100644 --- a/src/verify-round9.test.ts +++ b/src/verify-round9.test.ts @@ -222,7 +222,7 @@ describe("explorePaths — usage", () => { ]); }); - test("the event matches the one a run delivers, bar its budget", async () => { + test("the event matches the one a run delivers, bar its budget and call key", async () => { const seen: unknown[] = []; const run = await runAgent( createBudgetMachine((event) => seen.push(event)), @@ -233,7 +233,7 @@ describe("explorePaths — usage", () => { outputs: { summarize: ["ok", "ok2"] }, usage: { summarize: [{ totalTokens: 10 }] }, }); - const { budget: _budget, ...live } = seen[0] as Record; + const { budget: _budget, callKey: _callKey, ...live } = seen[0] as Record; expect(report.terminals[0]!.path[0]).toEqual(live); }); From cd142a732ccbfe2ebb40466af4a661d4a8c16042 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 12:04:18 -0500 Subject: [PATCH 02/11] fix: review findings on crash recovery - finish writes the last batch before deciding the outcome, so a rejected final write settles cause: journal - harness: host events are tracked by entry id, not matched by shape - harness: replay uses the actors option's sources - harness: a child machine under two sources is tracked under both - harness: crash after the last append too --- src/crash-recovery.test.ts | 129 ++++++++++++++++++++++++++++++++++++- src/crash-recovery.ts | 69 +++++++++++++------- src/run-agent.ts | 3 + 3 files changed, 176 insertions(+), 25 deletions(-) diff --git a/src/crash-recovery.test.ts b/src/crash-recovery.test.ts index 86b76d8b..98a70ecd 100644 --- a/src/crash-recovery.test.ts +++ b/src/crash-recovery.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "vitest"; -import { createAsyncLogic } from "xstate"; +import { createAsyncLogic, createCallbackLogic } from "xstate"; import { z } from "zod"; import { checkCrashRecovery } from "./crash-recovery.js"; import { createInMemoryEventLogStore, type AgentEventLogStore } from "./event-log-store.js"; @@ -7,7 +7,9 @@ import { AGENT_USAGE_EVENT_TYPE, createAgentSchemas, createDecisionLogic, + createAgentRuntime, createTextLogic, + runToQuiescence, setupAgent, type AgentDecisionExecutor, } from "./index.js"; @@ -294,3 +296,128 @@ describe("child inputs across a crash", () => { expect(report.failures).toEqual([]); }); }); + +describe("the final write", () => { + test("a rejected last batch settles the run as a journal error", async () => { + const machine = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: { GO: z.object({}) } }), + }).createMachine({ + context: {}, + initial: "waiting", + states: { waiting: { on: { GO: { target: "done" } } }, done: { type: "final" } }, + }); + const inner = createInMemoryEventLogStore(); + let appends = 0; + const store: AgentEventLogStore = { + read: (threadId, options) => inner.read(threadId, options), + length: (threadId) => inner.length(threadId), + fork: (input) => inner.fork(input), + async append(input) { + if (appends++ > 0) throw new Error("disk full"); + await inner.append(input); + }, + }; + const first = await runToQuiescence(createAgentRuntime(machine, { store, threadId: "t" }), {}); + expect(first.status).toBe("idle"); + const second = await runToQuiescence(createAgentRuntime(machine, { store, threadId: "t" }), { + event: { type: "GO" }, + }); + expect(second.status === "error" && second.cause).toBe("journal"); + }); +}); + +describe("harness coverage", () => { + test("a child machine under two sources is tracked under both", async () => { + const worker = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { tool: createAsyncLogic({ run: async () => 1 }) }, + }).createMachine({ + context: {}, + initial: "tool", + states: { + tool: { invoke: { id: "tool", src: "tool", onDone: { target: "done" } } }, + done: { type: "final" }, + }, + }); + const parent = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { first: worker, second: worker }, + }).createMachine({ + context: {}, + initial: "a", + states: { + a: { invoke: { id: "a", src: "first", onDone: { target: "b" } } }, + b: { invoke: { id: "b", src: "second", onDone: { target: "done" } } }, + done: { type: "final" }, + }, + }); + const report = await checkCrashRecovery(parent, {}); + expect(report.failures).toEqual([]); + expect(report.baseline.effects.map((effect) => effect.key)).toEqual([ + "actor:a/tool", + "actor:b/tool", + ]); + }); + + test("a child machine supplied through `actors` replays", async () => { + const worker = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { tool: createAsyncLogic({ run: async () => 1 }) }, + }).createMachine({ + context: {}, + initial: "tool", + states: { + tool: { invoke: { id: "tool", src: "tool", onDone: { target: "done" } } }, + done: { type: "final" }, + }, + }); + const parent = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { worker: createAsyncLogic({ run: async () => 0 }) }, + }).createMachine({ + context: {}, + initial: "running", + states: { + running: { invoke: { id: "worker", src: "worker", onDone: { target: "done" } } }, + done: { type: "final" }, + }, + }); + const report = await checkCrashRecovery(parent, { actors: { worker } }); + expect(report.failures).toEqual([]); + expect(report.baseline.effects.map((effect) => effect.key)).toEqual(["actor:worker/tool"]); + }); + + test("a host event is not mistaken for an internal event of the same shape", async () => { + const parent = setupAgent({ + schemas: createAgentSchemas({ + context: z.object({ pings: z.number() }), + events: { PING: z.object({}) }, + }), + actors: { + pinger: createCallbackLogic(({ sendBack }) => { + sendBack({ type: "PING" }); + }), + }, + }).createMachine({ + context: { pings: 0 }, + initial: "first", + states: { + first: { + invoke: { src: "pinger" }, + on: { + PING: ({ context }) => ({ target: "second", context: { pings: context.pings + 1 } }), + }, + }, + second: { + on: { + PING: ({ context }) => ({ target: "done", context: { pings: context.pings + 1 } }), + }, + }, + done: { type: "final", output: ({ context }) => context }, + }, + }); + const report = await checkCrashRecovery(parent, { events: [{ type: "PING" }] }); + expect(report.baseline.output).toEqual({ pings: 2 }); + expect(report.failures).toEqual([]); + }); +}); diff --git a/src/crash-recovery.ts b/src/crash-recovery.ts index 0031c3f3..49c0714b 100644 --- a/src/crash-recovery.ts +++ b/src/crash-recovery.ts @@ -149,6 +149,8 @@ type StartedEffect = CrashEffect & { site: string; completed: boolean; request?: /** One run's effects and how it decides to crash. */ interface RunWorld { effects: StartedEffect[]; + /** The log entry ids of the host events this run delivered, in order. */ + hostEntries: string[]; /** Each model call's result and how long it took, by key. */ answers: Map; /** The uninterrupted run's answers: a call it made gets the same result again. */ @@ -187,7 +189,7 @@ export async function checkCrashRecovery( const newStore = options.store ?? createInMemoryEventLogStore; const baselineStore = observedStore(newStore(), baselineWorld, () => false); const baseline = await withTimeout( - runTurns(machine, options, baselineWorld, baselineStore.store, "baseline"), + runTurns(machine, options, baselineWorld, baselineStore.store), timeout, ); if (baseline === TIMED_OUT) { @@ -198,8 +200,8 @@ export async function checkCrashRecovery( const expected = outcomeOf(baseline); const points: CrashPoint[] = []; - // Before the first append, and after each one but the last. - for (const entries of new Set([0, ...baselineStore.appends.slice(0, -1)])) { + // Before the first append, and after each one. + for (const entries of new Set([0, ...baselineStore.appends])) { points.push({ kind: "log", entries }); } baselineWorld.effects.forEach((effect, index) => { @@ -285,7 +287,7 @@ async function crashAt( }); if (point.kind === "log" && point.entries === 0) dying.freeze(); const died = await withTimeout( - runTurns(machine, options, crashed, dying.store, "crash", controller.signal), + runTurns(machine, options, crashed, dying.store, undefined, controller.signal), timeout, ); crash(); @@ -297,7 +299,10 @@ async function crashAt( const resumed = newWorld(() => false, ctx.recorded); let result: AgentRunResult | typeof TIMED_OUT; try { - result = await withTimeout(runTurns(machine, options, resumed, backing, "resume"), timeout); + result = await withTimeout( + runTurns(machine, options, resumed, backing, crashed.hostEntries), + timeout, + ); } catch (error) { return [fail("threw", `the resume threw: ${String(error)}`)]; } @@ -355,7 +360,12 @@ async function crashAt( const log = await backing.read(THREAD); try { - const replayed = replay(machine, log); + const replayed = replay( + options.actors !== undefined + ? (machine.provide({ actors: options.actors as never }) as TMachine) + : machine, + log, + ); const fromLog = stableStringify( toJsonValue({ value: (replayed.snapshot as { value?: unknown }).value, @@ -389,7 +399,8 @@ async function runTurns( options: CheckCrashRecoveryOptions, world: RunWorld, store: AgentEventLogStore, - phase: "baseline" | "crash" | "resume", + /** Resuming: the host events the crashed run delivered, by entry id. */ + deliveredBefore?: readonly string[], signal?: AbortSignal, ): Promise> { const tracked = trackActors(machine, world); @@ -403,10 +414,23 @@ async function runTurns( ]), ) : undefined; - const turn = (event?: EventObject) => - runToQuiescence( + const turn = (event?: EventObject) => { + // The host's event is the first root entry like it after the run opens. + let awaiting = event !== undefined ? canonicalEventJson(event) : undefined; + return runToQuiescence( createAgentRuntime(tracked, { ...options.runtime, + onEvent: (entry) => { + options.runtime?.onEvent?.(entry); + if ( + awaiting !== undefined && + entry.actorPath === undefined && + canonicalEventJson(entry.event) === awaiting + ) { + world.hostEntries.push(entry.id); + awaiting = undefined; + } + }, executors, ...(actors !== undefined ? { actors } : {}), store, @@ -418,9 +442,14 @@ async function runTurns( ...(event !== undefined ? { event: event as never } : {}), }, ); + }; const events = options.events ?? []; - let next = phase === "resume" ? delivered(await store.read(THREAD), events) : 0; + let next = 0; + if (deliveredBefore !== undefined) { + const durable = new Set((await store.read(THREAD)).map((entry) => entry.id)); + while (next < deliveredBefore.length && durable.has(deliveredBefore[next]!)) next++; + } let result = await turn(); while (result.status === "idle" && next < events.length && !world.crashed) { result = await turn(events[next++]); @@ -430,23 +459,13 @@ async function runTurns( return result; } -/** How many of `events` the log already holds, matched in order on the root. */ -function delivered(log: readonly AgentLogEntry[], events: readonly EventObject[]): number { - let next = 0; - for (const entry of log) { - if (next >= events.length) break; - if (entry.actorPath !== undefined) continue; - if (canonicalEventJson(entry.event) === canonicalEventJson(events[next])) next++; - } - return next; -} - function newWorld( crashesAt: (effect: StartedEffect) => boolean, recorded?: ReadonlyMap, ): RunWorld { return { effects: [], + hostEntries: [], answers: new Map(), ...(recorded !== undefined ? { recorded } : {}), crashed: false, @@ -639,10 +658,12 @@ function never(signal: AbortSignal | undefined): Promise { function trackActors( machine: TMachine, world: RunWorld, - visited = new Set(), + // The machines above this one: a machine that invokes itself stops here, + // while the same machine under two sources is tracked under both. + ancestors: ReadonlySet = new Set(), ): TMachine { - if (visited.has(machine)) return machine; - visited.add(machine); + if (ancestors.has(machine)) return machine; + const visited = new Set([...ancestors, machine]); const sources = (machine.sources?.actors ?? {}) as Record; const wrapped: Record = {}; for (const [key, logic] of Object.entries(sources)) { diff --git a/src/run-agent.ts b/src/run-agent.ts index fec25fed..2df61bb6 100644 --- a/src/run-agent.ts +++ b/src/run-agent.ts @@ -4315,6 +4315,9 @@ export function createAgentRuntime( appendLogEntry(event); } } + // The last step's entries are written before the outcome is decided, + // so a write that fails here settles the run `cause: "journal"`. + await drainJournal(); const snapshot = current as AnyMachineSnapshot; let outcome: AgentRunOutcome; if (stopReason !== undefined) { From 7a04c1ab4e2f919cacca923dc26366279a5e33b5 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 12:09:57 -0500 Subject: [PATCH 03/11] fix: a model-call barrier waits for the step to settle; harness reports crashed-run throws - awaitJournal no longer flushes mid-step, so a child's input and the reply it sends back stay in one batch when the same transition starts a model call (next-task fallback for hosts not reading nextEvent) - checkCrashRecovery reports a throw from the crashed run or the store factory as threw at that point and keeps checking --- docs/persistence.md | 2 +- src/crash-recovery.test.ts | 66 ++++++++++++++++++++++++++++++++++++++ src/crash-recovery.ts | 26 +++++++++++---- src/run-agent.ts | 36 ++++++++++++++++++--- 4 files changed, 118 insertions(+), 12 deletions(-) diff --git a/docs/persistence.md b/docs/persistence.md index 87c5d877..e4f50ae9 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -124,7 +124,7 @@ Each resumed run is checked against the uninterrupted one. Each failure names it | `outcome` | The resume ended with a different status, output, state, or context | | `replay` | The final log does not replay to the resumed state | | `stuck` | The crashed run or its resume did not settle within `timeout` (default 5000 ms) | -| `threw` | The resume threw | +| `threw` | The crashed run, its resume, or the `store` factory threw | - An effect in flight at the crash may run twice. That is the at-least-once contract (see [Idempotency keys](hosts.md#idempotency-keys)), and it is not reported. - Model calls are keyed by `callKey` without the execution id, and plain actors by invoke site. Only actors bound through string-keyed sources are seen. diff --git a/src/crash-recovery.test.ts b/src/crash-recovery.test.ts index 98a70ecd..657aecb8 100644 --- a/src/crash-recovery.test.ts +++ b/src/crash-recovery.test.ts @@ -421,3 +421,69 @@ describe("harness coverage", () => { expect(report.failures).toEqual([]); }); }); + +describe("child replies across a model call", () => { + test("a child's reply lands with its input when the same transition starts a model call", async () => { + const child = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: { REPORT: z.object({}) } }), + }).createMachine({ + context: {}, + on: { + REPORT: ({ parent }, enq) => { + if (parent) enq.sendTo(parent, { type: "REPORTED" }); + }, + }, + }); + const parent = setupAgent({ + schemas: createAgentSchemas({ + context: z.object({ reported: z.boolean() }), + events: { GO: z.object({}), REPORTED: z.object({}) }, + }), + actors: { child, draft: createTextLogic({ model: "m", prompt: "Draft." }) }, + }).createMachine({ + context: { reported: false }, + invoke: { id: "child", src: "child" }, + initial: "waiting", + on: { REPORTED: ({ context }) => ({ context: { ...context, reported: true } }) }, + states: { + waiting: { + on: { + GO: ({ children }, enq) => { + enq.sendTo(children.child, { type: "REPORT" }); + return { target: "drafting" }; + }, + }, + }, + drafting: { invoke: { id: "draft", src: "draft", onDone: { target: "done" } } }, + done: { type: "final", output: ({ context }) => context }, + }, + }); + const report = await checkCrashRecovery(parent, { executors, events: [{ type: "GO" }] }); + expect(report.baseline.output).toEqual({ reported: true }); + expect(report.failures).toEqual([]); + }); +}); + +describe("a store that fails", () => { + test("a read that rejects is reported per crash point", async () => { + let stores = 0; + const report = await checkCrashRecovery(createPipeline(), { + executors, + events: [{ type: "REVIEW", ok: true }], + store: () => { + const inner = createInMemoryEventLogStore(); + // The baseline's store works; every later one fails to read. + const broken = stores++ > 0; + return { + read: (threadId, options) => + broken ? Promise.reject(new Error("offline")) : inner.read(threadId, options), + length: (threadId) => inner.length(threadId), + fork: (input) => inner.fork(input), + append: (input) => inner.append(input), + }; + }, + }); + expect(report.failures.length).toBe(report.points.length); + expect(new Set(report.failures.map((failure) => failure.problem))).toEqual(new Set(["threw"])); + }); +}); diff --git a/src/crash-recovery.ts b/src/crash-recovery.ts index 49c0714b..728116d8 100644 --- a/src/crash-recovery.ts +++ b/src/crash-recovery.ts @@ -102,7 +102,7 @@ export type CrashRecoveryProblem = | "replay" /** The crashed run or its resume never settled within `timeout`. */ | "stuck" - /** The resume threw. */ + /** The crashed run, its resume, or the store factory threw. */ | "threw"; /** One problem at one crash point. */ @@ -259,7 +259,12 @@ async function crashAt( ...(key !== undefined ? { key } : {}), }); options.reset?.(); - const backing = (options.store ?? createInMemoryEventLogStore)(); + let backing: AgentEventLogStore; + try { + backing = (options.store ?? createInMemoryEventLogStore)(); + } catch (error) { + return [fail("threw", `the store factory threw: ${String(error)}`)]; + } // ─── The run that dies ─── const controller = new AbortController(); @@ -286,10 +291,19 @@ async function crashAt( return true; }); if (point.kind === "log" && point.entries === 0) dying.freeze(); - const died = await withTimeout( - runTurns(machine, options, crashed, dying.store, undefined, controller.signal), - timeout, - ); + let died: AgentRunResult | typeof TIMED_OUT | undefined; + try { + died = await withTimeout( + runTurns(machine, options, crashed, dying.store, undefined, controller.signal), + timeout, + ); + } catch (error) { + // Dying is the point; anything else thrown is a finding. + if (!(error instanceof CrashError)) { + crash(); + return [fail("threw", `the crashed run threw: ${String(error)}`)]; + } + } crash(); if (died === TIMED_OUT) { return [fail("stuck", `the run did not stop within ${timeout}ms after the crash`)]; diff --git a/src/run-agent.ts b/src/run-agent.ts index 2df61bb6..3d1697db 100644 --- a/src/run-agent.ts +++ b/src/run-agent.ts @@ -3016,8 +3016,13 @@ export function createAgentRuntime( // it sends back. A crash between them would otherwise keep the cause and // lose the effect, which a fold cannot re-derive. let unflushed: AgentLogEntry[] = []; + // Barriers waiting for the loop to finish its current step. + let flushWaiters: Array<() => void> = []; const flushJournal = (): void => { + const waiters = flushWaiters; + flushWaiters = []; + for (const resolve of waiters) resolve(); if (store === undefined || unflushed.length === 0) { return; } @@ -3053,6 +3058,27 @@ export function createAgentRuntime( if (finished) flushJournal(); }; + /** + * Resolves once the entries so far are handed to the store as one settled + * step. Mid-step (a transition's effects still running, its children's + * replies not yet delivered), a barrier waits for the loop to finish the + * step, so it cannot write a child's input without the reply it sends back. + * The loop flushes when it next waits; a host that is not reading + * `nextEvent` gets the flush on the next task instead. + */ + const stepSettled = (): Promise => { + if (wake !== undefined || finished || stopReason !== undefined) { + flushJournal(); + return Promise.resolve(); + } + return new Promise((resolve) => { + flushWaiters.push(resolve); + // A host that drives `transition` itself may not come back to + // `nextEvent` until this call starts: by the next task, the step is over. + setTimeout(flushJournal, 0); + }); + }; + /** * Every journal write ISSUED SO FAR, including the stragglers appended after * the run settled. Re-reads the chain tail until it stops moving, so a write @@ -3079,7 +3105,7 @@ export function createAgentRuntime( if (store === undefined) { return; } - flushJournal(); + await stepSettled(); while (journalPending > 0) { await journalChain; } @@ -3373,10 +3399,10 @@ export function createAgentRuntime( // happen, but a child runs live, and a fold reproduces its state only from // its own inputs. Stopping a child is the root's doing, which the root's // own entries already fold. - pendingJournal: () => { - flushJournal(); - return store !== undefined && journalPending > 0 ? awaitJournal().catch(() => {}) : undefined; - }, + pendingJournal: () => + store !== undefined && (journalPending > 0 || unflushed.length > 0) + ? awaitJournal().catch(() => {}) + : undefined, onChildEvent: (self, event) => { if (finished || event.type === "xstate.stop") return; const actorPath = actorPathOf(self); From 730b30a2f6e5bb2c6c14ebee8a3fe68116e6c99c Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 12:16:15 -0500 Subject: [PATCH 04/11] fix: review findings on crash points, resume timeouts, and late usage - effect crash points are the n-th run of a key, not a global index, so concurrent effects starting in another order still hit them - points no run died at go to report.unreached instead of passing silently - a timed-out resume is aborted and its effects made inert - requests that cannot be fingerprinted are not compared - finish journals usage that arrives while the last batch is written - usage.dropped repeat: document that modelCalls still counts the call --- docs/persistence.md | 1 + docs/usage-and-budgets.md | 2 +- src/crash-recovery.test.ts | 1 + src/crash-recovery.ts | 77 +++++++++++++++++++++++++++++--------- src/run-agent.ts | 24 ++++++++---- 5 files changed, 79 insertions(+), 26 deletions(-) diff --git a/docs/persistence.md b/docs/persistence.md index e4f50ae9..8a2c7cc6 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -126,6 +126,7 @@ Each resumed run is checked against the uninterrupted one. Each failure names it | `stuck` | The crashed run or its resume did not settle within `timeout` (default 5000 ms) | | `threw` | The crashed run, its resume, or the `store` factory threw | +- `report.unreached` lists the points no run died at, with why: the crashed run never started that effect, or one write took the log past that length. Their resumes were not checked. An effect point is the n-th run of its key, so concurrent effects starting in another order still hit it. - An effect in flight at the crash may run twice. That is the at-least-once contract (see [Idempotency keys](hosts.md#idempotency-keys)), and it is not reported. - Model calls are keyed by `callKey` without the execution id, and plain actors by invoke site. Only actors bound through string-keyed sources are seen. - Model calls run live only in the uninterrupted run. After a crash, a call it made gets its recorded answer, after the same delay, so a fake may answer by call order. diff --git a/docs/usage-and-budgets.md b/docs/usage-and-budgets.md index 9cc98554..01881eb5 100644 --- a/docs/usage-and-budgets.md +++ b/docs/usage-and-budgets.md @@ -190,7 +190,7 @@ A wrapper is still host-side. The machine cannot read its counts or react to the - Delivery is opt-in. The event is delivered only when an active state declares a transition that matches it, either `'@agent.usage'` or the wildcard `'*'`. Otherwise no usage event is sent to the machine. Wildcard matching follows XState semantics, so a catch-all `on: { '*': … }` does receive `@agent.usage`. Narrow the handler on `event.type` when the catch-all is meant for other events. - Declare the handler at machine level. A root `on: { '@agent.usage': … }` catches every call regardless of which state made it. A state-scoped handler drops calls made in other states. - Delivery goes to the run's root machine, including usage from requests inside an invoked child machine. Attribute those calls with `id`, `src`, and `model`. -- Each call is counted once across a crash. A call in flight at a crash re-executes on resume, and when the log already holds its usage, matched by `callKey`, the repeat is not delivered or journaled. `onTrace` reports it as `usage.dropped` with `reason: "repeat"`. +- Each call is counted once across a crash. A call in flight at a crash re-executes on resume, and when the log already holds its usage, matched by `callKey`, the repeat is not delivered or journaled. `onTrace` reports it as `usage.dropped` with `reason: "repeat"`. Its tokens are not added to `result.usage` again, but the call still counts toward `usage.modelCalls`. The payload has this shape. diff --git a/src/crash-recovery.test.ts b/src/crash-recovery.test.ts index 657aecb8..53f45fcd 100644 --- a/src/crash-recovery.test.ts +++ b/src/crash-recovery.test.ts @@ -88,6 +88,7 @@ describe("checkCrashRecovery", () => { events: [{ type: "REVIEW", ok: true }], }); expect(report.failures).toEqual([]); + expect(report.unreached).toEqual([]); expect(report.ok).toBe(true); expect(report.baseline.status).toBe("done"); expect(report.baseline.output).toMatchObject({ fetched: 42, published: true }); diff --git a/src/crash-recovery.ts b/src/crash-recovery.ts index 728116d8..2fd6d275 100644 --- a/src/crash-recovery.ts +++ b/src/crash-recovery.ts @@ -84,6 +84,8 @@ export type CrashPoint = effect: CrashEffect; /** The effect's position among the baseline's effects, from 0. */ index: number; + /** Which run of `effect.key` it is, from 0: the run dies at that one, whatever order concurrent effects start in. */ + occurrence: number; }; /** What went wrong after a crash. */ @@ -129,6 +131,12 @@ export interface CrashRecoveryReport { }; /** Every point a run was killed at, in order. */ points: CrashPoint[]; + /** + * Points no run died at, with why: the crashed run never started that + * effect, or one write took the log past that length. Their resumes were + * not checked. + */ + unreached: Array<{ point: CrashPoint; reason: string }>; failures: CrashRecoveryFailure[]; } @@ -204,22 +212,31 @@ export async function checkCrashRecovery( for (const entries of new Set([0, ...baselineStore.appends])) { points.push({ kind: "log", entries }); } + const seen = new Map(); baselineWorld.effects.forEach((effect, index) => { - points.push({ kind: "effect", effect: { kind: effect.kind, key: effect.key }, index }); + const occurrence = seen.get(effect.key) ?? 0; + seen.set(effect.key, occurrence + 1); + points.push({ + kind: "effect", + effect: { kind: effect.kind, key: effect.key }, + index, + occurrence, + }); }); const failures: CrashRecoveryFailure[] = []; + const unreached: CrashRecoveryReport["unreached"] = []; for (const point of points) { - failures.push( - ...(await crashAt(point, { - machine, - options, - timeout, - baselineCounts, - recorded: baselineWorld.answers, - expected, - })), - ); + const found = await crashAt(point, { + machine, + options, + timeout, + baselineCounts, + recorded: baselineWorld.answers, + expected, + }); + if (Array.isArray(found)) failures.push(...found); + else unreached.push({ point, reason: found.unreached }); } return { @@ -231,6 +248,7 @@ export async function checkCrashRecovery( effects: baselineWorld.effects.map(({ kind, key }) => ({ kind, key })), }, points, + unreached, failures, }; } @@ -250,7 +268,7 @@ interface CrashContext { async function crashAt( point: CrashPoint, ctx: CrashContext, -): Promise { +): Promise { const { machine, options, timeout } = ctx; const fail = (problem: CrashRecoveryProblem, message: string, key?: string) => ({ point, @@ -268,9 +286,10 @@ async function crashAt( // ─── The run that dies ─── const controller = new AbortController(); - let started = 0; + let runs = 0; const crashed = newWorld( - () => point.kind === "effect" && started++ === point.index, + (effect) => + point.kind === "effect" && effect.key === point.effect.key && runs++ === point.occurrence, ctx.recorded, ); const crash = () => { @@ -278,15 +297,27 @@ async function crashAt( crashed.crashed = true; controller.abort(new CrashError()); }; + // Why the run did not die at the point, if it did not. + let missed: string | undefined = + point.kind === "log" && point.entries === 0 + ? undefined + : point.kind === "log" + ? "the log never reached that length" + : "the run never started that effect"; const onEffect = crashed.onEffect; crashed.onEffect = (effect) => { const dies = onEffect(effect); - // The effect has started; the process dies before its result is recorded. - if (dies) queueMicrotask(crash); + if (dies) { + missed = undefined; + // The effect has started; the process dies before its result is recorded. + queueMicrotask(crash); + } return dies; }; const dying = observedStore(backing, crashed, (accepted) => { if (point.kind !== "log" || accepted < point.entries) return false; + missed = + accepted === point.entries ? undefined : `one write took the log to ${accepted} entries`; crash(); return true; }); @@ -308,19 +339,26 @@ async function crashAt( if (died === TIMED_OUT) { return [fail("stuck", `the run did not stop within ${timeout}ms after the crash`)]; } + if (missed !== undefined) { + return { unreached: missed }; + } // ─── The resume ─── const resumed = newWorld(() => false, ctx.recorded); + const resumeController = new AbortController(); let result: AgentRunResult | typeof TIMED_OUT; try { result = await withTimeout( - runTurns(machine, options, resumed, backing, crashed.hostEntries), + runTurns(machine, options, resumed, backing, crashed.hostEntries, resumeController.signal), timeout, ); } catch (error) { return [fail("threw", `the resume threw: ${String(error)}`)]; } if (result === TIMED_OUT) { + // Stop it, so nothing it still runs touches the next point's fakes. + resumed.crashed = true; + resumeController.abort(new CrashError()); return [fail("stuck", `the resume did not settle within ${timeout}ms`)]; } @@ -360,7 +398,10 @@ async function crashAt( ); for (const effect of resumed.effects) { if (effect.kind !== "model" || !crashedRequests.has(effect.key)) continue; - if (crashedRequests.get(effect.key) !== effect.request) { + const before = crashedRequests.get(effect.key); + // A request that cannot be fingerprinted is not compared. + if (before === undefined || effect.request === undefined) continue; + if (before !== effect.request) { failures.push( fail("request-changed", `'${effect.key}' re-executed with a different request`, effect.key), ); diff --git a/src/run-agent.ts b/src/run-agent.ts index 3d1697db..7b2a3342 100644 --- a/src/run-agent.ts +++ b/src/run-agent.ts @@ -249,9 +249,11 @@ export type AgentTraceEvent } | { type: "emit"; event: EmittedFrom } | { - /** A reserved `@agent.usage` event the run declined to deliver. The - * tokens still fold into {@link AgentRunResult.usage}; only the machine - * event is dropped. */ + /** A reserved `@agent.usage` event the run declined to deliver. For + * `settled`, the tokens still fold into {@link AgentRunResult.usage}; + * only the machine event is dropped. For `repeat`, the tokens are not + * added again (the log already holds them), but the call still counts + * toward `modelCalls`. */ type: "usage.dropped"; event: AgentUsageEvent; /** @@ -4336,13 +4338,21 @@ export function createAgentRuntime( // A decision's usage is queued after the event it chose; when that event // ends the machine, the usage is never handed out. Journal it out of // band, like usage the machine declares no transition for. - for (const event of mailbox) { - if (event.type === AGENT_USAGE_EVENT_TYPE) { + // The last step's entries are written before the outcome is decided, + // so a write that fails here settles the run `cause: "journal"`. Usage + // that arrives during the write is journaled too. + const journaledUsage = new Set(); + for (;;) { + const usage = mailbox.filter( + (event) => event.type === AGENT_USAGE_EVENT_TYPE && !journaledUsage.has(event), + ); + if (usage.length === 0) break; + for (const event of usage) { + journaledUsage.add(event); appendLogEntry(event); } + await drainJournal(); } - // The last step's entries are written before the outcome is decided, - // so a write that fails here settles the run `cause: "journal"`. await drainJournal(); const snapshot = current as AnyMachineSnapshot; let outcome: AgentRunOutcome; From 71f5cf580e3e3c5e940abd3d3689a036cc5f9ebe Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 12:19:31 -0500 Subject: [PATCH 05/11] fix(testing): unreached crash points fail the check; document timeout isolation limit --- docs/persistence.md | 2 +- src/crash-recovery.test.ts | 35 +++++++++++++++++++++++++++++++++++ src/crash-recovery.ts | 11 ++++++++--- 3 files changed, 44 insertions(+), 4 deletions(-) diff --git a/docs/persistence.md b/docs/persistence.md index 8a2c7cc6..8c57c142 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -126,7 +126,7 @@ Each resumed run is checked against the uninterrupted one. Each failure names it | `stuck` | The crashed run or its resume did not settle within `timeout` (default 5000 ms) | | `threw` | The crashed run, its resume, or the `store` factory threw | -- `report.unreached` lists the points no run died at, with why: the crashed run never started that effect, or one write took the log past that length. Their resumes were not checked. An effect point is the n-th run of its key, so concurrent effects starting in another order still hit it. +- `report.unreached` lists the points no run died at, with why: the crashed run never started that effect, or one write took the log past that length. Their resumes were not checked, so `report.ok` is `false`. An effect point is the n-th run of its key, so concurrent effects starting in another order still hit it. - An effect in flight at the crash may run twice. That is the at-least-once contract (see [Idempotency keys](hosts.md#idempotency-keys)), and it is not reported. - Model calls are keyed by `callKey` without the execution id, and plain actors by invoke site. Only actors bound through string-keyed sources are seen. - Model calls run live only in the uninterrupted run. After a crash, a call it made gets its recorded answer, after the same delay, so a fake may answer by call order. diff --git a/src/crash-recovery.test.ts b/src/crash-recovery.test.ts index 53f45fcd..f7570e36 100644 --- a/src/crash-recovery.test.ts +++ b/src/crash-recovery.test.ts @@ -488,3 +488,38 @@ describe("a store that fails", () => { expect(new Set(report.failures.map((failure) => failure.problem))).toEqual(new Set(["threw"])); }); }); + +describe("unreached points", () => { + test("a point no run dies at is reported and fails the check", async () => { + // The first run takes the long way; every run after skips the second tool. + let runs = 0; + const machine = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { + route: createAsyncLogic({ run: async () => (runs++ === 0 ? "long" : "short") }), + extra: createAsyncLogic({ run: async () => 1 }), + }, + }).createMachine({ + context: {}, + initial: "routing", + states: { + routing: { + invoke: { + src: "route", + onDone: ({ event }) => ({ target: event.output === "long" ? "extra" : "done" }), + }, + }, + extra: { invoke: { src: "extra", onDone: { target: "done" } } }, + done: { type: "final" }, + }, + }); + const report = await checkCrashRecovery(machine, {}); + expect( + report.unreached.some( + ({ point, reason }) => + point.kind === "effect" && reason === "the run never started that effect", + ), + ).toBe(true); + expect(report.ok).toBe(false); + }); +}); diff --git a/src/crash-recovery.ts b/src/crash-recovery.ts index 2fd6d275..8572d154 100644 --- a/src/crash-recovery.ts +++ b/src/crash-recovery.ts @@ -57,7 +57,12 @@ export interface CheckCrashRecoveryOptions { AgentRuntimeOptions, "executors" | "actors" | "store" | "threadId" | "signal" | "timers" | "abortWith" >; - /** Milliseconds one run (the baseline, a crashed run, or its resume) may take. Default 5000. */ + /** + * Milliseconds one run (the baseline, a crashed run, or its resume) may take. + * Default 5000. A run that times out is aborted and starts no more effects, + * but an effect already running that ignores its abort signal can still + * finish after the next point's `reset`. + */ timeout?: number; } @@ -118,7 +123,7 @@ export interface CrashRecoveryFailure { /** What {@link checkCrashRecovery} found. */ export interface CrashRecoveryReport { - /** True when every crash point recovered. */ + /** True when every crash point was reached and recovered: no failures, nothing unreached. */ ok: boolean; /** The uninterrupted run. */ baseline: { @@ -240,7 +245,7 @@ export async function checkCrashRecovery( } return { - ok: failures.length === 0, + ok: failures.length === 0 && unreached.length === 0, baseline: { status: baseline.status, ...(baseline.status === "done" ? { output: baseline.output } : {}), From 9f7a988c1f7e0dde4a2f5a71bfbc24be01d55740 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 12:20:42 -0500 Subject: [PATCH 06/11] fix: finish drains before every usage scan --- src/run-agent.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/run-agent.ts b/src/run-agent.ts index 7b2a3342..823ecfd9 100644 --- a/src/run-agent.ts +++ b/src/run-agent.ts @@ -4343,6 +4343,7 @@ export function createAgentRuntime( // that arrives during the write is journaled too. const journaledUsage = new Set(); for (;;) { + await drainJournal(); const usage = mailbox.filter( (event) => event.type === AGENT_USAGE_EVENT_TYPE && !journaledUsage.has(event), ); @@ -4351,9 +4352,7 @@ export function createAgentRuntime( journaledUsage.add(event); appendLogEntry(event); } - await drainJournal(); } - await drainJournal(); const snapshot = current as AnyMachineSnapshot; let outcome: AgentRunOutcome; if (stopReason !== undefined) { From 2770ffa60dc36928d6afd28fa8cf48f884d4da75 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 12:26:03 -0500 Subject: [PATCH 07/11] docs(testing): timeout covers the baseline throwing --- src/crash-recovery.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/crash-recovery.ts b/src/crash-recovery.ts index 8572d154..e27fc936 100644 --- a/src/crash-recovery.ts +++ b/src/crash-recovery.ts @@ -59,9 +59,10 @@ export interface CheckCrashRecoveryOptions { >; /** * Milliseconds one run (the baseline, a crashed run, or its resume) may take. - * Default 5000. A run that times out is aborted and starts no more effects, - * but an effect already running that ignores its abort signal can still - * finish after the next point's `reset`. + * Default 5000. An uninterrupted run that times out throws, so no point + * runs after it. A crashed run or resume that times out is aborted and + * starts no more effects, but an effect already running that ignores its + * abort signal can still finish after the next point's `reset`. */ timeout?: number; } From d4c818e980157ed4a960f8cee2f4cdc93508c956 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 16:17:35 -0500 Subject: [PATCH 08/11] fix: usage counts real spend; a call's usage and result land in one append --- .changeset/crash-recovery.md | 2 +- docs/usage-and-budgets.md | 2 +- src/crash-recovery.test.ts | 2 +- src/run-agent.ts | 40 +++++++++++++------------------ src/serialize-trace-event.test.ts | 2 +- 5 files changed, 20 insertions(+), 28 deletions(-) diff --git a/.changeset/crash-recovery.md b/.changeset/crash-recovery.md index 9a186168..ea6d83b1 100644 --- a/.changeset/crash-recovery.md +++ b/.changeset/crash-recovery.md @@ -6,6 +6,6 @@ Add `checkCrashRecovery` to `@statelyai/agent/testing`: it kills a durable run a Fix: a child machine that finished just before a crash (its last entry journaled, its parent's `xstate.done.actor` not yet) no longer leaves the resumed parent waiting forever. The restored child's completion is now delivered to its parent, at any depth. -Fix: a call re-executed after a crash no longer reports its usage twice. `@agent.usage` now carries the call's `callKey`; when the log already holds usage for that key, the repeat is dropped (`usage.dropped` with `reason: "repeat"`), so a token budget counts each call once. +`@agent.usage` now carries the call's `callKey`. Fix: a parent's `sendTo` into a journaled child machine, and the child's reply, are no longer lost when the run dies between appends. The store now receives every entry since the loop last waited in one atomic append (`expectedIndex` is the batch's first index), so an event and the child inputs it caused land together. diff --git a/docs/usage-and-budgets.md b/docs/usage-and-budgets.md index 01881eb5..7cc6bd04 100644 --- a/docs/usage-and-budgets.md +++ b/docs/usage-and-budgets.md @@ -190,7 +190,7 @@ A wrapper is still host-side. The machine cannot read its counts or react to the - Delivery is opt-in. The event is delivered only when an active state declares a transition that matches it, either `'@agent.usage'` or the wildcard `'*'`. Otherwise no usage event is sent to the machine. Wildcard matching follows XState semantics, so a catch-all `on: { '*': … }` does receive `@agent.usage`. Narrow the handler on `event.type` when the catch-all is meant for other events. - Declare the handler at machine level. A root `on: { '@agent.usage': … }` catches every call regardless of which state made it. A state-scoped handler drops calls made in other states. - Delivery goes to the run's root machine, including usage from requests inside an invoked child machine. Attribute those calls with `id`, `src`, and `model`. -- Each call is counted once across a crash. A call in flight at a crash re-executes on resume, and when the log already holds its usage, matched by `callKey`, the repeat is not delivered or journaled. `onTrace` reports it as `usage.dropped` with `reason: "repeat"`. Its tokens are not added to `result.usage` again, but the call still counts toward `usage.modelCalls`. +- Usage counts real spend across a crash. A call that re-executes on resume reports its usage again, under the same `callKey`, so a token budget sees both attempts. A call in flight at the crash that never reported usage is not counted. The payload has this shape. diff --git a/src/crash-recovery.test.ts b/src/crash-recovery.test.ts index f7570e36..8f98db10 100644 --- a/src/crash-recovery.test.ts +++ b/src/crash-recovery.test.ts @@ -220,7 +220,7 @@ describe("checkCrashRecovery", () => { }); describe("usage across a crash", () => { - test("a re-executed call's usage is counted once", async () => { + test("a finished call's usage and result land in one append, so no crash re-runs it", async () => { const machine = setupAgent({ schemas: createAgentSchemas({ context: z.object({ tokens: z.number() }), events: {} }), actors: { ask: createTextLogic({ model: "m", prompt: "q" }) }, diff --git a/src/run-agent.ts b/src/run-agent.ts index 823ecfd9..f8b9b64c 100644 --- a/src/run-agent.ts +++ b/src/run-agent.ts @@ -249,18 +249,13 @@ export type AgentTraceEvent } | { type: "emit"; event: EmittedFrom } | { - /** A reserved `@agent.usage` event the run declined to deliver. For - * `settled`, the tokens still fold into {@link AgentRunResult.usage}; - * only the machine event is dropped. For `repeat`, the tokens are not - * added again (the log already holds them), but the call still counts - * toward `modelCalls`. */ + /** A reserved `@agent.usage` event the run declined to deliver. The + * tokens still fold into {@link AgentRunResult.usage}; only the machine + * event is dropped. */ type: "usage.dropped"; event: AgentUsageEvent; - /** - * `'settled'`: the call settled after the run's cycle had resolved. - * `'repeat'`: a call re-executed after a crash; the log already holds its usage. - */ - reason: "settled" | "repeat"; + /** `'settled'`: the call settled after the run's cycle had resolved. */ + reason: "settled"; } | ( | { @@ -3020,6 +3015,10 @@ export function createAgentRuntime( let unflushed: AgentLogEntry[] = []; // Barriers waiting for the loop to finish its current step. let flushWaiters: Array<() => void> = []; + // A text call reported usage; its result follows within microtasks. The + // next idle flush waits a task so both land in one append: a crash between + // them would re-run a call that already finished. + let resultOwed = false; const flushJournal = (): void => { const waiters = flushWaiters; @@ -3259,19 +3258,7 @@ export function createAgentRuntime( usage, budget: { modelCalls: modelCallsSpent + modelCallCount, maxModelCalls }, }; - // At-least-once execution, exactly-once accounting: the log already - // holds this call's spend from before the crash. - if ( - source.callKey !== undefined && - logEntries.some( - (entry) => - entry.event.type === AGENT_USAGE_EVENT_TYPE && - (entry.event as AgentUsageEvent).callKey === source.callKey, - ) - ) { - onTrace({ type: "usage.dropped", event, reason: "repeat" }); - return; - } + if (source.kind === "text") resultOwed = true; let delivered = false; deliverUsageEvent( event, @@ -3746,7 +3733,12 @@ export function createAgentRuntime( return queued; } // Nothing queued: the entries so far are one settled step. Write them. - flushJournal(); + if (resultOwed) { + resultOwed = false; + setTimeout(flushJournal, 0); + } else { + flushJournal(); + } if (stopReason !== undefined || finished || !workInFlight()) { return QUIESCENT as unknown as EventObject; } diff --git a/src/serialize-trace-event.test.ts b/src/serialize-trace-event.test.ts index a16ad46f..d0cf0ce7 100644 --- a/src/serialize-trace-event.test.ts +++ b/src/serialize-trace-event.test.ts @@ -42,7 +42,7 @@ type HandWrittenJsonSerializableTraceEvent = { | { type: "stream.chunk"; request: JsonValue; chunk: string } | { type: "machine.transition"; from: JsonValue; snapshot: JsonValue; event: JsonValue } | { type: "emit"; event: JsonValue } - | { type: "usage.dropped"; event: JsonValue; reason: "settled" | "repeat" } + | { type: "usage.dropped"; event: JsonValue; reason: "settled" } | { type: "run.end"; status: "done"; output: JsonValue; snapshot: JsonValue } | { type: "run.end"; From 9eb8ec06b643ae628f51ed3d02084087150a3fe9 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 16:29:42 -0500 Subject: [PATCH 09/11] fix: a plain actor's finishing step keeps its effects; review fixes --- docs/persistence.md | 2 +- docs/usage-and-budgets.md | 2 +- src/crash-recovery.ts | 5 +- src/durable-child-state.test.ts | 113 +++++++++++++++++++++++++++++++- src/run-agent.ts | 5 +- 5 files changed, 122 insertions(+), 5 deletions(-) diff --git a/docs/persistence.md b/docs/persistence.md index 8c57c142..60633b7d 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -130,7 +130,7 @@ Each resumed run is checked against the uninterrupted one. Each failure names it - An effect in flight at the crash may run twice. That is the at-least-once contract (see [Idempotency keys](hosts.md#idempotency-keys)), and it is not reported. - Model calls are keyed by `callKey` without the execution id, and plain actors by invoke site. Only actors bound through string-keyed sources are seen. - Model calls run live only in the uninterrupted run. After a crash, a call it made gets its recorded answer, after the same delay, so a fake may answer by call order. -- Plain actors run live every time, so they must give the same result however often they run. A fake that keeps state passes `reset`, which runs before every run. +- Plain actors run live every time, so they must give the same result however often they run. A fake that keeps state passes `reset`, which runs before the baseline and each crashed run. It does not run before the resume, so the fake keeps what the crashed run did, as a real dependency would. - The outcome check compares final states, so a machine whose result depends on which concurrent call finishes first can fail it after a crash. Make the result order-independent, or test that part without concurrency. - `events` are host events, delivered one per idle settle. A resume skips the ones its log already holds. - `store` takes a factory for your own store adapter, one fresh store per run. A store that acknowledges a write it then loses shows up as `reran`. diff --git a/docs/usage-and-budgets.md b/docs/usage-and-budgets.md index 7cc6bd04..b13767f1 100644 --- a/docs/usage-and-budgets.md +++ b/docs/usage-and-budgets.md @@ -190,7 +190,7 @@ A wrapper is still host-side. The machine cannot read its counts or react to the - Delivery is opt-in. The event is delivered only when an active state declares a transition that matches it, either `'@agent.usage'` or the wildcard `'*'`. Otherwise no usage event is sent to the machine. Wildcard matching follows XState semantics, so a catch-all `on: { '*': … }` does receive `@agent.usage`. Narrow the handler on `event.type` when the catch-all is meant for other events. - Declare the handler at machine level. A root `on: { '@agent.usage': … }` catches every call regardless of which state made it. A state-scoped handler drops calls made in other states. - Delivery goes to the run's root machine, including usage from requests inside an invoked child machine. Attribute those calls with `id`, `src`, and `model`. -- Usage counts real spend across a crash. A call that re-executes on resume reports its usage again, under the same `callKey`, so a token budget sees both attempts. A call in flight at the crash that never reported usage is not counted. +- Usage counts real spend across a crash. A call that re-executes on resume reports its usage again, so a token budget sees both attempts. A call in flight at the crash that never reported usage is not counted. The payload has this shape. diff --git a/src/crash-recovery.ts b/src/crash-recovery.ts index e27fc936..c7d37217 100644 --- a/src/crash-recovery.ts +++ b/src/crash-recovery.ts @@ -50,7 +50,7 @@ export interface CheckCrashRecoveryOptions { * an entry it acknowledged is read back after the crash. */ store?: () => AgentEventLogStore; - /** Called before every run (the baseline and each crash), to reset fakes the executors or actors keep. */ + /** Called before the baseline and before each crashed run, to reset fakes the executors or actors keep. Not called before the resume: like a real dependency, a fake keeps what the crashed run did to it. */ reset?: () => void; /** Other runtime options (`maxModelCalls`, `verification`, …), shared by every run. */ runtime?: Omit< @@ -359,6 +359,9 @@ async function crashAt( timeout, ); } catch (error) { + // Stop it, so nothing it still runs touches the next point's fakes. + resumed.crashed = true; + resumeController.abort(new CrashError()); return [fail("threw", `the resume threw: ${String(error)}`)]; } if (result === TIMED_OUT) { diff --git a/src/durable-child-state.test.ts b/src/durable-child-state.test.ts index 36ec5b86..d27e3d20 100644 --- a/src/durable-child-state.test.ts +++ b/src/durable-child-state.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "vitest"; -import { createAsyncLogic } from "xstate"; +import { createAsyncLogic, createStepLogic } from "xstate"; import { z } from "zod"; import { createAgentRuntime, @@ -610,3 +610,114 @@ describe("a child that finished just before a crash", () => { expect(calls).toEqual([]); }); }); + +describe("a plain actor that finishes in the step it starts an effect", () => { + test("still runs that effect when the journal holds it back", async () => { + const ran: string[] = []; + const step = createStepLogic({ + context: {}, + run: ({ event }, enq) => { + if (event.type !== "GO") return; + enq.effect(() => void ran.push("effect")); + return { status: "done", output: "ok" }; + }, + }); + const child = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: { PING: z.object({}) } }), + actors: { step }, + }).createMachine({ + context: {}, + initial: "waiting", + states: { + waiting: { + invoke: { id: "step", src: "step", onDone: { target: "done" } }, + on: { + PING: ({ children }, enq) => { + enq.sendTo(children.step as never, { type: "GO" }); + }, + }, + }, + done: { type: "final" }, + }, + }); + const root = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: { WORK: z.object({}) } }), + actors: { child }, + }).createMachine({ + context: {}, + initial: "run", + states: { + run: { + invoke: { id: "child", src: "child", onDone: { target: "done" } }, + on: { + WORK: ({ children }, enq) => { + enq.sendTo(children.child as never, { type: "PING" }); + }, + }, + }, + done: { type: "final" }, + }, + }); + const store = createInMemoryEventLogStore(); + const result = await runToQuiescence(createAgentRuntime(root, { store, threadId: "t" }), { + input: undefined, + event: { type: "WORK" }, + } as AgentRunInit); + expect(result.status).toBe("done"); + expect(ran).toEqual(["effect"]); + }); +}); + +describe("a restored child completion inside an active parent", () => { + test("reaches its parent once", async () => { + const inner = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { summarize: createTextLogic({ model: "test-model", prompt: "Sum." }) }, + }).createMachine({ + context: {}, + initial: "work", + states: { + work: { invoke: { id: "summarize", src: "summarize", onDone: { target: "done" } } }, + done: { type: "final" }, + }, + }); + // `middle` stays put on its child's completion, so a second delivery would count twice. + const middle = setupAgent({ + schemas: createAgentSchemas({ context: z.object({ done: z.number() }), events: {} }), + actors: { inner }, + }).createMachine({ + context: { done: 0 }, + invoke: { + id: "inner", + src: "inner", + onDone: ({ context }) => ({ context: { done: context.done + 1 } }), + }, + }); + const root = setupAgent({ + schemas: createAgentSchemas({ context: z.object({}), events: {} }), + actors: { middle }, + }).createMachine({ context: {}, invoke: { id: "middle", src: "middle" } }); + const generateText = async () => ({ result: "sum" }); + const full = await runToQuiescence( + createAgentRuntime(root, { executors: { generateText } }), + {}, + ); + const cut = full.events.findIndex( + (entry) => + entry.event.type === "xstate.done.actor" && + (entry.event as { actorId?: string }).actorId === "inner", + ); + expect(cut).toBeGreaterThan(0); + const resumed = await runToQuiescence( + createAgentRuntime(root, { executors: { generateText } }), + { + events: full.events.slice(0, cut), + }, + ); + const middleDone = (persisted: unknown) => + (persisted as { children: { middle: { snapshot: { context: { done: number } } } } }).children + .middle.snapshot.context.done; + expect(middleDone(full.persist())).toBe(1); + expect(middleDone(resumed.persist())).toBe(1); + }); +}); diff --git a/src/run-agent.ts b/src/run-agent.ts index f8b9b64c..efc60be2 100644 --- a/src/run-agent.ts +++ b/src/run-agent.ts @@ -2266,6 +2266,9 @@ function journalGatedLogic( if (!Array.isArray(effects) || !effects.some((e) => (e as Effect)?.type === "@xstate.effect")) { return result; } + // The step that finishes the actor still runs its effects; only a stop + // before the journal catches up cancels them. + const finishing = (snapshot as { status?: string } | undefined)?.status === "done"; return [ snapshot, effects.map((effect) => { @@ -2275,7 +2278,7 @@ function journalGatedLogic( const waiting = pending(); if (waiting === undefined) return exec(...args); void waiting.then(() => { - if (self.getSnapshot().status === "active") exec(...args); + if (finishing || self.getSnapshot().status === "active") exec(...args); }); }; return { ...(effect as object), exec: gated, action: gated }; From 438c42a5ea17d1e74cc45100f4f7f2daee29f2dc Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 16:36:45 -0500 Subject: [PATCH 10/11] fix: a held effect is dropped when its cause fails to journal or the run stops --- docs/usage-and-budgets.md | 2 +- src/durable-child-state.test.ts | 29 +++++++++++++++++++++++++++-- src/run-agent.ts | 20 ++++++++++++-------- 3 files changed, 40 insertions(+), 11 deletions(-) diff --git a/docs/usage-and-budgets.md b/docs/usage-and-budgets.md index b13767f1..cadc9aa9 100644 --- a/docs/usage-and-budgets.md +++ b/docs/usage-and-budgets.md @@ -190,7 +190,7 @@ A wrapper is still host-side. The machine cannot read its counts or react to the - Delivery is opt-in. The event is delivered only when an active state declares a transition that matches it, either `'@agent.usage'` or the wildcard `'*'`. Otherwise no usage event is sent to the machine. Wildcard matching follows XState semantics, so a catch-all `on: { '*': … }` does receive `@agent.usage`. Narrow the handler on `event.type` when the catch-all is meant for other events. - Declare the handler at machine level. A root `on: { '@agent.usage': … }` catches every call regardless of which state made it. A state-scoped handler drops calls made in other states. - Delivery goes to the run's root machine, including usage from requests inside an invoked child machine. Attribute those calls with `id`, `src`, and `model`. -- Usage counts real spend across a crash. A call that re-executes on resume reports its usage again, so a token budget sees both attempts. A call in flight at the crash that never reported usage is not counted. +- Usage counts real spend across a crash. A call that re-executes on resume reports its usage again, so a token budget sees both attempts when the first attempt's usage was journaled before the crash. A call in flight at the crash that never reported usage is not counted. The payload has this shape. diff --git a/src/durable-child-state.test.ts b/src/durable-child-state.test.ts index d27e3d20..0c9a8be4 100644 --- a/src/durable-child-state.test.ts +++ b/src/durable-child-state.test.ts @@ -612,8 +612,7 @@ describe("a child that finished just before a crash", () => { }); describe("a plain actor that finishes in the step it starts an effect", () => { - test("still runs that effect when the journal holds it back", async () => { - const ran: string[] = []; + function createFinishing(ran: string[]) { const step = createStepLogic({ context: {}, run: ({ event }, enq) => { @@ -658,6 +657,12 @@ describe("a plain actor that finishes in the step it starts an effect", () => { done: { type: "final" }, }, }); + return root; + } + + test("still runs that effect when the journal holds it back", async () => { + const ran: string[] = []; + const root = createFinishing(ran); const store = createInMemoryEventLogStore(); const result = await runToQuiescence(createAgentRuntime(root, { store, threadId: "t" }), { input: undefined, @@ -666,6 +671,26 @@ describe("a plain actor that finishes in the step it starts an effect", () => { expect(result.status).toBe("done"); expect(ran).toEqual(["effect"]); }); + + test("drops that effect when the write of its cause fails", async () => { + const ran: string[] = []; + const root = createFinishing(ran); + const store = createInMemoryEventLogStore(); + const failing = { + ...store, + append: async (args: Parameters[0]) => { + if (args.entries.some((entry) => entry.event.type === "PING")) throw new Error("disk full"); + return store.append(args); + }, + }; + const result = await runToQuiescence( + createAgentRuntime(root, { store: failing, threadId: "t" }), + { input: undefined, event: { type: "WORK" } } as AgentRunInit, + ); + expect(result.status).not.toBe("done"); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(ran).toEqual([]); + }); }); describe("a restored child completion inside an active parent", () => { diff --git a/src/run-agent.ts b/src/run-agent.ts index efc60be2..8ff40109 100644 --- a/src/run-agent.ts +++ b/src/run-agent.ts @@ -1144,9 +1144,10 @@ interface RunAgentBindContext { * The journal writes still in flight, or `undefined` when none are (or the * run has no store). A plain actor a child machine starts waits for it, so * the child's earlier entries are durable before the actor's side effect - * runs. Unset off the runtime path. + * runs. Resolves `false` when the write failed or the run stopped + * meanwhile. Unset off the runtime path. */ - pendingJournal?: () => Promise | undefined; + pendingJournal?: () => Promise | undefined; } /** @@ -2257,7 +2258,7 @@ function journaledMachine( */ function journalGatedLogic( logic: AnyActorLogic, - pending: () => Promise | undefined, + pending: () => Promise | undefined, ): AnyActorLogic { type Result = [unknown, unknown[]]; type Effect = { type?: string; exec: (...args: unknown[]) => unknown }; @@ -2266,8 +2267,8 @@ function journalGatedLogic( if (!Array.isArray(effects) || !effects.some((e) => (e as Effect)?.type === "@xstate.effect")) { return result; } - // The step that finishes the actor still runs its effects; only a stop - // before the journal catches up cancels them. + // The step that finishes the actor still runs its effects; a stop, or a + // failed write of the entries that caused them, cancels them. const finishing = (snapshot as { status?: string } | undefined)?.status === "done"; return [ snapshot, @@ -2277,8 +2278,8 @@ function journalGatedLogic( const gated = (...args: unknown[]) => { const waiting = pending(); if (waiting === undefined) return exec(...args); - void waiting.then(() => { - if (finishing || self.getSnapshot().status === "active") exec(...args); + void waiting.then((ok) => { + if (ok && (finishing || self.getSnapshot().status === "active")) exec(...args); }); }; return { ...(effect as object), exec: gated, action: gated }; @@ -3393,7 +3394,10 @@ export function createAgentRuntime( // own entries already fold. pendingJournal: () => store !== undefined && (journalPending > 0 || unflushed.length > 0) - ? awaitJournal().catch(() => {}) + ? awaitJournal().then( + () => stopReason === undefined, + () => false, + ) : undefined, onChildEvent: (self, event) => { if (finished || event.type === "xstate.stop") return; From f9761ab79ae414e1a5e7b53a2c207bf13e78c2f6 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 6 Nov 2026 16:40:44 -0500 Subject: [PATCH 11/11] test: a held finishing effect is dropped when the run aborts mid-write --- src/durable-child-state.test.ts | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/src/durable-child-state.test.ts b/src/durable-child-state.test.ts index 0c9a8be4..a7097e3c 100644 --- a/src/durable-child-state.test.ts +++ b/src/durable-child-state.test.ts @@ -691,6 +691,30 @@ describe("a plain actor that finishes in the step it starts an effect", () => { await new Promise((resolve) => setTimeout(resolve, 10)); expect(ran).toEqual([]); }); + + test("drops that effect when the run is aborted while its cause is written", async () => { + const ran: string[] = []; + const root = createFinishing(ran); + const store = createInMemoryEventLogStore(); + const controller = new AbortController(); + const slow = { + ...store, + append: async (args: Parameters[0]) => { + if (args.entries.some((entry) => entry.event.type === "PING")) { + controller.abort(new Error("host stop")); + await new Promise((resolve) => setTimeout(resolve, 10)); + } + return store.append(args); + }, + }; + const result = await runToQuiescence( + createAgentRuntime(root, { store: slow, threadId: "t", signal: controller.signal }), + { input: undefined, event: { type: "WORK" } } as AgentRunInit, + ); + expect(result.status).not.toBe("done"); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(ran).toEqual([]); + }); }); describe("a restored child completion inside an active parent", () => {