From efe9dd029b1fd4b333010f84997e4568eb760eea Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Sat, 24 Oct 2026 09:20:50 +0300 Subject: [PATCH 1/3] fix(log): a fork mints its own executionId and records forkedFrom forkEventLog and store.fork gave the fork its parent's executionId, so a forked thread's calls got the parent's callKeys and a cache keyed on them could return the parent's result. The fork's init entry now gets a fresh executionId and forkedFrom: { executionId, logIndex }. The in-memory, Durable Object and file stores fork through forkEventLog, and the conformance suite checks the lineage. --- .changeset/fork-lineage.md | 8 +++ docs/decisions.md | 2 +- docs/event-log.md | 10 ++-- docs/hosts.md | 4 +- .../cloudflare-agent-host/event-log-store.ts | 28 +++++++--- examples/code-review-loop/file-store.ts | 9 ++-- src/event-log-store-conformance.ts | 52 +++++++++++++++++-- src/event-log-store.test.ts | 12 ++++- src/event-log-store.ts | 10 ++-- src/event-log.test.ts | 19 +++++++ src/event-log.ts | 24 +++++++-- src/policy-decider.ts | 9 ++-- src/run-agent.test.ts | 51 ++++++++++++++++++ src/text-logic.ts | 8 +-- 14 files changed, 209 insertions(+), 37 deletions(-) create mode 100644 .changeset/fork-lineage.md diff --git a/.changeset/fork-lineage.md b/.changeset/fork-lineage.md new file mode 100644 index 00000000..b3b1ba16 --- /dev/null +++ b/.changeset/fork-lineage.md @@ -0,0 +1,8 @@ +--- +"@statelyai/agent": patch +--- + +**A fork is its own lineage.** `forkEventLog` and `store.fork` give the fork's init entry a fresh `metadata.executionId` and `metadata.forkedFrom: { executionId, logIndex }`. Before, a fork kept its parent's execution id, so a call the fork made at the same site got the same `info.callKey` as the parent's, and a cache keyed on it could return the parent's result. + +- **A custom store's `fork` must store what `forkEventLog` returns** for the prefix. `assertEventLogStoreConformance` now checks the lineage, and its fixtures start with a real init entry. +- A snapshot taken on the parent no longer takes the fast path when resuming the fork; the fork's log is folded instead. diff --git a/docs/decisions.md b/docs/decisions.md index 79d8f2bb..e7d2f683 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -300,7 +300,7 @@ await runAgent(deskMachine, { input, executors: { ...modelExecutors, decide } }) - The result carries the policy's token usage, and none when it decided with a rule. - A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. Give each of the policy's decisions an `onError` that falls back to an event the environment always accepts. - Every other `runAgent` option passes through: `store` and `metadata` to journal the policy's runs, `onTrace` to trace them. -- Each decision's policy run gets its own `store` thread, `policy:${info.callKey}:${hash}` by default, where `hash` is the SHA-256 of the policy input: a thread that already has entries is resumed, so a shared one would return an earlier choice. The hash keeps a fork that asks something else at the same call site off its parent's thread. Pass `threadId: (request, info) => ...` to name them yourself; it is required with a `store` when the decider is called without a `callKey`. +- Each decision's policy run gets its own `store` thread, `policy:${info.callKey}:${hash}` by default, where `hash` is the SHA-256 of the policy input: a thread that already has entries is resumed, so a shared one would return an earlier choice. The hash keeps a machine that asks something else at the same call site after a version bridge, which keeps the call key, off the earlier thread. Pass `threadId: (request, info) => ...` to name them yourself; it is required with a `store` when the decider is called without a `callKey`. - The policy's input must be JSON, since it opens the run's event log. The default drops keys whose value is `undefined`; any other value that is not JSON throws before the run. To route by request name, wrap it: `decide: (request, info) => (request.name === "chooseResolution" ? policyDecide(request, info) : modelExecutors.decide!(request, info))`, where `policyDecide` is the executor `createPolicyDecider` returned. diff --git a/docs/event-log.md b/docs/event-log.md index d80fb36c..9b66bfa4 100644 --- a/docs/event-log.md +++ b/docs/event-log.md @@ -189,13 +189,17 @@ Replayability rests on pure transitions. ## Fork -`forkEventLog(entries, upToIndex)` returns the prefix `[0, upToIndex)` as a new, still self-contained log. `upToIndex` is exclusive and must leave the init entry in place. A fork copies the init entry, so it inherits the parent's execution id. +`forkEventLog(entries, upToIndex)` returns the prefix `[0, upToIndex)` as a new, still self-contained log. `upToIndex` is exclusive and must leave the init entry in place. + +- The fork is its own lineage. Its init entry gets a fresh `metadata.executionId` and `metadata.forkedFrom: { executionId, logIndex }`, naming the parent's id and the cutoff. Every other entry is copied as is. +- Call keys derive from the execution id, so a call the fork makes never shares an `info.callKey` with one its parent makes at the same site. +- A snapshot of the parent does not take the fork's fast path. Resuming the fork with one folds its log instead. ```ts no-check const branch = forkEventLog(entries, 8); ``` -`store.fork({ threadId, newThreadId, upToIndex })` does the same at the storage layer. To diverge, append different entries to the new thread. +`store.fork({ threadId, newThreadId, upToIndex })` does the same at the storage layer. To diverge, append different entries to the new thread. A store you write stores what `forkEventLog` returns for the prefix; the conformance suite checks it. ## Decision attempts @@ -265,7 +269,7 @@ const next = await store.length("session-1"); // the next expectedIndex - `append` is atomic. A stale writer fails with `AgentEventLogConflictError`, carrying `threadId`, `expectedIndex`, and `actualIndex`, so two hosts resuming one thread resolve to exactly one winner. Entry indices must be contiguous from `expectedIndex`, and ids unique within the thread. - `createAgentRuntime(machine, { store, threadId })` drives all three: `read` to resume, `append` per entry, `read` again to check an explicit `events` log against the thread. - `read` and `length` are the per-thread reads. Everything else about a thread is derived from its entries. -- `fork` copies a prefix onto a fresh thread. +- `fork` copies a prefix onto a fresh thread, with the fork's own lineage on its init entry (see [Fork](#fork)). - `listThreads` is optional: it lists the threads that hold at least one entry, so a caller need not record thread ids itself. ```ts no-check diff --git a/docs/hosts.md b/docs/hosts.md index d53286b9..07092565 100644 --- a/docs/hosts.md +++ b/docs/hosts.md @@ -32,10 +32,10 @@ Execution is at-least-once. A host runs the request and then journals its comple - `requestId` is the invoke site, and `n` counts that site's completions in the log — each iteration of a looped state gets its own key. - A decision retry appends its attempt ordinal: `:#.`, where `attempt` is the number of prior failed attempts. Each retry is a different request, so it gets a different key. - A resumed run re-executing an in-flight request passes the same `callKey` as the original attempt. -- A fork copies the init entry, so it keeps the parent's lineage id and can reuse results cached under the parent's keys for the requests it has not changed. +- A fork mints its own lineage id (see [Fork](event-log.md#fork)), so none of its calls shares a key with its parent's. - It is `undefined` off the `runToQuiescence` path (a bare `provideExecutors` bind) and on a run with no event log. -The key names the call _site_, not the request. A fork inherits the parent's lineage id, so a fork that changes what it asks at the same invoke site produces the same key for a different request. Cache on `callKey` **and** a fingerprint of the request, and reuse the cached result only when the request also matches. Pass `callKey` to a provider as its dedupe key for that identical request: +The key names the call _site_, not the request. A version bridge keeps the lineage id, so a migrated machine that asks something else at the same invoke site produces the same key for a different request. Cache on `callKey` **and** a fingerprint of the request, and reuse the cached result only when the request also matches. Pass `callKey` to a provider as its dedupe key for that identical request: ```ts no-check const executors = { diff --git a/examples/cloudflare-agent-host/event-log-store.ts b/examples/cloudflare-agent-host/event-log-store.ts index e4adae53..c253e2ff 100644 --- a/examples/cloudflare-agent-host/event-log-store.ts +++ b/examples/cloudflare-agent-host/event-log-store.ts @@ -29,7 +29,7 @@ import { type AgentEventLogStore, type AgentLogEntry, } from "@statelyai/agent"; -import { assertAgentLogEntry } from "@statelyai/agent/log"; +import { assertAgentLogEntry, forkEventLog } from "@statelyai/agent/log"; export interface DurableObjectEventLogStoreOptions { /** Table holding the log entries. Defaults to `agent_event_log`. */ table?: string; @@ -179,13 +179,25 @@ export function createDurableObjectEventLogStore( `cannot fork up to index ${upTo}; expected an integer in [1, ${sourceLength}].`, ); } - sql.exec( - `INSERT INTO ${table} (thread_id, idx, entry_id, entry) - SELECT ?, idx, entry_id, entry FROM ${table} WHERE thread_id = ? AND idx < ?`, - newThreadId, - threadId, - upTo, - ); + const prefix = sql + .exec( + `SELECT entry FROM ${table} WHERE thread_id = ? AND idx < ? ORDER BY idx ASC`, + threadId, + upTo, + ) + .toArray() + .map((row) => JSON.parse(String(row.entry)) as AgentLogEntry); + // `forkEventLog` gives the fork its own lineage: a fresh executionId + // and `forkedFrom` on its init entry. + for (const entry of forkEventLog(prefix, upTo)) { + sql.exec( + `INSERT INTO ${table} (thread_id, idx, entry_id, entry) VALUES (?, ?, ?, ?)`, + newThreadId, + entry.index, + entry.id, + JSON.stringify(entry), + ); + } }); }, diff --git a/examples/code-review-loop/file-store.ts b/examples/code-review-loop/file-store.ts index 008a3bcd..438f0d6c 100644 --- a/examples/code-review-loop/file-store.ts +++ b/examples/code-review-loop/file-store.ts @@ -20,7 +20,11 @@ import { } from "node:fs"; import path from "node:path"; import type { AgentEventLogStore, AgentLogEntry } from "@statelyai/agent"; -import { AgentEventLogConflictError, assertAgentLogEntry } from "@statelyai/agent/log"; +import { + AgentEventLogConflictError, + assertAgentLogEntry, + forkEventLog, +} from "@statelyai/agent/log"; const EXTENSION = ".jsonl"; @@ -91,8 +95,7 @@ export function createFileEventLogStore(dir: string): AgentEventLogStore { } writeFileSync( fileOf(dir, newThreadId), - source - .slice(0, upTo) + forkEventLog(source, upTo) .map((entry) => `${JSON.stringify(entry)}\n`) .join(""), ); diff --git a/src/event-log-store-conformance.ts b/src/event-log-store-conformance.ts index 088ed114..3fce5d83 100644 --- a/src/event-log-store-conformance.ts +++ b/src/event-log-store-conformance.ts @@ -9,7 +9,13 @@ * @module */ import { AgentEventLogConflictError, type AgentEventLogStore } from "./event-log-store.js"; -import { AGENT_EVENT_SCHEMA_VERSION, type AgentLogEntry, type JsonValue } from "./event-log.js"; +import { + AGENT_EVENT_SCHEMA_VERSION, + AGENT_INIT_EVENT_TYPE, + getLogExecutionId, + type AgentLogEntry, + type JsonValue, +} from "./event-log.js"; /** Produces a fresh, empty store for one case. */ type CreateStore = () => Promise | AgentEventLogStore; @@ -40,8 +46,16 @@ function entry(index: number, type: string, metadata?: Record } as AgentLogEntry; } +// A log's own first entry: the reserved init entry, naming its lineage. +function initFixture(executionId = "exec-src"): AgentLogEntry { + return entry(0, AGENT_INIT_EVENT_TYPE, { executionId }); +} + +// Entries `start…start+count`; from 0, the first is the init entry. function entriesFrom(start: number, count: number): AgentLogEntry[] { - return Array.from({ length: count }, (_unused, i) => entry(start + i, `e${start + i}`)); + return Array.from({ length: count }, (_unused, i) => + start + i === 0 ? initFixture() : entry(start + i, `e${start + i}`), + ); } function assertJsonEqual(actual: unknown, expected: unknown, message: string): void { @@ -273,7 +287,7 @@ const cases: Array<{ name: string; run: (create: CreateStore) => Promise } await store.append({ threadId: "fork-1", expectedIndex: 1, entries: [entry(1, "branch")] }); assertJsonEqual( (await store.read("fork-1")).map((e) => (e.event as { type: string }).type), - ["e0", "branch"], + [AGENT_INIT_EVENT_TYPE, "branch"], "a forked thread must append independently from its copied length", ); if ((await store.length("src")) !== 3) { @@ -281,6 +295,36 @@ const cases: Array<{ name: string; run: (create: CreateStore) => Promise } } }, }, + { + name: "a fork is its own lineage: a fresh executionId, and forkedFrom naming the parent", + async run(create) { + const store = await create(); + const source = entriesFrom(0, 3); + await store.append({ threadId: "src", expectedIndex: 0, entries: source }); + await store.fork({ threadId: "src", newThreadId: "fork-a", upToIndex: 2 }); + await store.fork({ threadId: "src", newThreadId: "fork-b", upToIndex: 2 }); + + const [a, b] = [await store.read("fork-a"), await store.read("fork-b")]; + const ids = [getLogExecutionId(a), getLogExecutionId(b)]; + if (typeof ids[0] !== "string" || typeof ids[1] !== "string") { + fail("a fork's init entry must carry a string metadata.executionId"); + } + if (ids.includes("exec-src") || ids[0] === ids[1]) { + fail("every fork must mint its own executionId, not reuse the parent's or a sibling's"); + } + assertJsonEqual( + a[0]!.metadata?.forkedFrom, + { executionId: "exec-src", logIndex: 2 }, + "a fork's init entry must record forkedFrom: { executionId, logIndex }", + ); + assertJsonEqual( + a.slice(1), + source.slice(1, 2), + "a fork must copy every entry after the init entry unchanged", + ); + assertJsonEqual(await store.read("src"), source, "forking must not touch the source thread"); + }, + }, { name: "fork rejects a non-empty target, an unknown source, and an out-of-range cutoff (0 included)", async run(create) { @@ -351,7 +395,7 @@ const cases: Array<{ name: string; run: (create: CreateStore) => Promise } "listThreads on an empty store must list nothing", ); for (const threadId of ["user:b", "user:a", "admin:x", "user:c"]) { - await store.append({ threadId, expectedIndex: 0, entries: [entry(0, "start")] }); + await store.append({ threadId, expectedIndex: 0, entries: [initFixture()] }); } await store.fork({ threadId: "user:a", newThreadId: "user:d" }); diff --git a/src/event-log-store.test.ts b/src/event-log-store.test.ts index 97d8fa69..e090db44 100644 --- a/src/event-log-store.test.ts +++ b/src/event-log-store.test.ts @@ -1,5 +1,9 @@ import { describe, expect, it, test } from "vitest"; -import { AGENT_EVENT_SCHEMA_VERSION, type AgentLogEntry } from "./event-log.js"; +import { + AGENT_EVENT_SCHEMA_VERSION, + AGENT_INIT_EVENT_TYPE, + type AgentLogEntry, +} from "./event-log.js"; import { AgentEventLogConflictError, createInMemoryEventLogStore } from "./event-log-store.js"; import { assertEventLogStoreConformance } from "./event-log-store-conformance.js"; @@ -72,7 +76,11 @@ describe("createInMemoryEventLogStore listThreads", () => { async function storeWith(threadIds: string[]) { const store = createInMemoryEventLogStore(); for (const threadId of threadIds) { - await store.append({ threadId, expectedIndex: 0, entries: [testEntry(0, "start")] }); + await store.append({ + threadId, + expectedIndex: 0, + entries: [testEntry(0, AGENT_INIT_EVENT_TYPE)], + }); } return store; } diff --git a/src/event-log-store.ts b/src/event-log-store.ts index 214370e7..8e3c805e 100644 --- a/src/event-log-store.ts +++ b/src/event-log-store.ts @@ -5,7 +5,7 @@ * @module */ import { AgentError } from "./errors.js"; -import { assertAgentLogEntry, type AgentLogEntry } from "./event-log.js"; +import { assertAgentLogEntry, forkEventLog, type AgentLogEntry } from "./event-log.js"; /** Options for {@link AgentEventLogStore.listThreads}. */ export interface ListThreadsOptions { @@ -64,8 +64,10 @@ export interface AgentEventLogStore { * init entry and stays a self-contained log. The fork then appends * independently. Rejects (plain `Error`) if `newThreadId` already has * entries, the source is unknown, or `upToIndex` is out of range. - * Implementations may copy-on-write or physically copy; observable behavior - * must match a full copy. + * + * The fork is its own lineage: store what {@link forkEventLog} returns for + * the prefix, whose init entry carries a fresh `metadata.executionId` and + * `metadata.forkedFrom`. Every other entry must match the source. */ fork(input: { threadId: string; @@ -188,7 +190,7 @@ export function createInMemoryEventLogStore(): AgentEventLogStore { } threads.set( newThreadId, - source.slice(0, upTo).map((entry) => clone(entry)), + forkEventLog(source, upTo).map((entry) => clone(entry)), ); }, diff --git a/src/event-log.test.ts b/src/event-log.test.ts index 6cda4f1e..b40fb5ba 100644 --- a/src/event-log.test.ts +++ b/src/event-log.test.ts @@ -608,6 +608,25 @@ describe("forkEventLog", () => { expect(replay(machine, forked).snapshot.value).toBe("working"); }); + test("gives the fork its own executionId and records the parent in forkedFrom", () => { + const entries = buildLog(); + const parent = getLogExecutionId(entries); + + const forked = forkEventLog(entries, 2); + const sibling = forkEventLog(entries, 2); + + expect(getLogExecutionId(forked)).toEqual(expect.any(String)); + expect(getLogExecutionId(forked)).not.toBe(parent); + expect(getLogExecutionId(forked)).not.toBe(getLogExecutionId(sibling)); + expect(forked[0]!.metadata?.forkedFrom).toEqual({ + ...(parent !== undefined ? { executionId: parent } : {}), + logIndex: 2, + }); + // The parent's own entries are untouched. + expect(getLogExecutionId(entries)).toBe(parent); + expect(entries[0]!.metadata?.forkedFrom).toBeUndefined(); + }); + test("rejects a cutoff that drops the init entry or overruns the log", () => { const entries = buildLog(); diff --git a/src/event-log.ts b/src/event-log.ts index 26cadee1..7e002a28 100644 --- a/src/event-log.ts +++ b/src/event-log.ts @@ -751,8 +751,8 @@ function isSnapshotInitEntry(entry: AgentLogEntry): boolean { * Minted once per run and inherited verbatim by every resume of that log, so * it names one lineage rather than one process. Entry ids cannot serve this * role — the built-in ids are index-derived, so every log's init entry is - * `evt_00000000`. A fork copies the init entry, so it shares its parent's - * execution id by construction. + * `evt_00000000`. A fork ({@link forkEventLog}, `store.fork`) mints its own + * and records the parent's in `metadata.forkedFrom`. * * Returns `undefined` for a log with no init entry and for one written without * the field — callers must treat the id as optional. @@ -835,6 +835,12 @@ export function validateReplayEntries( /** * The prefix `[0, upToIndex)` of a log, as a new log. `upToIndex` is exclusive * and must leave the init entry in place, so the fork is still self-contained. + * + * The fork is its own lineage: its init entry gets a fresh + * `metadata.executionId` and `metadata.forkedFrom: { executionId, logIndex }` + * naming the parent's id (when it had one) and the cutoff. Call keys derive + * from the execution id, so a fork's calls never collide with its parent's. + * Every other entry is copied as is. */ export function forkEventLog( entries: readonly AgentLogEntry[], @@ -851,7 +857,19 @@ export function forkEventLog( `forkEventLog: the forked prefix must start with the reserved '${AGENT_INIT_EVENT_TYPE}' entry.`, ); } - return forked; + const [init, ...rest] = forked; + const parentExecutionId = getLogExecutionId(forked); + const forkedFrom: Record = { + ...(parentExecutionId !== undefined ? { executionId: parentExecutionId } : {}), + logIndex: upToIndex, + }; + return [ + { + ...init!, + metadata: { ...init!.metadata, executionId: crypto.randomUUID(), forkedFrom }, + }, + ...rest, + ]; } // --- Occurrence + usage ---------------------------------------------------- diff --git a/src/policy-decider.ts b/src/policy-decider.ts index 4b25b45f..4dd59afb 100644 --- a/src/policy-decider.ts +++ b/src/policy-decider.ts @@ -16,9 +16,10 @@ export interface PolicyDeciderOptions extends O * retry of one, needs its own thread: a thread that already has entries is * resumed, so a shared one would return an earlier decision's choice. Default * with a `store`: `policy:${info.callKey}:${hash}`, where `hash` is the - * SHA-256 of the policy input's canonical JSON. The call key is unique per call and attempt and stays the - * same when a crashed run re-issues the call; the hash keeps a fork that asks - * something else at the same call site off the parent's thread. Required + * SHA-256 of the policy input's canonical JSON. The call key is unique per + * call and attempt and stays the same when a crashed run re-issues the call; + * the hash keeps a machine that asks something else at the same call site + * after a version bridge, which keeps the call key, off the earlier thread. Required * with a `store` when the decider is called without a `callKey`. */ threadId?: (request: AgentExecutorDecisionRequest, info?: AgentRequestExecutorInfo) => string; @@ -90,7 +91,7 @@ export function createPolicyDecider( info?: AgentRequestExecutorInfo, ) { if (threadId) return threadId(request, info); - // The call key names the call site, not the request: a fork shares it. + // The call key names the call site, not the request: a version bridge keeps it. if (info?.callKey) return `policy:${info.callKey}:${await sha256(canonicalEventJson(policyInput))}`; throw new AgentError( diff --git a/src/run-agent.test.ts b/src/run-agent.test.ts index 3625f6c5..1eec6aa4 100644 --- a/src/run-agent.test.ts +++ b/src/run-agent.test.ts @@ -3959,6 +3959,57 @@ describe("runAgent event log", () => { expect(secondCalls[0]!.callKey).toMatch(/^[0-9a-f-]+:b#1$/); }); + test("a forked thread's calls get their own callKeys, not the parent's", async () => { + const askMachine = setup({ actors: { answer } }).createMachine({ + id: "ask-later", + context: () => ({}), + initial: "waiting", + states: { + waiting: { on: { GO: { target: "asking" } } }, + asking: { + invoke: { + id: "ask", + src: "answer", + input: () => ({}), + onDone: { target: "done" }, + } as never, + }, + done: { type: "final" }, + }, + }); + const store = createInMemoryEventLogStore(); + const generateText = async () => ({ result: "ok" }); + await runToQuiescence( + createAgentRuntime(askMachine, { store, threadId: "parent", executors: { generateText } }), + { input: undefined }, + ); + await store.fork({ threadId: "parent", newThreadId: "branch" }); + + const keyOf = async (threadId: string) => { + const keys: Array = []; + await runToQuiescence( + createAgentRuntime(askMachine, { + store, + threadId, + executors: { + generateText: async (_request, info) => { + keys.push(info?.callKey); + return { result: "ok" }; + }, + }, + }), + { event: { type: "GO" } }, + ); + return keys[0]!; + }; + const [parentKey, branchKey] = [await keyOf("parent"), await keyOf("branch")]; + + const branchId = getLogExecutionId(await store.read("branch"))!; + expect(branchKey).toBe(`${branchId}:ask#1`); + expect(parentKey).toBe(`${getLogExecutionId(await store.read("parent"))}:ask#1`); + expect(branchKey).not.toBe(parentKey); + }); + test("decision retries get distinct, replay-stable callKeys", async () => { const schemas = createAgentSchemas({ context: z.object({}), diff --git a/src/text-logic.ts b/src/text-logic.ts index bcf6939a..3942c2cd 100644 --- a/src/text-logic.ts +++ b/src/text-logic.ts @@ -807,9 +807,11 @@ export interface AgentRequestExecutorInfo { * `a` being the number of prior failed attempts), so a retry never collides * with the rejected attempt it is replacing. * - * It identifies the CALL SITE, not the request: a fork inherits its parent's - * lineage id, so cache on `callKey` together with a fingerprint of the - * request and reuse a cached result only when the request also matches. + * It identifies the CALL SITE, not the request: a version bridge keeps the + * lineage id, so a migrated machine can ask something else under the same + * key. Cache on `callKey` together with a fingerprint of the request and + * reuse a cached result only when the request also matches. A fork mints + * its own lineage id, so its keys never match its parent's. * * Undefined off the `createAgentRuntime` path, and for a log with no `executionId`. */ From 95692cb7f86d4c5e00c3a584a095d3cabdd120d6 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Sat, 24 Oct 2026 09:28:14 +0300 Subject: [PATCH 2/3] docs(log): forkedFrom.executionId is present only when the parent has one --- .changeset/fork-lineage.md | 2 +- docs/event-log.md | 2 +- src/event-log.ts | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/.changeset/fork-lineage.md b/.changeset/fork-lineage.md index b3b1ba16..8f92bc33 100644 --- a/.changeset/fork-lineage.md +++ b/.changeset/fork-lineage.md @@ -2,7 +2,7 @@ "@statelyai/agent": patch --- -**A fork is its own lineage.** `forkEventLog` and `store.fork` give the fork's init entry a fresh `metadata.executionId` and `metadata.forkedFrom: { executionId, logIndex }`. Before, a fork kept its parent's execution id, so a call the fork made at the same site got the same `info.callKey` as the parent's, and a cache keyed on it could return the parent's result. +**A fork is its own lineage.** `forkEventLog` and `store.fork` give the fork's init entry a fresh `metadata.executionId` and `metadata.forkedFrom: { executionId?, logIndex }` (the parent's id, when it has one, and the cutoff). Before, a fork kept its parent's execution id, so a call the fork made at the same site got the same `info.callKey` as the parent's, and a cache keyed on it could return the parent's result. - **A custom store's `fork` must store what `forkEventLog` returns** for the prefix. `assertEventLogStoreConformance` now checks the lineage, and its fixtures start with a real init entry. - A snapshot taken on the parent no longer takes the fast path when resuming the fork; the fork's log is folded instead. diff --git a/docs/event-log.md b/docs/event-log.md index 9b66bfa4..7ed12efe 100644 --- a/docs/event-log.md +++ b/docs/event-log.md @@ -191,7 +191,7 @@ Replayability rests on pure transitions. `forkEventLog(entries, upToIndex)` returns the prefix `[0, upToIndex)` as a new, still self-contained log. `upToIndex` is exclusive and must leave the init entry in place. -- The fork is its own lineage. Its init entry gets a fresh `metadata.executionId` and `metadata.forkedFrom: { executionId, logIndex }`, naming the parent's id and the cutoff. Every other entry is copied as is. +- The fork is its own lineage. Its init entry gets a fresh `metadata.executionId` and `metadata.forkedFrom: { executionId?, logIndex }`: the cutoff, and the parent's id when the parent's init entry has one. Every other entry is copied as is. - Call keys derive from the execution id, so a call the fork makes never shares an `info.callKey` with one its parent makes at the same site. - A snapshot of the parent does not take the fork's fast path. Resuming the fork with one folds its log instead. diff --git a/src/event-log.ts b/src/event-log.ts index 7e002a28..cf735967 100644 --- a/src/event-log.ts +++ b/src/event-log.ts @@ -837,8 +837,8 @@ export function validateReplayEntries( * and must leave the init entry in place, so the fork is still self-contained. * * The fork is its own lineage: its init entry gets a fresh - * `metadata.executionId` and `metadata.forkedFrom: { executionId, logIndex }` - * naming the parent's id (when it had one) and the cutoff. Call keys derive + * `metadata.executionId` and `metadata.forkedFrom: { executionId?, logIndex }`: + * the cutoff, and the parent's id when it had one. Call keys derive * from the execution id, so a fork's calls never collide with its parent's. * Every other entry is copied as is. */ From 337efa88562886184236a5c23f8b38ae9750a06a Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Sat, 24 Oct 2026 09:45:55 +0300 Subject: [PATCH 3/3] fix(log): explicit events must match the thread's lineage; an untouched fork bridges with its parent's snapshot --- .changeset/fork-lineage.md | 3 ++- docs/event-log.md | 2 ++ src/run-agent.test.ts | 53 ++++++++++++++++++++++++++++++++++++++ src/run-agent.ts | 17 ++++++++++-- 4 files changed, 72 insertions(+), 3 deletions(-) diff --git a/.changeset/fork-lineage.md b/.changeset/fork-lineage.md index 8f92bc33..f5ab2d9b 100644 --- a/.changeset/fork-lineage.md +++ b/.changeset/fork-lineage.md @@ -5,4 +5,5 @@ **A fork is its own lineage.** `forkEventLog` and `store.fork` give the fork's init entry a fresh `metadata.executionId` and `metadata.forkedFrom: { executionId?, logIndex }` (the parent's id, when it has one, and the cutoff). Before, a fork kept its parent's execution id, so a call the fork made at the same site got the same `info.callKey` as the parent's, and a cache keyed on it could return the parent's result. - **A custom store's `fork` must store what `forkEventLog` returns** for the prefix. `assertEventLogStoreConformance` now checks the lineage, and its fixtures start with a real init entry. -- A snapshot taken on the parent no longer takes the fast path when resuming the fork; the fork's log is folded instead. +- A snapshot taken on the parent no longer takes the fast path when resuming the fork; the fork's log is folded instead. A version bridge still accepts it while the fork has appended nothing since its cutoff. +- Explicit `events` for a stored thread must also match its init entry's `executionId`, so a parent's log is refused for its fork's thread. diff --git a/docs/event-log.md b/docs/event-log.md index 7ed12efe..0aa9a7cb 100644 --- a/docs/event-log.md +++ b/docs/event-log.md @@ -194,6 +194,8 @@ Replayability rests on pure transitions. - The fork is its own lineage. Its init entry gets a fresh `metadata.executionId` and `metadata.forkedFrom: { executionId?, logIndex }`: the cutoff, and the parent's id when the parent's init entry has one. Every other entry is copied as is. - Call keys derive from the execution id, so a call the fork makes never shares an `info.callKey` with one its parent makes at the same site. - A snapshot of the parent does not take the fork's fast path. Resuming the fork with one folds its log instead. +- A version bridge accepts the parent's snapshot while the fork has appended nothing since its cutoff, since the fork's log is then the parent's. After the fork appends, bridge it with its own snapshot. +- Passing the parent's log as `events` for the fork's `threadId` is refused (`event-log-conflict`): the init entries name different lineages. ```ts no-check const branch = forkEventLog(entries, 8); diff --git a/src/run-agent.test.ts b/src/run-agent.test.ts index 1eec6aa4..56988876 100644 --- a/src/run-agent.test.ts +++ b/src/run-agent.test.ts @@ -38,6 +38,7 @@ import { } from "./index.js"; import { AGENT_INIT_EVENT_TYPE, + forkEventLog, getLogExecutionId, getUsageFromEvents, replay, @@ -4304,6 +4305,58 @@ describe("runAgent event log", () => { expect(bridged.status).toBe("done"); }); + test("version bridge: a fork bridges with its parent's snapshot until it moves on", async () => { + const v1 = makeLoggedMachine({ version: "1" }); + const parent = await runToQuiescence(createAgentRuntime(v1, { executors: executors() }), { + input: undefined, + }); + const parentTail = JSON.parse(JSON.stringify(parent.persist())) as Snapshot; + const forked = forkEventLog(parent.events, parent.events.length); + const v2 = makeLoggedMachine({ + version: "2", + migrate: (snapshot) => ({ ...(snapshot as object), version: "2" }), + }); + + // Untouched since the cutoff: the parent's tail snapshot is the fork's tail too. + const bridged = await runToQuiescence(createAgentRuntime(v2, { executors: executors() }), { + events: forked, + snapshot: parentTail, + event: { type: "APPROVE" } as never, + }); + expect(bridged.status).toBe("done"); + expect(getLogExecutionId(bridged.events)).toBe(getLogExecutionId(forked)); + + // Once the fork appends, the parent's snapshot no longer caches its tail. + const moved = await runToQuiescence(createAgentRuntime(v1, { executors: executors() }), { + events: forked, + event: { type: "APPROVE" } as never, + }); + expect(moved.events.length).toBeGreaterThan(forked.length); + await expect( + runToQuiescence(createAgentRuntime(v2, { executors: executors() }), { + events: moved.events, + snapshot: parentTail, + }), + ).rejects.toBeInstanceOf(AgentMachineVersionMismatchError); + }); + + test("a parent's log passed as a fork thread's `events` is refused", async () => { + const machine = makeLoggedMachine({ version: "1" }); + const store = createInMemoryEventLogStore(); + const parent = await runToQuiescence( + createAgentRuntime(machine, { store, threadId: "parent", executors: executors() }), + { input: undefined }, + ); + await store.fork({ threadId: "parent", newThreadId: "branch" }); + + await expect( + runToQuiescence( + createAgentRuntime(machine, { store, threadId: "branch", executors: executors() }), + { events: parent.events }, + ), + ).rejects.toMatchObject({ code: "event-log-conflict" }); + }); + test("a log written by another machine is never a resume", async () => { const machine = makeLoggedMachine({ version: "1" }); const other = makeLoggedMachine({ version: "1", id: "other-logged" }); diff --git a/src/run-agent.ts b/src/run-agent.ts index 7ea49368..158a30a1 100644 --- a/src/run-agent.ts +++ b/src/run-agent.ts @@ -2128,7 +2128,9 @@ function assertThreadMatchesEvents( storedEntry.id === givenEntry.id && storedEntry.index === givenEntry.index && canonicalEventJson(storedEntry.event) === canonicalEventJson(givenEntry.event) && - (storedHash === undefined || givenHash === undefined || storedHash === givenHash); + (storedHash === undefined || givenHash === undefined || storedHash === givenHash) && + // A fork differs from its parent only here: its init entry's lineage id. + (index > 0 || getLogExecutionId([storedEntry]) === getLogExecutionId([givenEntry])); if (!same) { throw new AgentError( "event-log-conflict", @@ -3026,7 +3028,18 @@ export function createAgentRuntime( // version: an older or foreign snapshot would silently roll the // thread back to state the log has already moved past. const bridgeMeta = readAgentMeta(init.snapshot); - if (inheritedExecutionId === undefined || bridgeMeta?.logId !== inheritedExecutionId) { + // A fork that has appended nothing since its cutoff holds exactly + // its parent's entries, so the parent's snapshot of them is its tail too. + const forkedFrom = (resumeEvents[0]!.metadata as { forkedFrom?: unknown } | undefined) + ?.forkedFrom as { executionId?: unknown; logIndex?: unknown } | undefined; + const parentCache = + forkedFrom?.executionId !== undefined && + bridgeMeta?.logId === forkedFrom.executionId && + forkedFrom.logIndex === resumeEvents.length; + if ( + inheritedExecutionId === undefined || + (bridgeMeta?.logId !== inheritedExecutionId && !parentCache) + ) { throw new AgentMachineVersionMismatchError( tail.id, tail.index,