feat(temporal): Made a session outlive the client that started it.

packages/core/src/event.ts

@@ -4,7 +4,7 @@ import { Cause, Context, Duration, Effect, Layer, Option, PubSub, Queue, Schema
 import { Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
-import { and, asc, eq, gt, inArray } from "drizzle-orm"
+import { and, asc, eq, gt, inArray, isNull } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
 import { Flag } from "./flag/flag"
@@ -188,6 +188,41 @@ export interface LayerOptions {
 /** Chosen to be well under what a person notices in a transcript while staying one cheap indexed
  * read per subscribed session. In-process commits still wake instantly; this only catches what the
  * wake cannot see, so it is worth its cost only where another process writes: see `pollingNode`. */
+// Whether the token already on the row is a later attempt of the same activity execution than the
+// one claiming. Tokens are `run:activityId:attempt`, so only the attempt is comparable: two
+// different activity ids are two different units of work and neither supersedes the other.
+const supersededBy = (held: string, claimer: string): boolean => {
+  const split = (token: string) => {
+    const cut = token.lastIndexOf(":")
+    const head = token.slice(0, cut)
+    const idAt = head.lastIndexOf(":")
+    const id = head.slice(idAt + 1)
+    return {
+      run: head.slice(0, idAt),
+      id,
+      // Temporal hands out activity ids as an increasing sequence within a run, so they order the
+      // units of work. A token from an earlier step is a zombie, whatever its attempt number says.
+      activity: Number(id),
+      attempt: Number(token.slice(cut + 1)),
+    }
+  }
+  const a = split(held)
+  const b = split(claimer)
+  // Different runs cannot be ordered from the tokens alone, and a continue-as-new legitimately
+  // starts a new one, so those are allowed through. A zombie from a run that rolled over is the
+  // case this does not cover.
+  if (a.run !== b.run) return false
+  const ordered = Number.isInteger(a.activity) && Number.isInteger(b.activity)
+  // Activity ids are an increasing sequence when Temporal assigns them, but a caller may set its
+  // own. Without numbers to compare, two different units of work cannot be ordered, and only two
+  // attempts of the same one can.
+  if (ordered && a.activity !== b.activity) return a.activity > b.activity
+  // Compared as written, not as parsed: two ids that are not numbers both parse to NaN, and NaN
+  // read as equal made every later activity look like a retry of the one before it.
+  if (!ordered && a.id !== b.id) return false
+  return Number.isInteger(a.attempt) && Number.isInteger(b.attempt) && a.attempt > b.attempt
+}
+
 const DEFAULT_LIVE_POLL = Duration.seconds(1)
 
 // An operator's override, in milliseconds, for either node. Read at layer build rather than at
@@ -570,13 +605,64 @@ export const layerWith = (options?: LayerOptions) =>
           .pipe(Effect.orDie)
       }
 
+      // A compare and set, not a write. Two attempts of one activity can be alive at once and they
+      // do not arrive in order: a paused attempt 1 that resumes after attempt 2 has claimed used to
+      // take the log back, and then every publish from attempt 2's tool activities died on the
+      // fence for a step that was going fine.
       function claim(aggregateID: string, ownerID: string) {
-        return db
-          .update(EventSequenceTable)
-          .set({ owner_id: ownerID })
-          .where(eq(EventSequenceTable.aggregate_id, aggregateID))
-          .run()
-          .pipe(Effect.orDie)
+        return Effect.gen(function* () {
+          const row = yield* db
+            .select({ ownerID: EventSequenceTable.owner_id })
+            .from(EventSequenceTable)
+            .where(eq(EventSequenceTable.aggregate_id, aggregateID))
+            .get()
+            .pipe(Effect.orDie)
+          // No sequence row yet, so there is nothing to fence and nothing to lose a race to: the
+          // first publish inserts the row with this owner on it.
+          if (row === undefined) return
+          if (row.ownerID != null && supersededBy(row.ownerID, ownerID)) {
+            yield* Effect.die(
+              new InvalidDurableEventError({
+                type: "session.claim",
+                message: `Stale claim for aggregate ${aggregateID}: held by ${row.ownerID}, claimer ${ownerID}`,
+              }),
+            )
+          }
+          // Conditional on what was just read, because the read and the write are two statements
+          // and over a network store they are two requests. Two attempts of one activity reaching
+          // here together both passed the check above, and an unconditional write let the loser
+          // land last and fence out the winner's tools.
+          yield* db
+            .update(EventSequenceTable)
+            .set({ owner_id: ownerID })
+            .where(
+              and(
+                eq(EventSequenceTable.aggregate_id, aggregateID),
+                row.ownerID == null
+                  ? isNull(EventSequenceTable.owner_id)
+                  : eq(EventSequenceTable.owner_id, row.ownerID),
+              ),
+            )
+            .run()
+            .pipe(Effect.orDie)
+          // Read back rather than trusting a driver-specific affected-row count. Losing means
+          // somebody claimed between the two statements, and a loser that carried on would publish
+          // under a token the fence rejects.
+          const after = yield* db
+            .select({ ownerID: EventSequenceTable.owner_id })
+            .from(EventSequenceTable)
+            .where(eq(EventSequenceTable.aggregate_id, aggregateID))
+            .get()
+            .pipe(Effect.orDie)
+          if (after?.ownerID !== ownerID) {
+            yield* Effect.die(
+              new InvalidDurableEventError({
+                type: "session.claim",
+                message: `Lost the claim for aggregate ${aggregateID}: held by ${after?.ownerID}, claimer ${ownerID}`,
+              }),
+            )
+          }
+        })
       }
 
       const subscribe = <D extends Definition>(definition: D): Stream.Stream<Payload<D>> =>

packages/core/src/session/execution/worktree.ts

@@ -12,9 +12,9 @@ export * as WorktreeMaterializer from "./worktree"
 // hosts still cannot see each other's writes, because those are not captured until the step is
 // sealed. One worker per worktree is what makes a step's tools share a tree.
 
-import { rm, writeFile } from "node:fs/promises"
+import { readdir, rm, writeFile } from "node:fs/promises"
 import path from "path"
-import { Cause, Context, Effect, Layer } from "effect"
+import { Cause, Context, Effect, Layer, Schema } from "effect"
 import { ChildProcess } from "effect/unstable/process"
 import { and, asc, desc, eq } from "drizzle-orm"
 import { Database } from "../../database/database"
@@ -26,25 +26,54 @@ import { Global } from "../../global"
 import { AppProcess } from "../../process"
 import { AbsolutePath } from "../../schema"
 import { SnapshotPackTable } from "../../snapshot/sql"
+import { chainHead, isBehind, orderChain } from "../../snapshot/chain"
 import { readWorktreeTip, writeWorktreeTip } from "../../snapshot/tip"
 
 export interface Interface {
   /**
    * Make sure the session's directory holds the newest state the shared store has for it,
    * rebuilding its worktree from stored snapshot packs when it is missing or behind. A directory
    * with no stored packs, and a tree this host has neither built nor captured from, are left
-   * alone. Never fails the caller.
+   * alone.
+   *
+   * A rebuild that fails dies with `WorktreeMaterializeError` rather than returning: running the
+   * step against whatever is in the directory tells the model those files are the project, and for
+   * a fresh host that is nothing at all. Tagged so the activity boundary retries it elsewhere.
+   *
+   * `pauseBeforeLock` waits between reading the directory and taking the lock. Zero everywhere but
+   * the check that reproduces what a concurrent drain does in that gap: nothing outside this module
+   * can hold a caller there, and what the check asserts is the real outcome, whether a failed
+   * rebuild removes a directory this call did not create.
    */
-  readonly ensure: (directory: string) => Effect.Effect<void>
+  readonly ensure: (
+    directory: string,
+    options?: { readonly pauseBeforeLock?: number },
+  ) => Effect.Effect<void>
 }
 
 export class Service extends Context.Service<Service, Interface>()(
   "@opencode/v2/WorktreeMaterializer",
 ) {}
 
-// HEAD of a rebuilt tree, which doubles as the mark that says the tree is ours to move.
+// HEAD of a rebuilt tree, so the rebuilt repo reads as a clean checkout rather than an unborn
+// branch over a full untracked tree.
 const RESTORED = "refs/heads/opencode-restore"
 
+/** A rebuild that did not finish. Tagged so the boundary can tell it from a refusal and retry it. */
+export class WorktreeMaterializeError extends Schema.TaggedErrorClass<WorktreeMaterializeError>()(
+  "WorktreeMaterializer.MaterializeError",
+  { message: Schema.String },
+) {}
+
+// Nothing in it at all, so there is no work to protect and nothing to lose by checking a tree out
+// over it. Unreadable counts as not empty: a directory we cannot look into is not one to overwrite.
+const isEmptyDir = (dir: string) =>
+  Effect.promise(() =>
+    readdir(dir)
+      .then((entries) => entries.length === 0)
+      .catch(() => false),
+  )
+
 const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
@@ -61,13 +90,15 @@ const layer = Layer.effect(
         .create({ worktree, gitDirectory: AbsolutePath.make(path.join(worktree, ".git")) })
         .pipe(Effect.orDie)
       // Index every pack shipped for this worktree; objects accumulate, the newest tree wins.
-      const rows = yield* db
+      const stored = yield* db
         .select()
         .from(SnapshotPackTable)
         .where(eq(SnapshotPackTable.worktree, tip.worktree))
-        .orderBy(asc(SnapshotPackTable.time_created))
         .all()
         .pipe(Effect.orDie)
+      // A pack cannot be indexed before the one it was built on, and the write clock does not order
+      // them: two hosts disagree about the time, and one behind puts its pack first.
+      const rows = orderChain(stored)
       const packDirectory = path.join(repository.gitDirectory, "objects", "pack")
       yield* fs.ensureDir(packDirectory).pipe(Effect.orDie)
       for (const row of rows) {
@@ -120,82 +151,80 @@ const layer = Layer.effect(
     const behind = Effect.fnUntraced(function* (tip: typeof SnapshotPackTable.$inferSelect) {
       const held = yield* readWorktreeTip(global.data, tip.worktree)
       if (!held || held === tip.tree) return false
-      const shipped = yield* db
-        .select({ time: SnapshotPackTable.time_created })
+      const rows = yield* db
+        .select()
         .from(SnapshotPackTable)
-        .where(and(eq(SnapshotPackTable.worktree, tip.worktree), eq(SnapshotPackTable.tree, held)))
-        .orderBy(desc(SnapshotPackTable.time_created))
-        .limit(1)
-        .get()
+        .where(eq(SnapshotPackTable.worktree, tip.worktree))
+        .all()
         .pipe(Effect.orDie)
-      return shipped !== undefined && shipped.time < tip.time_created
+      return isBehind(rows, held)
     })
 
-    // Whether this tree is one we built from packs. A checkout the host already had is somebody's
-    // working copy: reading its captures is fine, but checking a stored tree out over it would
-    // rewrite files and HEAD under whoever owns it.
-    const rebuilt = (worktree: string) =>
-      proc
-        .run(
-          ChildProcess.make(
-            "git",
-            [
-              "--git-dir",
-              path.join(worktree, ".git"),
-              "rev-parse",
-              "--verify",
-              "--quiet",
-              RESTORED,
-            ],
-            { cwd: worktree, extendEnv: true },
-          ),
-        )
-        .pipe(
-          Effect.map((result) => result.exitCode === 0),
-          Effect.catchCause(() => Effect.succeed(false)),
-        )
-
-    const ensure = Effect.fn("WorktreeMaterializer.ensure")(function* (directory: string) {
+    const ensure = Effect.fn("WorktreeMaterializer.ensure")(function* (
+      directory: string,
+      options?: { readonly pauseBeforeLock?: number },
+    ) {
       // The newest capture whose session ran in this directory decides which worktree to rebuild,
       // and which state a tree that is already here has to be brought to.
-      const tip = yield* db
-        .select()
-        .from(SnapshotPackTable)
-        .where(eq(SnapshotPackTable.directory, directory))
-        .orderBy(desc(SnapshotPackTable.time_created))
-        .limit(1)
-        .get()
-        .pipe(Effect.orDie)
+      const tip = chainHead(
+        yield* db
+          .select()
+          .from(SnapshotPackTable)
+          .where(eq(SnapshotPackTable.directory, directory))
+          .all()
+          .pipe(Effect.orDie),
+      )
       if (!tip) return
-      const present = yield* fs.existsSafe(tip.worktree)
-      if (present) {
-        if (!(yield* behind(tip))) return
-        if (!(yield* rebuilt(tip.worktree))) {
-          yield* Effect.logWarning("worktree is behind the store and was not built from it", {
-            worktree: tip.worktree,
-            tree: tip.tree,
-          })
-          return
-        }
-      }
+      // An empty directory is not somebody's working copy, so the rule that protects one does not
+      // apply to it. Treating it as present is what stops a fresh host from ever building the tree:
+      // it has no tip note, so `behind` says no, and the tools then run against nothing. A mounted
+      // path that exists but holds nothing is the ordinary shape of a host that has never seen this
+      // project, which is exactly the case the packs are for.
+      const present = (yield* fs.existsSafe(tip.worktree)) && !(yield* isEmptyDir(tip.worktree))
+      // `behind` is already the whole rule. It is false unless this host has a note of its own, and
+      // a note means this host agreed to that state: either it built the tree from packs or it
+      // captured the tree from here. Moving it forward from a state it agreed to loses nothing.
+      //
+      // What used to gate this as well was whether the tree carried the marker `materialize`
+      // writes. Only a rebuilt tree ever has that, so a host that seeded the session from its own
+      // checkout never did, and once any other host shipped, every activity that host drew died
+      // here. A developer's checkout is protected by having no note at all, not by the marker.
+      if (present && !(yield* behind(tip))) return
+      if (options?.pauseBeforeLock) yield* Effect.sleep(options.pauseBeforeLock)
       yield* locks.withLock(tip.worktree)(
         Effect.gen(function* () {
-          // Re-check inside the lock: a concurrent drain may have done this already.
-          if ((yield* fs.existsSafe(tip.worktree)) && !(yield* behind(tip))) return
+          // Re-check inside the lock: a concurrent drain may have done this already. Same notion of
+          // present as above, or an empty directory bails out here instead and the tree that the
+          // outer check just decided to build never gets built.
+          const here =
+            (yield* fs.existsSafe(tip.worktree)) && !(yield* isEmptyDir(tip.worktree))
+          if (here && !(yield* behind(tip))) return
           yield* materialize(tip).pipe(
             Effect.catchCauseIf(
               (cause) => !Cause.hasInterrupts(cause),
               (cause) =>
                 Effect.gen(function* () {
                   // A half-built tree would pass the exists check forever, so what we created is
                   // removed. A tree that was already here is not ours to remove: a failed refresh
-                  // leaves it as stale as it was.
-                  if (!present)
+                  // leaves it as stale as it was. Asked of the reading taken inside the lock, which
+                  // is the only one that describes the directory this attempt started from: the
+                  // outer one is why the re-check exists, and a drain that materialized while this
+                  // one waited makes it name a directory that no longer exists.
+                  if (!here)
                     yield* Effect.promise(() => rm(tip.worktree, { recursive: true, force: true }))
-                  yield* Effect.logWarning("failed to materialize worktree", {
+                  yield* Effect.logError("failed to materialize worktree", {
                     worktree: tip.worktree,
                     cause,
                   })
+                  // Swallowing this ran the step against whatever was in the directory, which for
+                  // a fresh host is nothing at all. Tagged, so the activity boundary can retry it:
+                  // git and the filesystem fail for reasons that pass, and the alternative is a
+                  // turn that fails for good because one worker had a bad minute.
+                  return yield* Effect.die(
+                    new WorktreeMaterializeError({
+                      message: `could not materialize ${tip.worktree} at ${tip.tree}`,
+                    }),
+                  )
                 }),
             ),
           )

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

@@ -39,10 +39,21 @@ export interface StepResult {
 /** A tool call the provider asked for, recorded but not run, handed to the caller to dispatch.
  * Every id comes from the provider or the publisher and is carried, never regenerated: a second run
  * of the same step would mint different ones and the results would not match the log. */
+/** A call the model asked for, handed to whoever will run it.
+ *
+ * It names the call rather than carrying it. The arguments are already in the log when this is
+ * handed over: the streaming path ends the input fragment before it defers, so `Tool.Input.Ended`
+ * lands on the deferred path too and the dispatcher reads them off the pending call. Carrying them
+ * as well put them across a durable boundary twice, once as the model call's result and once as the
+ * tool call's input, so a step with large `write` bodies wrote them into history twice and a big
+ * enough one passed the payload limit.
+ *
+ * Taking them off failed once, and the reason is worth keeping: what the record holds is the raw
+ * JSON *string*, and handing that straight to a tool gave every one of them a string where its
+ * schema wanted an object. The dispatcher parses it. */
 export interface DeferredToolCall {
   readonly id: string
   readonly name: string
-  readonly input: unknown
   readonly assistantMessageID: string
 }