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

packages/core/package.json

@@ -40,15 +40,6 @@
     }
   },
   "devDependencies": {
-    "@opencode-ai/http-recorder": "workspace:*",
-    "@parcel/watcher-darwin-arm64": "2.5.1",
-    "@parcel/watcher-darwin-x64": "2.5.1",
-    "@parcel/watcher-linux-arm64-glibc": "2.5.1",
-    "@parcel/watcher-linux-arm64-musl": "2.5.1",
-    "@parcel/watcher-linux-x64-glibc": "2.5.1",
-    "@parcel/watcher-linux-x64-musl": "2.5.1",
-    "@parcel/watcher-win32-arm64": "2.5.1",
-    "@parcel/watcher-win32-x64": "2.5.1",
     "@tsconfig/bun": "catalog:",
     "@types/bun": "catalog:",
     "@types/cross-spawn": "catalog:",
@@ -58,6 +49,15 @@
     "@types/semver": "catalog:",
     "@types/turndown": "5.0.5",
     "@types/which": "3.0.4",
+    "@parcel/watcher-darwin-arm64": "2.5.1",
+    "@parcel/watcher-darwin-x64": "2.5.1",
+    "@parcel/watcher-linux-arm64-glibc": "2.5.1",
+    "@parcel/watcher-linux-arm64-musl": "2.5.1",
+    "@parcel/watcher-linux-x64-glibc": "2.5.1",
+    "@parcel/watcher-linux-x64-musl": "2.5.1",
+    "@parcel/watcher-win32-arm64": "2.5.1",
+    "@parcel/watcher-win32-x64": "2.5.1",
+    "@opencode-ai/http-recorder": "workspace:*",
     "drizzle-kit": "catalog:"
   },
   "dependencies": {
@@ -85,23 +85,23 @@
     "@effect/opentelemetry": "catalog:",
     "@effect/platform-node": "catalog:",
     "@effect/sql-sqlite-bun": "catalog:",
+    "@lydell/node-pty": "catalog:",
     "@ff-labs/fff-bun": "0.9.4",
     "@libsql/client": "^0.17.0",
-    "@lydell/node-pty": "catalog:",
     "@npmcli/arborist": "9.4.0",
     "@npmcli/config": "10.8.1",
     "@opencode-ai/effect-drizzle-sqlite": "workspace:*",
     "@opencode-ai/effect-sqlite-node": "workspace:*",
     "@opencode-ai/llm": "workspace:*",
-    "@opencode-ai/plugin": "workspace:*",
     "@opencode-ai/schema": "workspace:*",
-    "@openrouter/ai-sdk-provider": "2.9.0",
+    "@opencode-ai/plugin": "workspace:*",
     "@opentelemetry/api": "1.9.0",
     "@opentelemetry/context-async-hooks": "2.6.1",
     "@opentelemetry/exporter-trace-otlp-http": "0.214.0",
     "@opentelemetry/sdk-trace-base": "2.6.1",
     "@parcel/watcher": "2.5.1",
     "@silvia-odwyer/photon-node": "0.3.4",
+    "@openrouter/ai-sdk-provider": "2.9.0",
     "ai-gateway-provider": "3.2.0",
     "bun-pty": "0.4.8",
     "cross-spawn": "catalog:",
@@ -114,8 +114,8 @@
     "google-auth-library": "10.5.0",
     "gray-matter": "4.0.3",
     "htmlparser2": "8.0.2",
-    "ignore": "7.0.5",
     "immer": "11.1.4",
+    "ignore": "7.0.5",
     "jsonc-parser": "3.3.1",
     "mime-types": "3.0.2",
     "minimatch": "10.2.5",

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,39 @@ 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`. */
+// Same-run tokens order retries and generated activity IDs. Cross-run age needs a separate epoch.
+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 +603,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>> =>