feat(temporal): Carried the turn past a worker that lost a step it was running.

packages/core/src/session/runner/index.ts

@@ -59,6 +59,10 @@ export interface SealStepInput {
    * provider error, so a seal that re-derives this from the log would keep calling a provider that
    * just failed. Absent on a re-drive, where the log is all there is. */
   readonly needsContinuation?: boolean
+  /** This step is being closed away from the host that ran it, so the files are not this seal's to
+   * touch: it is standing in a directory that never saw the tools, and the host that did may still
+   * be inside one of them. Writing the step down is the whole job here. */
+  readonly withoutTheTree?: boolean
 }
 
 /** One recorded tool call, to run on its own. */

packages/core/src/session/runner/llm.ts

@@ -280,7 +280,7 @@ const layer = Layer.effect(
         return yield* Effect.die(continueAfterCompaction(currentStep))
       const startSnapshot = yield* snapshots.capture()
       // Ship the pre-step tree so another host can rebuild the worktree; best-effort inside push.
-      if (startSnapshot) yield* snapshotSync.push(startSnapshot)
+      if (startSnapshot) yield* snapshotSync.push(startSnapshot, session.id)
       const publisher = createLLMEventPublisher(events, {
         sessionID: session.id,
         agent: agent.id,
@@ -393,7 +393,7 @@ const layer = Layer.effect(
           if (stepSettlement && !publisher.hasProviderError() && !deferTools) {
             const endSnapshot = yield* snapshots.capture()
             // Ship the post-step tree: this is the state a resumed step on another host needs.
-            if (endSnapshot) yield* snapshotSync.push(endSnapshot)
+            if (endSnapshot) yield* snapshotSync.push(endSnapshot, session.id)
             const files =
               startSnapshot && endSnapshot
                 ? yield* snapshots
@@ -572,7 +572,7 @@ const layer = Layer.effect(
       yield* failInterruptedTools(input.sessionID)
       const startSnapshot = inFlight.snapshot?.start
       const endSnapshot = yield* snapshots.capture()
-      if (endSnapshot) yield* snapshotSync.push(endSnapshot)
+      if (endSnapshot) yield* snapshotSync.push(endSnapshot, input.sessionID)
       const files =
         startSnapshot && endSnapshot
           ? yield* snapshots
@@ -654,9 +654,11 @@ const layer = Layer.effect(
       // sends a request carrying a tool_use with no tool_result and the provider rejects it.
       yield* failInterruptedTools(input.sessionID, context)
       const startSnapshot = target.snapshot?.start
-      const endSnapshot = yield* snapshots.capture()
+      // A seal closing a step away from its host captures a directory that never ran the tools, so
+      // what it would ship is the state before them. The host that has them is the one that ships.
+      const endSnapshot = input.withoutTheTree ? undefined : yield* snapshots.capture()
       // Ship the post-step tree: this is the state a later step on another host needs.
-      if (endSnapshot) yield* snapshotSync.push(endSnapshot)
+      if (endSnapshot) yield* snapshotSync.push(endSnapshot, input.sessionID)
       const files =
         startSnapshot && endSnapshot
           ? yield* snapshots
@@ -823,7 +825,7 @@ const layer = Layer.effect(
       yield* shipping.withLock(location.directory)(
         Effect.gen(function* () {
           const afterTool = yield* snapshots.capture().pipe(Effect.catch(() => Effect.succeed(undefined)))
-          if (afterTool) yield* snapshotSync.push(afterTool)
+          if (afterTool) yield* snapshotSync.push(afterTool, input.sessionID)
         }),
       )
       return { outcome: "settled" } as ToolCallResult

packages/core/src/snapshot-sync.ts

@@ -13,6 +13,8 @@ import { ChildProcess } from "effect/unstable/process"
 import { eq } from "drizzle-orm"
 import { Database } from "./database/database"
 import { makeLocationNode } from "./effect/app-node"
+import { EventV2 } from "./event"
+import { EventSequenceTable } from "./event/sql"
 import { FSUtil } from "./fs-util"
 import { Git } from "./git"
 import { Global } from "./global"
@@ -26,8 +28,23 @@ import { readWorktreeTip, writeWorktreeTip } from "./snapshot/tip"
 import { Hash } from "./util/hash"
 
 export interface Interface {
-  /** A stale tip fails the caller; packing and insertion errors are logged. */
-  readonly push: (tree: Snapshot.ID) => Effect.Effect<void>
+  /**
+   * A stale tip fails the caller; packing and insertion errors are logged.
+   *
+   * `sessionID` is what the files are being shipped for. Given it, a publisher a newer attempt has
+   * superseded is refused, which is the one case the tip check cannot answer: a tool whose dispatch
+   * was abandoned is still standing on the tree it read, so its pack is clean and reverts whatever
+   * ran in its place. Omitted by callers with no session behind them, and then nothing is fenced.
+   *
+   * Two clauses a lost host is held to, and a change here has to keep both:
+   *
+   *   1. The seal that closes a step away from its host publishes nothing. Between the dispatch
+   *      failing and the next attempt claiming the log there is a window where nothing fences the
+   *      old host, and the only thing that makes it harmless is that nobody else publishes then.
+   *   2. Once the session has moved past that host, what it publishes for the closed step is
+   *      refused.
+   */
+  readonly push: (tree: Snapshot.ID, sessionID?: string) => Effect.Effect<void>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/v2/SnapshotSync") {}
@@ -73,7 +90,29 @@ const layer = Layer.effect(
         .all()
         .pipe(Effect.orDie, Effect.map(chainHead))
 
-    const push = Effect.fn("SnapshotSync.push")(function* (tree: Snapshot.ID) {
+    // Whether a newer attempt holds this session's log. The token is the one the drain claimed and
+    // provided, so this asks the same question the log's own fence asks of every append: is what is
+    // writing still the attempt the session is on?
+    const superseded = Effect.fn("SnapshotSync.superseded")(function* (sessionID: string) {
+      const owner = yield* EventV2.EventOwner
+      // Outside a drain nothing claimed anything, so there is nothing to be superseded by.
+      if (owner === undefined) return false
+      const row = yield* db
+        .select({ ownerID: EventSequenceTable.owner_id })
+        .from(EventSequenceTable)
+        .where(eq(EventSequenceTable.aggregate_id, sessionID))
+        .get()
+        .pipe(Effect.orDie)
+      return row?.ownerID != null && row.ownerID !== owner
+    })
+
+    const push = Effect.fn("SnapshotSync.push")(function* (tree: Snapshot.ID, sessionID?: string) {
+      // The files travel under the token the transcript does. A dispatch the session stopped
+      // waiting for keeps running, and what it publishes afterwards would otherwise be the newest
+      // state the store holds, because it is standing exactly where it was told to stand.
+      if (source && sessionID && (yield* superseded(sessionID))) {
+        return yield* Effect.die(new Error(`refusing to ship ${worktree}: a newer attempt holds ${sessionID}`))
+      }
       // A host that has not caught up must not publish its older files as the next tree.
       // This reading is not an atomic head claim and does not fence concurrent publishers.
       if (source) {

packages/core/test/worktree-materialize.test.ts

@@ -12,6 +12,8 @@ import { Effect, Fiber, Layer } from "effect"
 import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
 import { LayerNode } from "@opencode-ai/core/effect/layer-node"
 import { Database } from "@opencode-ai/core/database/database"
+import { EventV2 } from "@opencode-ai/core/event"
+import { EventSequenceTable } from "@opencode-ai/core/event/sql"
 import { Global } from "@opencode-ai/core/global"
 import { Location } from "@opencode-ai/core/location"
 import { AbsolutePath } from "@opencode-ai/core/schema"
@@ -448,3 +450,57 @@ describe("WorktreeMaterializer", () => {
   )
 })
 
+// A dispatch the session stopped waiting for keeps running, and the files it ships afterwards would
+// be the newest state the store holds: the tip check cannot refuse them, because the host is still
+// standing exactly where it was told to stand. The event log already fences a superseded attempt
+// out of the transcript, and the packs travel under the same token.
+describe("SnapshotSync owner fence", () => {
+  it.live("refuses a pack from an attempt the session has moved past", () =>
+    Effect.gen(function* () {
+      const tmp = yield* Effect.promise(() => tmpdir())
+      const root = realpathSync(tmp.path)
+      const worktree = path.join(root, "project")
+      const file = path.join(root, "shared.db")
+      yield* Effect.promise(async () => {
+        await mkdir(worktree, { recursive: true })
+        await $`git init -q ${worktree}`.quiet()
+        await $`git -C ${worktree} config user.email t@t`.quiet()
+        await $`git -C ${worktree} config user.name t`.quiet()
+        await writeFile(path.join(worktree, "tracked.txt"), "v1\n")
+        await $`git -C ${worktree} add .`.quiet()
+        await $`git -C ${worktree} commit -qm seed`.quiet()
+      })
+
+      const A = yield* Layer.build(captureStack(file, worktree, path.join(root, "host-a-data")))
+      const onDatabase = <A2, E2>(use: (db: Database.Interface["db"]) => Effect.Effect<A2, E2>) =>
+        Database.Service.use(({ db }) => use(db)).pipe(
+          Effect.orDie,
+          Effect.provide(Database.layerFromPath(file)),
+          Effect.scoped,
+        )
+      // The session is on a later attempt than the one that is about to publish.
+      yield* onDatabase((db) =>
+        db.insert(EventSequenceTable).values({ aggregate_id: "ses_fenced", seq: 1, owner_id: "run:1:2" }).run(),
+      )
+
+      const captured = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A))
+      if (!captured) throw new Error("expected a capture")
+      const shipped = (owner: string) =>
+        Effect.exit(
+          SnapshotSync.Service.use((s) => s.push(captured, "ses_fenced")).pipe(
+            Effect.provideService(EventV2.EventOwner, owner),
+            Effect.provide(A),
+          ),
+        )
+
+      const stale = yield* shipped("run:1:1")
+      expect(stale._tag).toBe("Failure")
+      expect(yield* onDatabase((db) => db.select().from(SnapshotPackTable).all())).toHaveLength(0)
+
+      // The attempt the session is actually on ships as usual.
+      const current = yield* shipped("run:1:2")
+      expect(current._tag).toBe("Success")
+      expect(yield* onDatabase((db) => db.select().from(SnapshotPackTable).all())).toHaveLength(1)
+    }),
+  )
+})

packages/temporal/src/drain.ts

@@ -57,6 +57,10 @@ export interface InSessionOptions {
   /** Take the event log for this attempt. True for the writer that supersedes the attempt before it;
    * the other units of a stepped turn ride that writer's token. */
   readonly claim?: boolean
+  /** Leave the project tree alone. For a seal closing a step away from its host: rebuilding here
+   * would put this host on the newest state while the one that ran the tools may still be writing,
+   * and nothing the seal does needs the files. */
+  readonly withoutTheTree?: boolean
 }
 
 /** Run `use` against a session's runner. Undefined when the session is gone, which is not a run
@@ -78,7 +82,7 @@ export const makeInSession =
       // Take the event log before running so a superseded attempt's later appends are fenced.
       if (options.claim && owner) yield* events.claim(session.id, owner)
       // A worker taking this step on a host without the project tree rebuilds it from snapshot packs.
-      yield* worktrees.ensure(session.location.directory)
+      if (!options.withoutTheTree) yield* worktrees.ensure(session.location.directory)
       return yield* SessionRunner.Service.use((runner) => use(runner, session)).pipe(
         Effect.provide(locations.get(session.location)),
       )

packages/temporal/src/stepped-drain.ts

@@ -54,6 +54,10 @@ export interface SealDrainInput {
   readonly assistantMessageID?: string
   readonly needsContinuation?: boolean
   readonly owner: string
+  /** This step is being closed away from the host that was running it. The tree is not this seal's
+   * to rebuild or to ship: the host that ran the tools is the only one holding what they did, and
+   * it may still be inside one of them. Writing the step down is the whole job. */
+  readonly withoutTheTree?: boolean
 }
 
 export interface SteppedDrainDeps {
@@ -120,14 +124,15 @@ export const makeSteppedDrains = ({ inSession, stepQueue }: SteppedDrainDeps) =>
     runAtBoundary(
       input.sessionID,
       signal,
-      inSession(input.sessionID, input.owner, {}, (runner, session) =>
+      inSession(input.sessionID, input.owner, { withoutTheTree: input.withoutTheTree }, (runner, session) =>
         runner
           .sealStep({
             sessionID: session.id,
             step: input.step,
             settlement: input.settlement,
             assistantMessageID: input.assistantMessageID,
             needsContinuation: input.needsContinuation,
+            withoutTheTree: input.withoutTheTree,
           })
           .pipe(Effect.map((result) => toStepResult(input.step, result))),
       ).pipe(Effect.map((result) => result ?? toStepResult(input.step))),

packages/temporal/src/stepped-turn.ts

@@ -220,5 +220,34 @@ export const makeSteppedTurn =
     if (unsettled.length > 0)
       log?.("step did not settle every call it dispatched", { step: model.step, calls: unsettled })
 
-    return seal(false)
+    // A pinned attempt that started and failed does not take the turn with it, even though nothing
+    // can say whether the tool over there has stopped: the log fences a superseded attempt out of
+    // the transcript, and the pack store refuses its files under the same token, so what that host
+    // is still doing cannot reach the session. The step is closed on the shared queue, where a
+    // worker that is answering can take it, and the turn goes on with whatever the seal says
+    // follows. What does not move is the rest of this step: its calls stay where their host has them.
+    const closeElsewhere = () => {
+      log?.("the step lost its host; closing it elsewhere and carrying the turn on", {
+        sessionID: input.sessionID,
+        step: model.step,
+      })
+      // Without the tree. This seal is standing in a directory that never ran the step's tools, so
+      // what it would ship is the state before them, and the host that did run them may still be
+      // inside one. Between the dispatch failing and the next step claiming the log there is a
+      // window where nothing fences that host, and the only thing that makes the window harmless
+      // is that nobody else publishes during it.
+      return activities.sealStep({ ...sealing(false), withoutTheTree: true })
+    }
+    const resumes = () => uncertain !== undefined
+    if (resumes()) return closeElsewhere()
+    try {
+      return await seal(false)
+    } catch (error) {
+      // The seal is the other way a step loses its host, and it is the half that has to be written
+      // down: a step with no ending recorded is one no follower ever hears about. Re-sealing where
+      // a worker is answering is what an ordinary retry of this activity does; the pinned rule is
+      // the only reason it did not already happen.
+      if (stopped || !resumes()) throw error
+      return await closeElsewhere()
+    }
   }

packages/temporal/test/pinned-retry.test.ts

@@ -0,0 +1,105 @@
+import { expect, it } from "bun:test"
+import { fileURLToPath } from "node:url"
+import { ApplicationFailure } from "@temporalio/common"
+import { TestWorkflowEnvironment } from "@temporalio/testing"
+import { Worker } from "@temporalio/worker"
+
+// The server must not retry a pinned body or move it while its physical state is unknown. What the
+// step does instead is close itself where a worker is answering: the call stays with the host that
+// has it, and the turn goes on rather than ending with that host.
+it("gives a pinned dispatch one attempt, keeps its call, and closes the step elsewhere", async () => {
+  const env = await TestWorkflowEnvironment.createLocal()
+  let phase: "tool" | "seal" = "tool"
+  const attempts = { tool: 0, seal: 0 }
+  const shared = { tool: 0, seal: 0 }
+  const sealed: Array<boolean | undefined> = []
+  try {
+    const worker = await Worker.create({
+      connection: env.nativeConnection,
+      namespace: env.namespace,
+      taskQueue: "pin-retry-main",
+      workflowsPath: fileURLToPath(new URL("../src/workflow.ts", import.meta.url)),
+      activities: {
+        runModelCall: async () => ({
+          kind: "called",
+          step: 1,
+          calls: phase === "tool" ? [{ id: "call_write", name: "write", assistantMessageID: "msg_write" }] : [],
+          owner: "run:1:1",
+          queue: "pin-retry-tools",
+        }),
+        runToolCall: async () => {
+          shared.tool++
+          return { outcome: "settled" }
+        },
+        sealStep: async (input: { withoutTheTree?: boolean }) => {
+          shared.seal++
+          sealed.push(input.withoutTheTree === true)
+          return { continue: false, step: 1, promotion: null }
+        },
+      },
+    })
+    const pinned = await Worker.create({
+      connection: env.nativeConnection,
+      namespace: env.namespace,
+      taskQueue: "pin-retry-tools",
+      activities: {
+        runToolCall: async () => {
+          attempts.tool++
+          throw ApplicationFailure.create({ message: "tool failed after dispatch", type: "ToolUnavailable" })
+        },
+        sealStep: async () => {
+          attempts.seal++
+          throw ApplicationFailure.create({ message: "seal failed after dispatch", type: "SealUnavailable" })
+        },
+      },
+    })
+    await worker.runUntil(() =>
+      pinned.runUntil(async () => {
+        for (const kind of ["tool", "seal"] as const) {
+          phase = kind
+          shared.tool = 0
+          shared.seal = 0
+          sealed.length = 0
+          const handle = await env.client.workflow.start("sessionTurn", {
+            workflowId: `pin-retry-session-${kind}`,
+            taskQueue: "pin-retry-main",
+            args: ["ses_pin_retry", { stepped: true, startWithWake: false }],
+          })
+          const resumed = handle.executeUpdate("resume").then(
+            () => "completed",
+            () => "failed",
+          )
+          let timer: ReturnType<typeof setTimeout> | undefined
+          try {
+            const outcome = await Promise.race([
+              resumed,
+              new Promise<string>((resolve) => {
+                timer = setTimeout(() => resolve("waiting for retries"), 2_000)
+              }),
+            ])
+            // The turn is handed back, not ended: the step was closed on the shared queue and the
+            // supervisor gets a result to carry on from.
+            expect(outcome).toBe("completed")
+            // One attempt on the pinned queue, and never a second one anywhere: a retry's queue
+            // timeout cannot rule out the first attempt still running.
+            expect(attempts[kind]).toBe(1)
+            // The call itself does not move. Only the seal does, and that is the step saying what
+            // it had rather than the tool being run somewhere else.
+            expect(shared.tool).toBe(0)
+            expect(shared.seal).toBe(1)
+            // And it seals without the tree: this worker never ran the step, and the one that did
+            // may still be inside a tool, so rebuilding here would put it on the newest state and
+            // shipping from here would publish the state before the step.
+            expect(sealed).toEqual([true])
+          } finally {
+            clearTimeout(timer)
+            await handle.terminate()
+            await resumed
+          }
+        }
+      }),
+    )
+  } finally {
+    await env.teardown()
+  }
+}, 120_000)