fix(log): a fork mints its own executionId and records forkedFrom

.changeset/fork-lineage.md

@@ -0,0 +1,9 @@
+---
+"@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 }` (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 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.

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.

docs/event-log.md

@@ -189,13 +189,19 @@ 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 }`: 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);
 ```
 
-`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 +271,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

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: `<executionId>:<requestId>#<n>.<attempt>`, 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 = {

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),
+          );
+        }
       });
     },
 

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(""),
       );

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> | AgentEventLogStore;
@@ -40,8 +46,16 @@ function entry(index: number, type: string, metadata?: Record<string, JsonValue>
   } 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,14 +287,44 @@ const cases: Array<{ name: string; run: (create: CreateStore) => Promise<void> }
       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) {
         fail("appending to a fork must not touch the source thread");
       }
     },
   },
+  {
+    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<void> }
         "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" });
 

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;
   }

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)),
       );
     },
 

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();
 

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 }`:
+ * 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.
  */
 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<string, JsonValue> = {
+    ...(parentExecutionId !== undefined ? { executionId: parentExecutionId } : {}),
+    logIndex: upToIndex,
+  };
+  return [
+    {
+      ...init!,
+      metadata: { ...init!.metadata, executionId: crypto.randomUUID(), forkedFrom },
+    },
+    ...rest,
+  ];
 }
 
 // --- Occurrence + usage ----------------------------------------------------