diff --git a/.changeset/crash-recovery.md b/.changeset/crash-recovery.md new file mode 100644 index 00000000..ea6d83b1 --- /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. + +`@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/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..60633b7d 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -70,6 +70,73 @@ 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 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, 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. +- 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`. + +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..cadc9aa9 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`. +- 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. @@ -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..8f98db10 --- /dev/null +++ b/src/crash-recovery.test.ts @@ -0,0 +1,525 @@ +import { describe, expect, test } from "vitest"; +import { createAsyncLogic, createCallbackLogic } 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, + createAgentRuntime, + createTextLogic, + runToQuiescence, + 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.unreached).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 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" }) }, + }).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([]); + }); +}); + +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([]); + }); +}); + +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"])); + }); +}); + +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 new file mode 100644 index 00000000..c7d37217 --- /dev/null +++ b/src/crash-recovery.ts @@ -0,0 +1,832 @@ +/** + * `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 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< + AgentRuntimeOptions, + "executors" | "actors" | "store" | "threadId" | "signal" | "timers" | "abortWith" + >; + /** + * Milliseconds one run (the baseline, a crashed run, or its resume) may take. + * 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; +} + +/** + * 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; + /** 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. */ +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 crashed run, its resume, or the store factory 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 was reached and recovered: no failures, nothing unreached. */ + 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[]; + /** + * 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[]; +} + +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[]; + /** 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. */ + 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), + 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. + for (const entries of new Set([0, ...baselineStore.appends])) { + points.push({ kind: "log", entries }); + } + const seen = new Map(); + baselineWorld.effects.forEach((effect, 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) { + 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 { + ok: failures.length === 0 && unreached.length === 0, + baseline: { + status: baseline.status, + ...(baseline.status === "done" ? { output: baseline.output } : {}), + entries: baselineEntries.length, + effects: baselineWorld.effects.map(({ kind, key }) => ({ kind, key })), + }, + points, + unreached, + 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?.(); + 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(); + let runs = 0; + const crashed = newWorld( + (effect) => + point.kind === "effect" && effect.key === point.effect.key && runs++ === point.occurrence, + ctx.recorded, + ); + const crash = () => { + if (crashed.crashed) return; + 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); + 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; + }); + if (point.kind === "log" && point.entries === 0) dying.freeze(); + 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`)]; + } + 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, resumeController.signal), + 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) { + // 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`)]; + } + + 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; + 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), + ); + } + } + + 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( + 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, + 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, + /** Resuming: the host events the crashed run delivered, by entry id. */ + deliveredBefore?: readonly string[], + 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) => { + // 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, + 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 = 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++]); + await result.drain?.(); + } + await result.drain?.(); + return result; +} + +function newWorld( + crashesAt: (effect: StartedEffect) => boolean, + recorded?: ReadonlyMap, +): RunWorld { + return { + effects: [], + hostEntries: [], + 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, + // 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 (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)) { + 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..a7097e3c 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, @@ -527,3 +527,246 @@ 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([]); + }); +}); + +describe("a plain actor that finishes in the step it starts an effect", () => { + function createFinishing(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" }, + }, + }); + 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, + event: { type: "WORK" }, + } as AgentRunInit); + 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([]); + }); + + 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", () => { + 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 f0a9d640..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; } /** @@ -1383,6 +1384,7 @@ async function runTextRequest( ...(src !== "" ? { src } : {}), model: request.model, ...(request.name !== undefined ? { name: request.name } : {}), + ...(callKey !== undefined ? { callKey } : {}), }, self, ); @@ -1598,6 +1600,7 @@ function createCountingDecide( ...(attemptRequest.id ? { id: attemptRequest.id } : {}), ...(src !== "" ? { src } : {}), model: attemptRequest.model, + ...(callKey !== undefined ? { callKey } : {}), }, self, ), @@ -1662,6 +1665,7 @@ function createCountingFill( ...(id !== "" ? { id } : {}), ...(src !== "" ? { src } : {}), model: request.model, + ...(info?.callKey !== undefined ? { callKey: info.callKey } : {}), }, self, ), @@ -2254,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 }; @@ -2263,6 +2267,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; 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, effects.map((effect) => { @@ -2271,8 +2278,8 @@ function journalGatedLogic( const gated = (...args: unknown[]) => { const waiting = pending(); if (waiting === undefined) return exec(...args); - void waiting.then(() => { - if (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 }; @@ -2984,8 +2991,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 +3010,29 @@ 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[] = []; + // 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; + flushWaiters = []; + for (const resolve of waiters) resolve(); + if (store === undefined || unflushed.length === 0) { return; } + const entries = unflushed; + unflushed = []; journalPending++; journalChain = journalChain .then(async () => { @@ -3014,8 +3041,8 @@ export function createAgentRuntime( } await store.append({ threadId: threadId!, - expectedIndex: entry.index, - entries: [entry], + expectedIndex: entries[0]!.index, + entries, }); }) .catch((error: unknown) => { @@ -3027,6 +3054,36 @@ 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(); + }; + + /** + * 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 @@ -3037,6 +3094,7 @@ export function createAgentRuntime( if (store === undefined) { return; } + flushJournal(); let tail = journalChain; for (;;) { await tail; @@ -3052,6 +3110,7 @@ export function createAgentRuntime( if (store === undefined) { return; } + await stepSettled(); while (journalPending > 0) { await journalChain; } @@ -3203,6 +3262,7 @@ export function createAgentRuntime( usage, budget: { modelCalls: modelCallsSpent + modelCallCount, maxModelCalls }, }; + if (source.kind === "text") resultOwed = true; let delivered = false; deliverUsageEvent( event, @@ -3333,7 +3393,12 @@ export function createAgentRuntime( // 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, + store !== undefined && (journalPending > 0 || unflushed.length > 0) + ? awaitJournal().then( + () => stopReason === undefined, + () => false, + ) + : undefined, onChildEvent: (self, event) => { if (finished || event.type === "xstate.stop") return; const actorPath = actorPathOf(self); @@ -3593,6 +3658,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 +3726,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 +3739,13 @@ export function createAgentRuntime( if (queued !== undefined) { return queued; } + // Nothing queued: the entries so far are one settled step. Write them. + if (resultOwed) { + resultOwed = false; + setTimeout(flushJournal, 0); + } else { + flushJournal(); + } if (stopReason !== undefined || finished || !workInFlight()) { return QUIESCENT as unknown as EventObject; } @@ -4046,6 +4179,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. @@ -4203,8 +4337,18 @@ 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 (;;) { + await drainJournal(); + 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); } } @@ -4283,6 +4427,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/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); });