feat(temporal): Split a step into a model call, one activity per tool call, and a seal.

packages/core/src/event.ts

@@ -3,7 +3,7 @@ export * as EventV2 from "./event"
 import { Cause, Context, Duration, Effect, Layer, Option, PubSub, Queue, Schema, 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"
@@ -184,6 +184,39 @@ export interface LayerOptions {
   readonly livePollInterval?: Duration.Input
 }
 
+// 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, which
+  // reads as equal and would make 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
+}
+
 /** 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`. */
@@ -568,13 +601,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 must not
+      // take the log back, or every publish from attempt 2's tool activities dies on the fence for
+      // a step that is 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 pass the check above, and an unconditional write would 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/message-updater.ts

@@ -208,6 +208,10 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
       },
       "session.next.step.ended": (event) => {
         return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
+          // A step's writers share one owner token, so two seal attempts are both admitted. The
+          // tool cases guard on status; this one has to guard on the step already being closed, or
+          // a late attempt overwrites the end snapshot and file diff with a different instant.
+          if (draft.time.completed) return
           draft.time.completed = event.data.timestamp
           draft.finish = event.data.finish
           draft.cost = event.data.cost

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

@@ -6,6 +6,7 @@ import { SessionSchema } from "../schema"
 import type { ContextSnapshotDecodeError, MessageDecodeError, SessionRunDeclinedError } from "../error"
 import type { SessionInput } from "../input"
 import { SessionRunnerModel } from "./model"
+import type { StepSettlement } from "./publish-llm-event"
 import type { SystemContext } from "../../system-context/index"
 import type { ToolOutputStore } from "../../tool-output-store"
 
@@ -34,6 +35,73 @@ export interface StepResult {
   readonly promotion: SessionInput.Delivery | undefined
 }
 
+/** 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. The arguments stay
+ * off it, because the record already holds them as the provider's raw JSON string, which the
+ * dispatcher parses. */
+export interface DeferredToolCall {
+  readonly id: string
+  readonly name: string
+  readonly assistantMessageID: string
+}
+
+/** Closing one step whose tools have been dispatched. */
+export interface SealStepInput {
+  readonly sessionID: SessionSchema.ID
+  readonly step: number
+  /** The provider's finish reason and token counts, from the attempt that opened this step. They
+   * live only in that process's memory, so they have to be carried here rather than read back. */
+  readonly settlement?: StepSettlement
+  /** The message the attempt opened. Absent only for a step that produced nothing to close. */
+  readonly assistantMessageID?: string
+  /** Whether the turn should keep going, decided by the attempt. Only the attempt knows it hit a
+   * 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
+}
+
+/** One recorded tool call, to run on its own. */
+export interface ToolCallInput {
+  readonly sessionID: SessionSchema.ID
+  readonly call: DeferredToolCall
+}
+
+/** How a dispatch ended, for the executor's own record: the log says what the model was told, not
+ * whether nothing ran (`already-settled`), the tool could not be settled (`failed`), or a repeat was
+ * refused because the tool does not declare itself repeatable (`unknown`). */
+export type ToolCallOutcome = "settled" | "already-settled" | "failed" | "unknown"
+
+export interface ToolCallResult {
+  readonly outcome: ToolCallOutcome
+}
+
+/** What one provider attempt produced. `calls` is empty unless the caller deferred dispatch, in
+ * which case the step is left open and `settlement` is what seals it. */
+export interface TurnAttemptResult {
+  readonly needsContinuation: boolean
+  readonly step: number
+  readonly calls: ReadonlyArray<DeferredToolCall>
+  readonly settlement?: StepSettlement
+  /** The assistant message this attempt opened, minted here so a seal in another process closes the
+   * right one even when the turn published no content of its own. */
+  readonly assistantMessageID?: string
+}
+
+/** What a model-only attempt produced. `settled` means the step is already over (a crashed step was
+ * finalized from the log, or the recovery gate found no work) and there is nothing to dispatch.
+ * `called` hands back the recorded calls plus the settlement whoever seals the step will need. */
+export type ModelCallResult =
+  | { readonly kind: "settled"; readonly result: StepResult }
+  | {
+      readonly kind: "called"
+      readonly step: number
+      readonly calls: ReadonlyArray<DeferredToolCall>
+      readonly settlement?: StepSettlement
+      readonly assistantMessageID?: string
+      readonly needsContinuation?: boolean
+    }
+
 /** Runs one local continuation from already-recorded Session history. */
 export interface Interface {
   /** Drains eligible durable work. Explicit runs perform one provider attempt even when no work is eligible. */
@@ -44,6 +112,23 @@ export interface Interface {
   /** Run exactly one step and report the next loop state, so a caller (e.g. a Temporal workflow)
    * can drive the turn one step at a time. Mirrors one iteration of `run`'s loop. */
   readonly runStep: (input: StepInput) => Effect.Effect<StepResult, RunError>
+  /** Run the provider attempt of one step and stop, handing back the tool calls it asked for instead
+   * of running them. The caller dispatches each one and then seals the step. This is what puts the
+   * model-to-tools loop in a durable executor's hands rather than inside a single activity. */
+  readonly runModelCall: (input: StepInput) => Effect.Effect<ModelCallResult, RunError>
+  /** Run one recorded tool call and publish its result. Safe to call twice for the same call: the
+   * second sees the settled result and does nothing, and a call whose first dispatch died after it
+   * started is run again only if the tool declares itself repeatable. */
+  readonly runToolCall: (input: ToolCallInput) => Effect.Effect<ToolCallResult, RunError>
+  /** Close a call a stop cut short, so it does not sit in the log as running until the next turn.
+   * A whole step closes the tools it opened on its way out; a dispatch that is its own unit of work
+   * has to be told. A call that never started, and one that already settled, are left alone. */
+  readonly failToolCall: (input: ToolCallInput) => Effect.Effect<void, RunError>
+  /** Close a step once its calls have been dispatched: snapshot, file diff, Step.Ended, and the
+   * loop decision. Safe to call twice: the second sees the step already closed and returns the same
+   * answer without publishing again. The carried message id is what keeps a retried seal from
+   * closing a different step. */
+  readonly sealStep: (input: SealStepInput) => Effect.Effect<StepResult, RunError>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/v2/SessionRunner") {}