From c5cfa3b96cd2f2cfd733050456caccc10a42e09c Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Mon, 14 Sep 2026 17:21:23 -0700 Subject: [PATCH] Split a step into a model call, one activity per tool call, and a seal. A whole-step activity holds the provider attempt and every tool under one timeout and one retry policy, so a tool waiting on a human or hanging takes the attempt with it. Behind `OPENCODE_TEMPORAL_STEPPED=1` the workflow drives the three units itself, under one owner token per step, with the tool arguments read off the log and a stop closing the step it cut short. --- packages/core/src/event.ts | 98 ++- packages/core/src/session/message-updater.ts | 4 + packages/core/src/session/runner/index.ts | 85 ++ packages/core/src/session/runner/llm.ts | 347 +++++++- .../src/session/runner/publish-llm-event.ts | 29 +- packages/core/test/event-claim.test.ts | 131 +++ packages/core/test/event.test.ts | 74 ++ .../core/test/lib/execution-conformance.ts | 133 ++- packages/core/test/lib/runner-harness.ts | 16 +- .../test/session-runner-model-call.test.ts | 794 ++++++++++++++++++ packages/temporal/README.md | 51 ++ packages/temporal/src/activities.ts | 103 ++- packages/temporal/src/boundary.ts | 82 ++ packages/temporal/src/config.ts | 33 +- packages/temporal/src/drain.ts | 120 +-- packages/temporal/src/executor.ts | 34 +- packages/temporal/src/protocol.ts | 5 + packages/temporal/src/run-error-codec.ts | 4 - packages/temporal/src/stepped-drain.ts | 129 +++ packages/temporal/src/stepped-turn.ts | 148 ++++ packages/temporal/src/workflow.ts | 46 +- ...ession-execution-temporal-contract.test.ts | 12 +- packages/temporal/test/stepped-turn.test.ts | 249 ++++++ 23 files changed, 2581 insertions(+), 146 deletions(-) create mode 100644 packages/core/test/event-claim.test.ts create mode 100644 packages/core/test/session-runner-model-call.test.ts create mode 100644 packages/temporal/src/boundary.ts create mode 100644 packages/temporal/src/stepped-drain.ts create mode 100644 packages/temporal/src/stepped-turn.ts create mode 100644 packages/temporal/test/stepped-turn.test.ts diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index b544531e51b5..e171463e12ab 100644 --- a/packages/core/src/event.ts +++ b/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 = (definition: D): Stream.Stream> => diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 46118a89fe4b..de50688b748a 100644 --- a/packages/core/src/session/message-updater.ts +++ b/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 diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index 1a58cd1865d0..9be39946695a 100644 --- a/packages/core/src/session/runner/index.ts +++ b/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 + 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 + 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 + /** 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 + /** 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 + /** 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 + /** 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 } export class Service extends Context.Service()("@opencode/v2/SessionRunner") {} diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index a9165f57f762..03501c828882 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -26,21 +26,33 @@ import { ToolRegistry } from "../../tool/registry" import { ToolOutputStore } from "../../tool-output-store" import { SessionContextEpoch } from "../context-epoch" import { SessionCompaction } from "../compaction" +import { SessionRunDeclinedError } from "../error" import { SessionEvent } from "../event" import { SessionHistory } from "../history" import { SessionInput } from "../input" import { SessionMessage } from "../message" import { SessionSchema } from "../schema" import { SessionStore } from "../store" -import { type RunError, Service, type StepInput } from "./index" +import { + type DeferredToolCall, + type ModelCallResult, + type RunError, + type SealStepInput, + type StepInput, + type ToolCallInput, + type ToolCallResult, + type TurnAttemptResult, + Service, +} from "./index" import { SessionRunnerModel } from "./model" -import { createLLMEventPublisher, emitToolResult } from "./publish-llm-event" +import { createLLMEventPublisher, emitToolResult, record } from "./publish-llm-event" import { toLLMMessages } from "./to-llm-message" import { MAX_STEPS_PROMPT } from "./max-steps" import { DEFAULT_MAX_STEPS, REPEAT_LIMIT, REPEATED_CALLS_PROMPT, trailingIdenticalToolSteps } from "./loop-guard" import { Snapshot } from "../../snapshot" import { SnapshotSync } from "../../snapshot-sync" import { makeLocationNode } from "../../effect/app-node" +import { KeyedMutex } from "../../effect/keyed-mutex" import { llmClient } from "../../effect/app-node-platform" /** @@ -93,6 +105,13 @@ import { llmClient } from "../../effect/app-node-platform" * explicit loop starts the next provider turn after local settlement. Configured agent step limits bound the loop. */ +// Shipping the tree, one at a time per directory. Two tools of one step run at once and both end by +// capturing and pushing: a capture writes the git index and a push compares against the store's +// head, so two of them in one directory race on both, and the loser's work is refused rather than +// shipped. Module-level, because what has to be excluded is two activities in one process, and each +// builds its own runner. +const shipping = KeyedMutex.makeUnsafe() + const layer = Layer.effect( Service, Effect.gen(function* () { @@ -201,6 +220,9 @@ const layer = Layer.effect( promotion: SessionInput.Delivery | undefined, step: number, recoverOverflow?: typeof compaction.compactAfterOverflow, + // When set, tool calls are recorded and returned instead of run, and the step is left open for + // whoever runs them. This is what lets a durable executor make each call its own unit of work. + deferTools = false, ) { const session = yield* getSession(sessionID) if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID) @@ -208,6 +230,7 @@ const layer = Layer.effect( const agent = yield* agents.select(session.agent) const initialized = yield* SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id) const toolFibers = yield* FiberSet.make() + const deferred: DeferredToolCall[] = [] let needsContinuation = false let currentStep = step if (promotion) { @@ -267,6 +290,7 @@ const layer = Layer.effect( ...(session.model?.variant === undefined ? {} : { variant: session.model.variant }), }, snapshot: startSnapshot, + deferCalls: deferTools, }) const withPublication = Semaphore.makeUnsafe(1).withPermit const publish = (event: LLMEvent, outputPaths: ReadonlyArray = []) => @@ -290,6 +314,13 @@ const layer = Layer.effect( } needsContinuation = true const assistantMessageID = yield* publisher.assistantMessageID(event.id) + // Tool.Called is already durable (the publish above), so handing the call back is enough + // for the caller to run it later. Nothing forks here, which is why the step ends when the + // stream does and the overlap between the model and its tools is lost. + if (deferTools) { + deferred.push({ id: event.id, name: event.name, assistantMessageID }) + return + } yield* Effect.uninterruptibleMask((restore) => restore( toolMaterialization.settle({ @@ -357,7 +388,9 @@ const layer = Layer.effect( yield* withPublication(publisher.failUnsettledTools(`Tool execution failed: ${message}`)) } const stepSettlement = publisher.stepSettlement() - if (stepSettlement && !publisher.hasProviderError()) { + // Deferred tools have not run yet, so the step is not over and the end snapshot would be + // taken before their side effects. Whoever runs them seals it, with the settlement below. + 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) @@ -387,7 +420,23 @@ const layer = Layer.effect( if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause) if (settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)) return yield* Effect.failCause(settled.cause) - return { needsContinuation: !publisher.hasProviderError() && needsContinuation, step: currentStep } + // A provider turn can finish having published nothing at all (no text, no tool call), and + // then no assistant message exists for the seal to close. The whole-step path never hits + // this because it mints one right here, inside Step.Ended. Mint it the same way and carry + // the id, rather than leave the seal to guess from the projection. + const assistantMessageID = + deferTools && stepSettlement && !publisher.hasProviderError() + ? yield* withPublication(publisher.startAssistant()) + : undefined + return { + needsContinuation: !publisher.hasProviderError() && needsContinuation, + step: currentStep, + // A provider error already failed every recorded call, so handing them back would only + // buy a dispatch that reads them as settled. + calls: publisher.hasProviderError() ? [] : deferred, + settlement: stepSettlement, + assistantMessageID, + } }), ) }, Effect.scoped) @@ -395,31 +444,32 @@ const layer = Layer.effect( sessionID: SessionSchema.ID, promotion: SessionInput.Delivery | undefined, step: number, - ) => Effect.Effect<{ readonly needsContinuation: boolean; readonly step: number }, RunError> + deferTools?: boolean, + ) => Effect.Effect - const runAfterOverflowCompaction: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) { - return yield* runTurnAttempt(sessionID, promotion, step).pipe( + const runAfterOverflowCompaction: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step, deferTools) { + return yield* runTurnAttempt(sessionID, promotion, step, undefined, deferTools).pipe( Effect.catchDefect( Effect.fnUntraced(function* (defect) { if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect) if (defect.transition._tag === "ContinueAfterOverflowCompaction") return yield* Effect.die("Post-compaction provider attempt cannot recover another overflow") yield* Effect.yieldNow - return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step) + return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step, deferTools) }), ), ) }) - const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) { - return yield* runTurnAttempt(sessionID, promotion, step, compaction.compactAfterOverflow).pipe( + const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step, deferTools) { + return yield* runTurnAttempt(sessionID, promotion, step, compaction.compactAfterOverflow, deferTools).pipe( Effect.catchDefect( Effect.fnUntraced(function* (defect) { if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect) yield* Effect.yieldNow if (defect.transition._tag === "ContinueAfterOverflowCompaction") - return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step) - return yield* runTurn(sessionID, undefined, defect.transition.step) + return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step, deferTools) + return yield* runTurn(sessionID, undefined, defect.transition.step, deferTools) }), ), ) @@ -433,8 +483,7 @@ const layer = Layer.effect( const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue") const entryContext = yield* getContext(input.sessionID) const recover = - !input.force && !hasSteer && !hasQueue && - (yield* hasRecoverableWork(input.sessionID, entryContext)) + !input.force && !hasSteer && !hasQueue && (yield* hasRecoverableWork(input.sessionID, entryContext)) if (!input.force && !hasSteer && !hasQueue && !recover) return yield* failInterruptedTools(input.sessionID, entryContext) let promotion: SessionInput.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined @@ -486,12 +535,8 @@ const layer = Layer.effect( (message): message is SessionMessage.Assistant => message.type === "assistant" && !message.time.completed, ) if (!inFlight) return undefined - const toolParts = inFlight.content.filter( - (part): part is SessionMessage.AssistantTool => part.type === "tool", - ) - const dispatched = toolParts.some( - (part) => part.state.status === "running" || part.state.status === "completed", - ) + const toolParts = inFlight.content.filter((part): part is SessionMessage.AssistantTool => part.type === "tool") + const dispatched = toolParts.some((part) => part.state.status === "running" || part.state.status === "completed") if (!dispatched) return undefined // A tool declared idempotent (a pure read) has no external side effect, so it is safe to // re-run: re-settle it for a real result instead of failing it. Everything else still open is @@ -551,16 +596,17 @@ const layer = Layer.effect( return yield* stepContinuation(input.sessionID, localTools, input.step) }) - // One iteration of `run`'s loop, exposed so a Temporal workflow can drive the turn one step at - // a time (each step = one runTurn = one provider attempt + its tools). Semantics match `run`. - const runStep = Effect.fn("SessionRunner.runStep")(function* (input: StepInput) { + // The checks every step runs before the provider is called. Shared by the whole-step path and + // the model-only path so the two cannot drift apart. Returns the terminal result when the step + // is already over, otherwise the promotion the turn should apply. + const stepPrologue = Effect.fn("SessionRunner.stepPrologue")(function* (input: StepInput) { // One projected-history load serves all the entry checks; nothing mutates the projection // between them. The turn itself reloads after the first mutation. const entryContext = yield* getContext(input.sessionID) // Re-drive of a crashed step: finalize it from the log rather than re-calling the model and // re-running its already-dispatched tools. const resumed = yield* resumeCrashedStep(input, entryContext) - if (resumed) return resumed + if (resumed) return { kind: "settled", result: resumed } as const let promotion = input.promotion if (input.first) { const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer") @@ -568,13 +614,11 @@ const layer = Layer.effect( // Same recovery gate as `run`: a first-step retry whose prompt was already promoted (and // whose crash predates any tool dispatch, so resumeCrashedStep had nothing to finalize) // must re-stream, not no-op. - if ( - !input.force && - !hasSteer && - !hasQueue && - !(yield* hasRecoverableWork(input.sessionID, entryContext)) - ) - return { continue: false, step: input.step, promotion: undefined } + if (!input.force && !hasSteer && !hasQueue && !(yield* hasRecoverableWork(input.sessionID, entryContext))) + return { + kind: "settled", + result: { continue: false, step: input.step, promotion: undefined }, + } as const promotion = hasSteer ? "steer" : hasQueue ? "queue" : undefined } // Close tools left pending/running by an interrupted attempt before every turn, not just the @@ -582,13 +626,252 @@ const layer = Layer.effect( // re-stream a request with a dangling tool_use and no tool_result, which the provider rejects // -- a retry poison loop. This is a no-op on a healthy step (the prior step settled its tools). yield* failInterruptedTools(input.sessionID, entryContext) - const result = yield* runTurn(input.sessionID, promotion, input.step) + return { kind: "run", promotion } as const + }) + + const sealStep = Effect.fn("SessionRunner.sealStep")(function* (input: SealStepInput) { + const context = yield* getContext(input.sessionID) + const inFlight = context.findLast( + (message): message is SessionMessage.Assistant => message.type === "assistant" && !message.time.completed, + ) + // On a retry that lands after Step.Ended was published there is nothing open, but the loop + // decision still has to come out the same, so it is read off the step we just closed. + const carried = input.assistantMessageID + const target = + (carried + ? context.findLast((m): m is SessionMessage.Assistant => m.type === "assistant" && m.id === carried) + : undefined) ?? + inFlight ?? + context.findLast((message): message is SessionMessage.Assistant => message.type === "assistant") + if (!target) return yield* stepContinuation(input.sessionID, false, input.step) + const toolParts = target.content.filter((part): part is SessionMessage.AssistantTool => part.type === "tool") + // A step continues so the model can see its tool results. Provider-executed calls need no + // follow-up turn, so a step holding only those finalizes as a plain stop. + const localTools = toolParts.some((part) => part.provider?.executed !== true) + if (target.time.completed) + return yield* stepContinuation(input.sessionID, input.needsContinuation ?? localTools, input.step) + // A dispatch that failed outright leaves its call open. Close it here, or the next attempt + // 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() + // Ship the post-step tree: this is the state a later step on another host needs. + if (endSnapshot) yield* snapshotSync.push(endSnapshot) + const files = + startSnapshot && endSnapshot + ? yield* snapshots + .files({ from: Snapshot.ID.make(startSnapshot), to: endSnapshot }) + .pipe(Effect.catch(() => Effect.succeed(undefined))) + : undefined + yield* events.publish(SessionEvent.Step.Ended, { + sessionID: input.sessionID, + timestamp: yield* DateTime.now, + assistantMessageID: target.id, + // The attempt's own settlement when the caller carried it; otherwise the same fallback the + // crash-resume path uses, which costs only the metering on that step. + finish: input.settlement?.finish ?? (localTools ? "tool-calls" : "stop"), + cost: 0, + tokens: input.settlement?.tokens ?? { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + snapshot: endSnapshot, + files, + }) + return yield* stepContinuation(input.sessionID, input.needsContinuation ?? localTools, input.step) + }) + + // A point read, not the whole projected history. The call carries the message that owns it, so + // decoding every message to find one part would make a step cost O(session) per tool instead + // of O(1). Message ids are looked up on their own, so the session has to be checked too: a call + // from another session must not resolve here. + const recordedCall = Effect.fnUntraced(function* (input: ToolCallInput) { + const owner = yield* store.message(SessionMessage.ID.make(input.call.assistantMessageID)) + if (owner?.sessionID !== input.sessionID || owner.message.type !== "assistant") return undefined + // findLast to agree with the projector, which writes tool updates the same way. + return owner.message.content.findLast( + (item): item is SessionMessage.AssistantTool => item.type === "tool" && item.id === input.call.id, + ) + }) + + const runToolCall = Effect.fn("SessionRunner.runToolCall")(function* (input: ToolCallInput) { + const session = yield* getSession(input.sessionID) + const assistantMessageID = SessionMessage.ID.make(input.call.assistantMessageID) + const part = yield* recordedCall(input) + // The call has to be in the log already: the attempt that produced it recorded its input + // before handing it over. Missing means the log moved under us (a fence), and running a tool + // whose call is not recorded would leave an orphan result. + if (!part || part.type !== "tool") + return yield* Effect.die(`Tool call ${input.call.id} is not recorded on session ${input.sessionID}`) + // At-least-once: a duplicate dispatch landing after the result did must not run anything. + if (part.state.status !== "pending" && part.state.status !== "running") + return { outcome: "already-settled" } satisfies ToolCallResult + const agent = yield* agents.select(session.agent) + const materialization = yield* tools.materialize(agent.info?.permissions) + // `running` means a dispatch already published Tool.Called and was about to run the tool, so + // the side effect may have happened and nothing that reads the log afterwards can tell. Only + // a tool that declares itself repeatable runs again. The rest are reported unknown and the + // model decides, because re-running the `git push` that may already have landed is the worse + // failure. The attempt number would say the same thing far less precisely: it counts every + // way a dispatch can die, including the ones that never reached the tool. + if (part.state.status === "running" && !materialization.idempotent(input.call.name)) { + yield* events.publish(SessionEvent.Tool.Failed, { + sessionID: input.sessionID, + timestamp: yield* DateTime.now, + assistantMessageID, + callID: input.call.id, + error: { type: "unknown", message: "The outcome of this tool call is unknown" }, + provider: { executed: false }, + }) + return { outcome: "unknown" } satisfies ToolCallResult + } + // The arguments, off the log rather than off the hand-off. A pending call holds the provider's + // raw JSON text, which is what the stream delivered; a re-dispatch of a running one reads the + // object the first dispatch recorded. A defect either way if the text is not JSON, because + // only the recording path could have written that, and a tool handed a string it cannot parse + // reports a wrong reason to the model. + const recorded = part.state + const args = + recorded.status === "pending" + ? yield* Effect.try({ + try: () => JSON.parse(recorded.input) as unknown, + catch: () => + new Error( + recorded.input === "" + ? `Tool call ${input.call.id} has no recorded input to run it with` + : `Tool call ${input.call.id} has a recorded input that is not JSON`, + ), + }).pipe(Effect.orDie) + : recorded.input + // The durable record that this call is being run, published before the tool can do anything. + // It is also the last point a fenced dispatch dies at: under a superseded owner this publish + // fails and the tool never runs, instead of running and losing its result. + yield* events.publish( + SessionEvent.Tool.Called, + { + sessionID: input.sessionID, + timestamp: yield* DateTime.now, + assistantMessageID, + callID: input.call.id, + tool: input.call.name, + input: record(args), + // Deferred calls are never provider-executed: those are filtered out before the hand-off. + provider: { executed: false }, + }, + materialization.idempotent(input.call.name) + ? undefined + : { + // A shared step owner cannot distinguish overlapping dispatches of one call. + id: EventV2.ID.make(`evt_dispatch_${assistantMessageID}_${input.call.id}`), + }, + ) + const settlement = yield* materialization + .settle({ + sessionID: input.sessionID, + agent: agent.id, + assistantMessageID, + call: LLMEvent.toolCall({ + id: input.call.id, + name: input.call.name, + input: args, + }), + }) + .pipe( + // The tool itself ran: what failed is storing its output. Letting that fail the dispatch + // would retry a side effect that already happened and finally tell the model the call was + // interrupted, which is not what happened to it. The whole-step path reports the same + // reason the same way. A decline is a defect, so it passes through this untouched. + Effect.catch((error) => Effect.succeed({ failure: error })), + // A decline is the user stopping the turn, not a tool that failed. It is named here + // rather than raised as a bare interrupt, so nothing downstream has to infer from the + // absence of a cancellation what the user meant. + Effect.catchCause((cause) => + isUserDeclined(cause) + ? Effect.fail(new SessionRunDeclinedError({ sessionID: input.sessionID })) + : Effect.failCause(cause), + ), + ) + if ("failure" in settlement) { + const reason = settlement.failure instanceof Error ? settlement.failure.message : String(settlement.failure) + yield* Effect.uninterruptible( + events.publish(SessionEvent.Tool.Failed, { + sessionID: input.sessionID, + timestamp: yield* DateTime.now, + assistantMessageID, + callID: input.call.id, + error: { type: "unknown", message: `Tool execution failed: ${reason}` }, + provider: { executed: false }, + }), + ) + return { outcome: "failed" } satisfies ToolCallResult + } + // The tool has run by here, so losing the result to an interrupt would hide a + // side effect that already happened. + yield* Effect.uninterruptible( + emitToolResult(events, { + sessionID: input.sessionID, + assistantMessageID, + callID: input.call.id, + result: settlement.result, + output: settlement.output, + outputPaths: settlement.outputPaths, + // Deferred calls are never provider-executed: those are filtered out before the hand-off. + provider: { executed: false }, + }), + ) + // Shipped from the host that ran the tool, because it is the only one holding what the tool + // did. The seal can land anywhere, and a capture there would ship a tree that never saw this + // write. Best effort in the same sense the seal's is: the result is already durable, and a + // pack that does not reach the store costs the next host a rebuild from further back. + 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) + }), + ) + return { outcome: "settled" } as ToolCallResult + }) + + const failToolCall = Effect.fn("SessionRunner.failToolCall")(function* (input: ToolCallInput) { + const part = yield* recordedCall(input) + // Only a call a dispatch had started is closed. One that never reached its tool is left + // pending for the next turn's entry check, and a settled one keeps the result it earned. + if (part?.state.status !== "running") return + yield* events.publish(SessionEvent.Tool.Failed, { + sessionID: input.sessionID, + timestamp: yield* DateTime.now, + assistantMessageID: SessionMessage.ID.make(input.call.assistantMessageID), + callID: input.call.id, + error: { type: "unknown", message: "Tool execution interrupted" }, + provider: { executed: false }, + }) + }) + + const runStep = Effect.fn("SessionRunner.runStep")(function* (input: StepInput) { + const prologue = yield* stepPrologue(input) + if (prologue.kind === "settled") return prologue.result + const result = yield* runTurn(input.sessionID, prologue.promotion, input.step) return yield* stepContinuation(input.sessionID, result.needsContinuation, result.step) }) + const runModelCall = Effect.fn("SessionRunner.runModelCall")(function* (input: StepInput) { + const prologue = yield* stepPrologue(input) + if (prologue.kind === "settled") return { kind: "settled", result: prologue.result } satisfies ModelCallResult + const result = yield* runTurn(input.sessionID, prologue.promotion, input.step, true) + return { + kind: "called", + step: result.step, + calls: result.calls, + settlement: result.settlement, + assistantMessageID: result.assistantMessageID, + needsContinuation: result.needsContinuation, + } satisfies ModelCallResult + }) + return Service.of({ run, runStep, + runModelCall, + runToolCall, + failToolCall, + sealStep, }) }), ) diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index 93dd3e86bb9c..272074a1d34a 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -11,6 +11,10 @@ type Input = { readonly agent: string readonly model: ModelV2.Ref readonly snapshot?: string + /** Record the calls the provider asked for, but leave Tool.Called to whoever dispatches them. + * The log then says which calls were only asked for and which one a process had started, which + * is the difference between re-running a side effect and reporting it as unknown. */ + readonly deferCalls?: boolean } const safe = (value: number | undefined) => Math.max(0, Number.isFinite(value) ? (value ?? 0) : 0) @@ -27,7 +31,16 @@ const tokens = (usage: Usage | undefined) => { } } -const record = (value: unknown): Record => +/** What the provider said about how the step ended. Held in memory by the publisher, so a caller + * that defers Step.Ended to another process has to carry it there itself. */ +export interface StepSettlement { + readonly finish: string + readonly tokens: ReturnType +} + +/** The shape a tool call's input is recorded in. Exported so a dispatcher publishing Tool.Called + * for a deferred call records it exactly as the streaming path would. */ +export const record = (value: unknown): Record => typeof value === "object" && value !== null && !Array.isArray(value) ? (value as Record) : { value } const message = (value: unknown) => { @@ -101,6 +114,9 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) { readonly assistantMessageID: SessionMessage.ID readonly name: string + // Whether the provider streamed the arguments. One that delivers the call whole sends none, + // and the fragment end would then record an empty input for a call that has one. + inputSeen: boolean inputEnded: boolean called: boolean settled: boolean @@ -113,7 +129,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) let assistantActive = false let assistantFailed = false let providerFailed = false - let stepSettlement: { readonly finish: string; readonly tokens: ReturnType } | undefined + let stepSettlement: StepSettlement | undefined const startAssistant = Effect.fnUntraced(function* () { if (assistantMessageID !== undefined) return assistantMessageID @@ -212,6 +228,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) tools.set(event.id, { assistantMessageID, name: event.name, + inputSeen: false, inputEnded: false, called: false, settled: false, @@ -341,6 +358,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) if (tool.name !== event.name) return yield* Effect.die(`Tool input name changed for ${event.id}: ${tool.name} -> ${event.name}`) if (tool.inputEnded) return yield* Effect.die(`Tool input delta after end: ${event.id}`) + tool.inputSeen = true yield* toolInput.append(event.id, event.text) yield* events.publish(SessionEvent.Tool.Input.Delta, { sessionID: input.sessionID, @@ -357,6 +375,10 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) case "tool-call": { if (!tools.has(event.id)) yield* startToolInput(event) const tool = tools.get(event.id)! + // The call carries the arguments whether or not they were streamed, and the record has to + // hold them either way: it is what a dispatcher reads to run the tool, and the fragment end + // would otherwise write an empty input for a provider that sends no deltas. + if (!tool.inputEnded && !tool.inputSeen) yield* toolInput.append(event.id, JSON.stringify(event.input ?? {})) if (!tool.inputEnded) yield* endToolInput(event) if (tool.name !== event.name) return yield* Effect.die(`Tool call name changed for ${event.id}: ${tool.name} -> ${event.name}`) @@ -364,6 +386,9 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) tool.called = true tool.providerExecuted = event.providerExecuted === true tool.providerMetadata = event.providerMetadata + // A provider-executed call has already run, so it is never deferred and is recorded here + // whatever the caller asked for. + if (input.deferCalls && !tool.providerExecuted) return yield* events.publish(SessionEvent.Tool.Called, { sessionID: input.sessionID, timestamp: yield* timestamp, diff --git a/packages/core/test/event-claim.test.ts b/packages/core/test/event-claim.test.ts new file mode 100644 index 000000000000..04a0a6194e7f --- /dev/null +++ b/packages/core/test/event-claim.test.ts @@ -0,0 +1,131 @@ +// The compare-and-set in `claim`, under the interleaving it exists for. +// +// Two attempts of one activity claim the log from two processes, so both can read the current owner +// before either writes. A pair of claims started together in one process never does that: they run +// to completion one after the other, which is why the concurrent test beside this one passes with +// the fix reverted. The seam here is at the database, not in `claim`: reads of the sequence table +// wait for each other while the barrier is armed, and `claim` itself is untouched. +import { describe, expect } from "bun:test" +import { Deferred, Effect, Exit, Layer } from "effect" +import { eq } from "drizzle-orm" +import { EventV2 } from "@opencode-ai/core/event" +import { EventSequenceTable } from "@opencode-ai/core/event/sql" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Location } from "@opencode-ai/core/location" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { WorkspaceV2 } from "@opencode-ai/core/workspace" +import { Session } from "@opencode-ai/schema/session" +import { SessionV1 } from "@opencode-ai/schema/session-v1" +import { location } from "./fixture/location" +import { testEffect } from "./lib/effect" + +const locationLayer = Layer.succeed( + Location.Service, + Location.Service.of( + location({ directory: AbsolutePath.make("project"), workspaceID: WorkspaceV2.ID.make("wrk_test") }), + ), +) + +// While armed, the first `want` reads wait for each other and are then released together. +const barrier = { + held: 0, + want: 0, + gate: undefined as Deferred.Deferred | undefined, +} + +const hold = () => + Effect.gen(function* () { + const gate = barrier.gate + if (!gate || barrier.want === 0) return + barrier.held++ + if (barrier.held >= barrier.want) { + barrier.want = 0 + yield* Deferred.succeed(gate, void 0) + return + } + yield* Deferred.await(gate) + }) + +// Waits after the read rather than before it. What has to interleave is two claims that both saw +// the same owner; holding before the read would serialize them and prove nothing. +const gated = (db: any): any => { + const wrap = (node: any): any => + new Proxy(node, { + get(target, prop, recv) { + const value = Reflect.get(target, prop, recv) + if (typeof value !== "function") return value + if (prop === "get" || prop === "all") + return (...args: any[]) => value.apply(target, args).pipe(Effect.tap(() => hold())) + return (...args: any[]) => { + const out = value.apply(target, args) + return out && typeof out === "object" ? wrap(out) : out + } + }, + }) + return new Proxy(db, { + get(target, prop, recv) { + const value = Reflect.get(target, prop, recv) + if (prop !== "select") return typeof value === "function" ? value.bind(target) : value + return (...args: any[]) => wrap(value.apply(target, args)) + }, + }) +} + +const gatedDatabase = Layer.effect( + Database.Service, + Effect.gen(function* () { + const real = yield* Database.Service + return { db: gated(real.db) } + }), +).pipe(Layer.provide(Database.layerFromPath(":memory:"))) + +const it = testEffect( + AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, Location.node]), [ + [Location.node, locationLayer], + [Database.node, gatedDatabase], + ]), +) + +const DurableMessage = SessionV1.Event.MessageRemoved + +describe("claim under a real interleaving", () => { + it.effect("only one of two claims that read the same owner is told it won", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = Session.ID.create() + yield* events.publish(DurableMessage, { + sessionID: aggregateID, + messageID: SessionV1.MessageID.ascending("msg_seed"), + }) + yield* events.claim(aggregateID, "run:11:1") + + barrier.gate = yield* Deferred.make() + barrier.held = 0 + barrier.want = 2 + + const outcomes = yield* Effect.all( + ["run:11:2", "run:11:3"].map((token) => events.claim(aggregateID, token).pipe(Effect.exit)), + { concurrency: "unbounded" }, + ) + barrier.gate = undefined + barrier.want = 0 + + const { db } = yield* Database.Service + const row = yield* db + .select({ ownerID: EventSequenceTable.owner_id }) + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, aggregateID)) + .get() + + // Asserted first, because without it the rest proves nothing: it says both claims really did + // read the same owner before either wrote. + expect(barrier.held).toBe(2) + // The one that loses must be told so. Two winners means the loser goes on to publish under a + // token the log has already fenced, and its tools die on a step that is running. + expect(outcomes.filter(Exit.isSuccess).length).toBe(1) + expect(["run:11:2", "run:11:3"]).toContain(row?.ownerID ?? "") + }), + ) +}) diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index d7ee0f69f274..e1c072a9148c 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -773,6 +773,41 @@ describe("EventV2", () => { }), ) + // Two attempts of one activity can be alive at once, and they do not arrive in order. A paused + // attempt 1 resuming after attempt 2 has claimed must not take the log back, or it fences out the + // tool activities of the step that is going. + it.effect("a resumed earlier attempt cannot take the log back", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = Session.ID.create() + yield* events.publish(DurableMessage, durableData(aggregateID, "seed")) + + // Activity ids as Temporal writes them: an increasing sequence within the run. + yield* events.claim(aggregateID, "run:11:1") + yield* events.claim(aggregateID, "run:11:2") + const stale = yield* events.claim(aggregateID, "run:11:1").pipe(Effect.exit) + expect(Exit.isFailure(stale)).toBe(true) + + const { db } = yield* Database.Service + const row = yield* db + .select({ ownerID: EventSequenceTable.owner_id }) + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, aggregateID)) + .get() + expect(row?.ownerID).toBe("run:11:2") + + // A later activity is a different unit of work, so it still takes the log: this is the seal + // claiming after the model call, not a zombie. + yield* events.claim(aggregateID, "run:12:1") + + // And an earlier one never does, whatever its attempt says. A model call paused before it + // claimed, with three steps completing under other activity ids while it was away, used to + // come back and fence out the step that was actually running. + const fromAnEarlierStep = yield* events.claim(aggregateID, "run:4:1").pipe(Effect.exit) + expect(Exit.isFailure(fromAnEarlierStep)).toBe(true) + }), + ) + it.effect("claim fences replay owners", () => Effect.gen(function* () { const events = yield* EventV2.Service @@ -1131,6 +1166,45 @@ describe("EventV2", () => { }), ) + it.effect("admits every writer holding the claimed token, concurrently", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const { db } = yield* Database.Service + const aggregateID = Session.ID.create() + // One step's writers share the token its model attempt claimed: the model call, each tool + // call, and the seal. They run as separate activities, so the fence has to admit all of them + // while still rejecting the attempt they superseded. + const step = "run-1:model-1:1" + const superseded = "run-1:model-1:0" + + yield* events.claim(aggregateID, step) + yield* Effect.all( + ["model", "tool-a", "tool-b", "seal"].map((writer) => + events + .publish(DurableMessage, durableData(aggregateID, writer)) + .pipe(Effect.provideService(EventV2.EventOwner, step)), + ), + { concurrency: "unbounded" }, + ) + const stale = yield* events + .publish(DurableMessage, durableData(aggregateID, "zombie")) + .pipe(Effect.provideService(EventV2.EventOwner, superseded), Effect.exit) + + const rows = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, aggregateID)) + .all() + .pipe(Effect.orDie) + + expect(String(stale)).toContain("Owner fence") + expect(rows.map((row) => row.seq)).toEqual([0, 1, 2, 3]) + expect(rows.map((row) => (row.data as { messageID: string }).messageID).sort()).toEqual( + ["model", "tool-a", "tool-b", "seal"].map((writer) => durableData(aggregateID, writer).messageID).sort(), + ) + }), + ) + it.live("tails another writer's events only when the layer asked for a tick", () => Effect.gen(function* () { // A second service over the SAME database with its OWN pubsub: exactly what a standalone diff --git a/packages/core/test/lib/execution-conformance.ts b/packages/core/test/lib/execution-conformance.ts index 759db5a98f3f..a57c243a27fe 100644 --- a/packages/core/test/lib/execution-conformance.ts +++ b/packages/core/test/lib/execution-conformance.ts @@ -24,6 +24,8 @@ import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionExecution } from "@opencode-ai/core/session/execution" import { SessionExecutionLocal } from "@opencode-ai/core/session/execution/local" import { SessionRunnerModel, ModelNotSelectedError } from "@opencode-ai/core/session/runner/model" +import { ApplicationTools } from "@opencode-ai/core/tool/application-tools" +import { Tool } from "@opencode-ai/core/tool/tool" import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" import { SessionTable } from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" @@ -39,7 +41,7 @@ import { Auth } from "@opencode-ai/llm/route" import { describe, expect } from "bun:test" import { realpathSync } from "node:fs" import { tmpdir } from "node:os" -import { Cause, Context, DateTime, Effect, Exit, Layer, Stream } from "effect" +import { Cause, Context, DateTime, Effect, Exit, Layer, Schema, Stream } from "effect" import { testEffect } from "./effect" // The per-location service build resolves the session directory on disk, so it must exist. @@ -77,13 +79,63 @@ const countingModel = () => { return { requests, stream } } +// Asks for one tool on the first request and answers on the second, which is the smallest turn that +// makes a driver dispatch a tool and come back to the model with its result. +const toolCallingModel = () => { + const requests: number[] = [] + const stream: LLMClientShape["stream"] = () => { + requests.push(1) + if (requests.length > 1) + return Stream.fromIterable([LLMEvent.stepStart({ index: 0 }), LLMEvent.stepFinish({ index: 0, reason: "stop" })]) + return Stream.fromIterable([ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call_contract", name: "contract_probe", input: {} }), + LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }), + ]) + } + return { requests, stream } +} + +// Counts its own executions, so "ran exactly once" is checked against the tool rather than only +// against the projection. +const probeTool = (ran: { count: number }) => + Tool.make({ + description: "contract probe", + input: Schema.Struct({}), + output: Schema.String, + execute: () => + Effect.sync(() => { + ran.count += 1 + return "probed" + }), + }) + +// A tool that never finishes, so a turn can be interrupted with one in flight. +const hangingTool = () => + Tool.make({ + description: "contract probe that never finishes", + input: Schema.Struct({}), + output: Schema.String, + execute: () => Effect.never, + }) + +const toolParts = (messages: ReadonlyArray) => + messages.flatMap((message) => + message.type === "assistant" + ? message.content.filter((part): part is SessionMessage.AssistantTool => part.type === "tool") + : [], + ) + // The executor under test, built as its own graph over the shared database file (the same way the // serve process builds it), with the model/LLM mocked. Any SessionExecution node with the standard // dependency set (the local coordinator, the Temporal driver) plugs in here. export const makeExecutionFor = (node: typeof SessionExecutionLocal.node) => (stream: LLMClientShape["stream"], models = okModels) => - AppNodeBuilder.build(node, [ + // The tool registry is named in the group so a scenario can register a probe into the same + // graph the driver runs on. One graph, one instance: what the test registers is what the + // dispatch finds. + AppNodeBuilder.build(LayerNode.group([node, ApplicationTools.node]), [ [LayerNodePlatform.llmClient, mockClient(stream)], [PermissionV2.node, permission], [ToolOutputStore.node, ToolOutputStore.nodeWithoutConfig], @@ -281,5 +333,82 @@ export const runContract = (label: string, makeExec: ReturnType + withIdleOverride( + Effect.gen(function* () { + yield* seedSession(sessionID) + yield* seedPrompt(sessionID) + const graph = yield* Layer.build(makeExec(stream)) + yield* Context.get(graph, ApplicationTools.Service).register({ + contract_probe: probeTool(ran), + }) + const exec = Context.get(graph, SessionExecution.Service) + yield* exec.wake(sessionID) + // Waiting on the turn, not on the step: the first step's assistant is completed the + // moment the step is sealed, while the follow-up provider call is still to come. + yield* until(exec.active, (active) => !active.has(sessionID)) + // Where the drivers could drift: a step's model call, its tool and its close are one + // activity in one mode and three in the other, so the tool has to be dispatched once, + // settled in the log, and answered by a second provider turn either way. + expect(ran.count).toBe(1) + expect(requests).toHaveLength(2) + const store = yield* SessionStore.Service + const context = yield* store.context(sessionID) + const parts = context.flatMap((message) => (message.type === "assistant" ? message.content : [])) + const call = parts.findLast( + (part): part is SessionMessage.AssistantTool => part.type === "tool" && part.id === "call_contract", + ) + expect(call?.state.status).toBe("completed") + }), + ), + 60000, + ) + } + + { + const { stream } = toolCallingModel() + const sessionID = SessionV2.ID.make(`ses_${slug}_interrupt_tool`) + it.live( + "closes a tool call the interrupt cut short", + () => + withIdleOverride( + Effect.gen(function* () { + yield* seedSession(sessionID) + yield* seedPrompt(sessionID) + const graph = yield* Layer.build(makeExec(stream)) + yield* Context.get(graph, ApplicationTools.Service).register({ + contract_probe: hangingTool(), + }) + const exec = Context.get(graph, SessionExecution.Service) + const store = yield* SessionStore.Service + yield* exec.wake(sessionID) + yield* until(store.context(sessionID), (context) => + toolParts(context).some((part) => part.state.status === "running"), + ) + + yield* exec.interrupt(sessionID) + + // A call left running is not a cosmetic loose end: the transcript shows a tool still + // going, and the next turn has to reconstruct what happened to it. A whole step closes + // the tools it opened on its way out, and a step whose calls are their own units of + // work has to end up in the same place. + yield* until(store.context(sessionID), (context) => + toolParts(context).every((part) => part.state.status !== "running"), + ) + const closed = toolParts(yield* store.context(sessionID)) + expect(closed.map((part) => part.state.status)).toEqual(["error"]) + yield* until(exec.active, (active) => !active.has(sessionID)) + }), + ), + 60000, + ) + } }) } diff --git a/packages/core/test/lib/runner-harness.ts b/packages/core/test/lib/runner-harness.ts index 5a262c693a26..8954477abd18 100644 --- a/packages/core/test/lib/runner-harness.ts +++ b/packages/core/test/lib/runner-harness.ts @@ -64,7 +64,15 @@ export const mockClient = (stream: LLMClientShape["stream"]) => export const emptyStep: LLMClientShape["stream"] = () => Stream.fromIterable([LLMEvent.stepStart({ index: 0 }), LLMEvent.stepFinish({ index: 0, reason: "stop" })]) -export const runnerHarness = (stream: LLMClientShape["stream"]) => +/** The fixtures a test can swap: a model with other limits, a config that compacts, a store that + * cannot keep tool output. Everything else in the graph stays the same. */ +export interface HarnessOverrides { + readonly model?: typeof models + readonly config?: typeof config + readonly outputStore?: typeof ToolOutputStore.nodeWithoutConfig | Layer.Layer +} + +export const runnerHarness = (stream: LLMClientShape["stream"], overrides: HarnessOverrides = {}) => testEffect( AppNodeBuilder.build( LayerNode.group([ @@ -86,13 +94,13 @@ export const runnerHarness = (stream: LLMClientShape["stream"]) => [ [LayerNodePlatform.llmClient, mockClient(stream)], [PermissionV2.node, permission], - [ToolOutputStore.node, ToolOutputStore.nodeWithoutConfig], - [SessionRunnerModel.node, models], + [ToolOutputStore.node, overrides.outputStore ?? ToolOutputStore.nodeWithoutConfig], + [SessionRunnerModel.node, overrides.model ?? models], [SystemContextRegistry.node, systemContext], [Location.node, Location.boundNode({ directory: PROJECT })], [SkillGuidance.node, skillGuidance], [ReferenceGuidance.node, referenceGuidance], - [Config.node, config], + [Config.node, overrides.config ?? config], [Snapshot.node, Snapshot.noopLayer], ], ), diff --git a/packages/core/test/session-runner-model-call.test.ts b/packages/core/test/session-runner-model-call.test.ts new file mode 100644 index 000000000000..7ee239d68f0d --- /dev/null +++ b/packages/core/test/session-runner-model-call.test.ts @@ -0,0 +1,794 @@ +// The model-only attempt: runModelCall performs one provider attempt, records the tool calls it +// asked for, and stops. The caller dispatches each call as its own unit of work and seals the step +// afterwards, which is what puts the model-to-tools loop in a durable executor rather than inside a +// single activity. These tests pin the three properties that split depends on: +// - the call is durable but not started, so its side effect has NOT run +// - the provider-minted callID and the publisher's assistantMessageID are handed back, never +// regenerated, so the dispatcher's result can be matched to the recorded call +// - the step is left open (no Step.Ended), because the tools have not run yet +// The contrast case is the same stream through runStep, which runs the tool, as a whole step does. +import type { LLMClientShape } from "@opencode-ai/llm/route" +import { LLMEvent } from "@opencode-ai/llm" +import { Database } from "@opencode-ai/core/database/database" +import { EventTable } from "@opencode-ai/core/event/sql" +import { EventV2 } from "@opencode-ai/core/event" +import { PermissionV2 } from "@opencode-ai/core/permission" +import { Config } from "@opencode-ai/core/config" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionRunner } from "@opencode-ai/core/session/runner" +import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" +import { ApplicationTools } from "@opencode-ai/core/tool/application-tools" +import { Tool } from "@opencode-ai/core/tool/tool" +import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" +import { SessionStore } from "@opencode-ai/core/session/store" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionRunDeclinedError } from "@opencode-ai/core/session/error" +import { ConfigCompaction } from "@opencode-ai/core/config/compaction" +import { SessionContextEpoch } from "@opencode-ai/core/session/context-epoch" +import { createLLMEventPublisher } from "@opencode-ai/core/session/runner/publish-llm-event" +import { ModelV2 } from "@opencode-ai/core/model" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { SystemContext } from "@opencode-ai/core/system-context" +import { eq } from "drizzle-orm" +import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat" +import { Auth } from "@opencode-ai/llm/route" +import { describe, expect, spyOn } from "bun:test" +import { Cause, DateTime, Deferred, Effect, Exit, Layer, Schema, Stream } from "effect" +import { emptyStep, runnerHarness, seedSession } from "./lib/runner-harness" + +// One tool call, then a clean finish: the shape a step that wants to keep going produces. +const callsTool: LLMClientShape["stream"] = () => + Stream.fromIterable([ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call_probe", name: "probe_write", input: {} }), + LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }), + ]) +// Calls the tool that declines. +const callsDecliningTool: LLMClientShape["stream"] = () => + Stream.fromIterable([ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call_probe", name: "probe_declines", input: {} }), + LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }), + ]) +// A tool call, then the provider dies mid-stream. The calls are recorded but the turn is over. +const callsToolThenFails: LLMClientShape["stream"] = () => + Stream.fromIterable([ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call_probe", name: "probe_write", input: {} }), + LLMEvent.providerError({ message: "upstream exploded" }), + ]) +// Calls the tools that die with their side effect in flight, which is what a worker crash leaves +// behind for the next dispatch to read. +const callsCrashingTool: LLMClientShape["stream"] = () => + Stream.fromIterable([ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call_probe", name: "probe_crashes", input: {} }), + LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }), + ]) +const callsCrashingIdempotentTool: LLMClientShape["stream"] = () => + Stream.fromIterable([ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call_probe", name: "probe_crashes_read", input: {} }), + LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }), + ]) +// The two shapes a provider delivers arguments in, against a tool that actually wants some. The +// hand-off names the call and nothing else, so what the tool receives comes off the log, and both +// shapes have to leave the same thing there. Whole first: no input deltas at all, which is what the +// fragment buffer would otherwise record as an empty input. +const callsEchoWhole: LLMClientShape["stream"] = () => + Stream.fromIterable([ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call_probe", name: "probe_echo", input: { text: "hello" } }), + LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }), + ]) +// Streamed, in pieces, which is what a provider that emits partial JSON does. +const callsEchoStreamed: LLMClientShape["stream"] = () => + Stream.fromIterable([ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolInputStart({ id: "call_probe", name: "probe_echo" }), + LLMEvent.toolInputDelta({ id: "call_probe", name: "probe_echo", text: '{"text":' }), + LLMEvent.toolInputDelta({ id: "call_probe", name: "probe_echo", text: '"hello"}' }), + LLMEvent.toolInputEnd({ id: "call_probe", name: "probe_echo" }), + LLMEvent.toolCall({ id: "call_probe", name: "probe_echo", input: { text: "hello" } }), + LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }), + ]) +// An answer and no tool call: the step is over as soon as the stream is, but the seal still has to +// happen, and there is a real assistant message for it to complete. +const textOnly: LLMClientShape["stream"] = () => + Stream.fromIterable([ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.textStart({ id: "txt_1" }), + LLMEvent.textDelta({ id: "txt_1", text: "done" }), + LLMEvent.textEnd({ id: "txt_1" }), + LLMEvent.stepFinish({ index: 0, reason: "stop" }), + ]) + +// A store that cannot keep what a tool produced. The tool itself still ran by then, which is the +// case a dispatch has to report honestly rather than as a call that never finished. +const failingOutputStore = Layer.mock(ToolOutputStore.Service, { + bound: () => Effect.fail(new ToolOutputStore.StorageError({ operation: "write", cause: new Error("disk full") })), +}) + +const sessionID = SessionV2.ID.make("ses_runner_model_call") + +// Probes that count their own executions, so "recorded but not run" and "not run twice" are +// checked against the tools themselves rather than only against the projection. The read probes +// declare themselves repeatable; the write probes do not, which is what decides whether a second +// dispatch runs the tool again. +const registerProbes = (ran: { write: number; read: number; echoed?: string }) => + Effect.gen(function* () { + yield* (yield* ApplicationTools.Service).register({ + // The one probe that wants an argument. Every other schema here is an empty struct, which + // accepts anything, so none of them can tell whether a tool was handed what the model asked + // for. This one records it. + probe_echo: Tool.make({ + description: "echo probe", + input: Schema.Struct({ text: Schema.String }), + output: Schema.String, + execute: (args: { readonly text: string }) => + Effect.sync(() => { + ran.echoed = args.text + return args.text + }), + }), + probe_write: Tool.make({ + description: "write probe", + input: Schema.Struct({}), + output: Schema.String, + execute: () => + Effect.sync(() => { + ran.write += 1 + return "wrote" + }), + }), + // permission.assert declines by dying with this, so a tool that dies the same way exercises + // the same classification without needing the permission service in the tool's context. + probe_declines: Tool.make({ + description: "declines", + input: Schema.Struct({}), + output: Schema.String, + execute: () => Effect.die(new PermissionV2.DeclinedError()), + }), + probe_read: Tool.make({ + description: "read probe", + idempotent: true, + input: Schema.Struct({}), + output: Schema.String, + execute: () => + Effect.sync(() => { + ran.read += 1 + return "read" + }), + }), + // Dying with the side effect already done is the case the log has to survive: the call is + // left in flight and no later reader can say whether the write landed. + probe_crashes: Tool.make({ + description: "crashing write probe", + input: Schema.Struct({}), + output: Schema.String, + execute: () => + Effect.gen(function* () { + ran.write += 1 + return yield* Effect.die(new Error("worker died")) + }), + }), + // The repeatable one, dying only on its first run so a second dispatch has a result to + // produce. + probe_crashes_read: Tool.make({ + description: "crashing read probe", + idempotent: true, + input: Schema.Struct({}), + output: Schema.String, + execute: () => + Effect.gen(function* () { + ran.read += 1 + if (ran.read === 1) return yield* Effect.die(new Error("worker died")) + return "read" + }), + }), + }) + }) + +const counters = () => ({ write: 0, read: 0 }) as { write: number; read: number; echoed?: string } + +// A seeded session with the probes registered, which every case starts from. +const ready = Effect.gen(function* () { + yield* seedSession(sessionID) + const ran = counters() + yield* registerProbes(ran) + return ran +}) + +const modelCall = (step = 2) => + Effect.flatMap(SessionRunner.Service, (runner) => + runner.runModelCall({ sessionID, step, promotion: undefined, first: false, force: false }), + ) + +// The attempt that handed back its calls, or a failure: every dispatch and seal case starts here. +const deferStep = Effect.gen(function* () { + const result = yield* modelCall() + if (result.kind !== "called") throw new Error("expected a deferred step") + return result +}) +const deferOneCall = Effect.map(deferStep, (result) => { + if (!result.calls[0]) throw new Error("expected a deferred call") + return result.calls[0] +}) + +const toolPart = (messages: ReadonlyArray, callID: string) => { + for (const message of messages) { + if (message.type !== "assistant") continue + for (const part of message.content) if (part.type === "tool" && part.id === callID) return part + } + return undefined +} + +const toolStatus = (messages: ReadonlyArray, callID: string) => { + const part = toolPart(messages, callID) + return part?.type === "tool" ? part.state.status : undefined +} + +const assistant = (messages: ReadonlyArray) => + messages.find((message) => message.type === "assistant") + +// Whether the step's assistant message is closed. Undefined when there is no message at all, so an +// assertion on either value fails rather than passing by accident. +const closed = (messages: ReadonlyArray) => { + const message = assistant(messages) + return message?.type === "assistant" ? Boolean(message.time.completed) : undefined +} + +const stepEndedCount = Effect.gen(function* () { + const { db } = yield* Database.Service + const rows = yield* db + .select({ type: EventTable.type }) + .from(EventTable) + .where(eq(EventTable.aggregate_id, sessionID)) + .all() + .pipe(Effect.orDie) + return rows.filter((row) => row.type.includes("step.ended")).length +}) + +describe("SessionRunner model-only attempt", () => { + runnerHarness(callsTool).effect("records the tool call, hands it back, and does not run it", () => + Effect.gen(function* () { + const ran = yield* ready + const store = yield* SessionStore.Service + + const result = yield* deferStep + + // The provider's callID and the publisher's assistant message id are carried out, because a + // second run of this step would mint different ones and nothing would match the log. + expect(result.calls.map((call) => call.id)).toEqual(["call_probe"]) + expect(result.calls[0]?.name).toBe("probe_write") + expect(result.calls[0]?.assistantMessageID).toBeTruthy() + // The provider's finish reason has to travel with the calls: it lives only in the publisher's + // memory, so whoever seals the step in another process cannot read it back from the log. + expect(result.settlement?.finish).toBe("tool-calls") + expect(ran.write).toBe(0) + + const context = yield* store.context(sessionID) + // Durably recorded and not started: whoever dispatches the call publishes Tool.Called, so + // `running` in the log means a process was about to run the tool. Recording it here instead + // would make every later dispatch of a crashed step report an unknown outcome for a tool + // nobody had touched. + expect(toolStatus(context, "call_probe")).toBe("pending") + // The step stays open, so no Step.Ended and no completed assistant message. + expect(closed(context)).toBe(false) + }), + ) + + runnerHarness(textOnly).effect("leaves a text-only step open with no calls to dispatch", () => + Effect.gen(function* () { + yield* seedSession(sessionID) + const store = yield* SessionStore.Service + + const result = yield* deferStep + + expect(result.calls).toHaveLength(0) + expect(result.settlement?.finish).toBe("stop") + // Sealing is uniform: even with nothing to dispatch, the step is closed by the seal, not here, + // so the answer is recorded but its message is still open. + expect(closed(yield* store.context(sessionID))).toBe(false) + }), + ) + + runnerHarness(callsTool).effect("still runs the tool and closes the step through runStep", () => + Effect.gen(function* () { + const ran = yield* ready + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + + yield* runner.runStep({ sessionID, step: 2, promotion: undefined, first: false, force: false }) + + // The contrast that makes the split meaningful: the whole-step path dispatches and seals. + expect(ran.write).toBe(1) + const context = yield* store.context(sessionID) + expect(toolStatus(context, "call_probe")).toBe("completed") + expect(closed(context)).toBe(true) + }), + ) +}) + +// Dispatching one recorded call on its own. The policy under test is what happens when a dispatch +// has already started the tool and died, so the side effect may have run and nothing can say: +// repeatable tools run again, the rest are reported unknown so the model decides. Same rule the +// crash-resume path already follows, off the same evidence. +describe("SessionRunner tool dispatch", () => { + runnerHarness(callsTool).effect("runs a deferred call and records its result", () => + Effect.gen(function* () { + const ran = yield* ready + const call = yield* deferOneCall + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + + const result = yield* runner.runToolCall({ sessionID, call }) + + expect(result.outcome).toBe("settled") + expect(ran.write).toBe(1) + expect(toolStatus(yield* store.context(sessionID), "call_probe")).toBe("completed") + }), + ) + + // The arguments come off the recorded call, not the hand-off, so a dispatch that reads them + // wrongly hands the tool something its schema refuses, and the model spends a turn being told its + // own input was not an object. Every other probe here takes an empty struct, which accepts that + // silently. + runnerHarness(callsEchoWhole).effect("hands the tool the arguments the model asked with", () => + Effect.gen(function* () { + yield* seedSession(sessionID) + const ran = counters() + yield* registerProbes(ran) + const call = yield* deferOneCall + const runner = yield* SessionRunner.Service + + const result = yield* runner.runToolCall({ sessionID, call }) + + expect(result.outcome).toBe("settled") + expect(ran.echoed).toBe("hello") + }), + ) + + // The same call, streamed in pieces instead of delivered whole. Both shapes have to leave the + // arguments in the log, because the dispatcher cannot tell which one produced the call. + runnerHarness(callsEchoStreamed).effect("and the same when the provider streamed them", () => + Effect.gen(function* () { + yield* seedSession(sessionID) + const ran = counters() + yield* registerProbes(ran) + const call = yield* deferOneCall + const runner = yield* SessionRunner.Service + + const result = yield* runner.runToolCall({ sessionID, call }) + + expect(result.outcome).toBe("settled") + expect(ran.echoed).toBe("hello") + }), + ) + + runnerHarness(callsTool).effect("executes one non-idempotent call once under overlapping dispatches", () => + Effect.gen(function* () { + yield* seedSession(sessionID) + const ran = counters() + yield* registerProbes(ran) + const call = yield* deferOneCall + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + const read = store.message + const gate = yield* Deferred.make() + let readers = 0 + const spy = spyOn(store, "message").mockImplementation((id) => + read(id).pipe( + Effect.tap((value) => + Effect.gen(function* () { + if (readers >= 2) return + expect(toolPart(value ? [value.message] : [], call.id)?.state.status).toBe("pending") + readers++ + if (readers === 2) yield* Deferred.succeed(gate, undefined) + yield* Deferred.await(gate) + }), + ), + ), + ) + const outcomes = yield* Effect.all( + [runner.runToolCall({ sessionID, call }), runner.runToolCall({ sessionID, call })].map(Effect.exit), + { concurrency: "unbounded" }, + ).pipe(Effect.ensuring(Effect.sync(() => spy.mockRestore()))) + + expect(readers).toBe(2) + expect(ran.write).toBe(1) + expect(outcomes.filter(Exit.isSuccess)).toHaveLength(1) + }), + ) + + runnerHarness(callsTool).effect("does nothing when the call already has a result", () => + Effect.gen(function* () { + const ran = yield* ready + const call = yield* deferOneCall + const runner = yield* SessionRunner.Service + + yield* runner.runToolCall({ sessionID, call }) + // A duplicate dispatch: at-least-once delivery means this happens, and it must not re-run. + const second = yield* runner.runToolCall({ sessionID, call }) + + expect(second.outcome).toBe("already-settled") + expect(ran.write).toBe(1) + }), + ) + + runnerHarness(callsCrashingTool).effect( + "reports a side-effecting call whose dispatch already started as unknown", + () => + Effect.gen(function* () { + const ran = yield* ready + const call = yield* deferOneCall + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + + // A dispatch that died with the tool in flight. What it leaves in the log is the whole + // evidence a later one gets: the call recorded as running, and no result. + const crashed = yield* runner.runToolCall({ sessionID, call }).pipe(Effect.exit) + expect(Exit.isFailure(crashed)).toBe(true) + expect(ran.write).toBe(1) + + const result = yield* runner.runToolCall({ sessionID, call }) + + expect(result.outcome).toBe("unknown") + expect(ran.write).toBe(1) + expect(toolStatus(yield* store.context(sessionID), "call_probe")).toBe("error") + }), + ) + + runnerHarness(callsTool, { outputStore: failingOutputStore }).effect( + "tells the model why a dispatched call failed", + () => + Effect.gen(function* () { + const ran = yield* ready + const call = yield* deferOneCall + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + + const result = yield* runner.runToolCall({ sessionID, call }) + + // The tool ran and only its output was lost. Letting that fail the dispatch would repeat + // the write on the next attempt, and the step would finally close the call as interrupted: + // a reason the model cannot act on, and not what happened to it. + expect(result.outcome).toBe("failed") + expect(ran.write).toBe(1) + const part = toolPart(yield* store.context(sessionID), "call_probe") + const failure = part?.type === "tool" && part.state.status === "error" ? part.state.error : undefined + expect(failure?.message).toContain("disk full") + }), + ) + + runnerHarness(callsCrashingTool).effect("closes a started call that a stop cut short", () => + Effect.gen(function* () { + yield* ready + const call = yield* deferOneCall + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + const recorded = Effect.map(store.context(sessionID), (context) => { + const part = toolPart(context, "call_probe") + return part?.type === "tool" ? part.state : undefined + }) + + // Nothing has started this call, so it is left for the next turn's entry check: closing it + // here would report a tool the model asked for as having been cut short. + yield* runner.failToolCall({ sessionID, call }) + expect((yield* recorded)?.status).toBe("pending") + + // A dispatch died with the tool in flight, which is what the log shows after a stop lands + // mid-tool: the call recorded as running, with no result. + yield* runner.runToolCall({ sessionID, call }).pipe(Effect.exit) + expect((yield* recorded)?.status).toBe("running") + + yield* runner.failToolCall({ sessionID, call }) + const closedCall = yield* recorded + expect(closedCall?.status).toBe("error") + const reason = closedCall?.status === "error" ? closedCall.error.message : "" + expect(reason).toBe("Tool execution interrupted") + + // A call that is already terminal keeps what it has, whatever lands afterwards. + yield* runner.failToolCall({ sessionID, call }) + expect(yield* recorded).toEqual(closedCall) + }), + ) + + runnerHarness(callsCrashingIdempotentTool).effect("re-runs a started call that declares itself repeatable", () => + Effect.gen(function* () { + const ran = yield* ready + const call = yield* deferOneCall + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + + yield* runner.runToolCall({ sessionID, call }).pipe(Effect.exit) + const result = yield* runner.runToolCall({ sessionID, call }) + + expect(result.outcome).toBe("settled") + expect(ran.read).toBe(2) + expect(toolStatus(yield* store.context(sessionID), "call_probe")).toBe("completed") + }), + ) +}) + +// Closing the step after its calls have been dispatched. This is the piece that cannot stay in the +// provider attempt: the end snapshot and the file diff have to be taken after the tools have run, +// and in a durable executor that is a different process. +describe("SessionRunner step seal", () => { + runnerHarness(callsTool).effect("closes a dispatched step and keeps the turn going", () => + Effect.gen(function* () { + yield* ready + const model = yield* deferStep + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + yield* runner.runToolCall({ sessionID, call: model.calls[0]! }) + + const result = yield* runner.sealStep({ sessionID, step: 2, settlement: model.settlement }) + + // Local tool calls mean the model has results to look at, so the turn continues. + expect(result.continue).toBe(true) + expect(result.step).toBe(3) + expect(closed(yield* store.context(sessionID))).toBe(true) + expect(yield* stepEndedCount).toBe(1) + }), + ) + + runnerHarness(textOnly).effect("closes a text-only step and ends the turn", () => + Effect.gen(function* () { + yield* seedSession(sessionID) + const model = yield* deferStep + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + + const result = yield* runner.sealStep({ sessionID, step: 2, settlement: model.settlement }) + + expect(result.continue).toBe(false) + expect(closed(yield* store.context(sessionID))).toBe(true) + }), + ) + + runnerHarness(callsTool).effect("seals once and answers the same on a repeat", () => + Effect.gen(function* () { + yield* ready + const model = yield* deferStep + const runner = yield* SessionRunner.Service + yield* runner.runToolCall({ sessionID, call: model.calls[0]! }) + + const first = yield* runner.sealStep({ sessionID, step: 2, settlement: model.settlement }) + // A seal that published Step.Ended and then died is retried. The loop decision has to survive + // that, or the turn would stop one step early. + const second = yield* runner.sealStep({ sessionID, step: 2, settlement: model.settlement }) + + expect(second).toEqual(first) + expect(yield* stepEndedCount).toBe(1) + }), + ) + + runnerHarness(callsTool).effect("closes a call the dispatcher never settled", () => + Effect.gen(function* () { + const ran = yield* ready + const model = yield* deferStep + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + + // The tool activity exhausted its retries and never published a result. Sealing has to close + // the call anyway: a request carrying a tool_use with no tool_result is rejected outright, so + // leaving it open would poison every later attempt. + const result = yield* runner.sealStep({ sessionID, step: 2, settlement: model.settlement }) + + expect(result.continue).toBe(true) + expect(toolStatus(yield* store.context(sessionID), "call_probe")).toBe("error") + expect(ran.write).toBe(0) + }), + ) +}) + +// A content-free provider turn must still close. The publisher mints the assistant message lazily +// on first content, so after such a stream there is no message in the log for a seal to find. The +// whole-step path survives it because Step.Ended mints one on the way past; a seal running in +// another process has no publisher to mint with, so the attempt hands it the id. +describe("SessionRunner seal of a silent turn", () => { + runnerHarness(emptyStep).effect("closes a turn that published no content of its own", () => + Effect.gen(function* () { + yield* seedSession(sessionID) + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + + const model = yield* deferStep + expect(model.assistantMessageID).toBeTruthy() + + const result = yield* runner.sealStep({ + sessionID, + step: model.step, + settlement: model.settlement, + assistantMessageID: model.assistantMessageID, + }) + + expect(result.continue).toBe(false) + expect(closed(yield* store.context(sessionID))).toBe(true) + }), + ) +}) + +// Compaction restarts the provider attempt by dying with a transition defect that runTurn catches +// and re-enters. That recursion has to carry `deferTools`, or a step that happens to compact +// silently falls back to running its tools inline and the caller never gets the calls it was meant +// to dispatch. +describe("SessionRunner model-only attempt under compaction", () => { + // The summary request is the one with no tools. It needs text back, while the turn itself needs a + // tool call, so the mock has to answer them differently. + const compactingStream: LLMClientShape["stream"] = (request) => + request.tools.length === 0 + ? Stream.fromIterable([ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.textStart({ id: "sum" }), + LLMEvent.textDelta({ id: "sum", text: "## Objective\n- keep going" }), + LLMEvent.textEnd({ id: "sum" }), + LLMEvent.stepFinish({ index: 0, reason: "stop" }), + ]) + : callsTool(request) + + const tightModel = SessionRunnerModel.layerWith(() => + Effect.succeed( + OpenAIChat.route + .with({ endpoint: { baseURL: "https://api.openai.com/v1" }, auth: Auth.bearer("fixture") }) + // Small enough that a seeded turn already overflows the headroom, so the attempt compacts. + .with({ limits: { context: 4_000, output: 50 } }) + .model({ id: "gpt-4o-mini" }), + ), + ) + const compactingConfig = Layer.succeed( + Config.Service, + Config.Service.of({ + entries: () => + Effect.succeed([ + new Config.Document({ + type: "document", + info: new Config.Info({ + compaction: new ConfigCompaction.Info({ + buffer: 3_000, + keep: new ConfigCompaction.Keep({ tokens: 1_000 }), + }), + }), + }), + ]), + }), + ) + + runnerHarness(compactingStream, { model: tightModel, config: compactingConfig }).effect( + "still defers dispatch after a compaction restart", + () => + Effect.gen(function* () { + const ran = yield* ready + // The context epoch has to exist BEFORE the history is seeded. It records the sequence it + // was created at, and the runner only reads entries after that baseline, so seeding first + // would put the whole conversation behind the baseline and the request would come out empty. + const { db } = yield* Database.Service + yield* SessionContextEpoch.initialize(db, Effect.succeed(SystemContext.empty), sessionID) + // A finished turn already in the log, sized into a narrow band. It has to exceed the request + // headroom (context - buffer = 1000 tokens) so the attempt compacts at all, while the + // summary prompt it produces has to stay under context - summaryOutput (3950) or compaction + // bails out on its own guard and never restarts the attempt. + const events = yield* EventV2.Service + const seeder = createLLMEventPublisher(events, { + sessionID, + agent: "build", + model: { id: ModelV2.ID.make("gpt-4o-mini"), providerID: ProviderV2.ID.make("openai") }, + }) + yield* seeder.publish(LLMEvent.textStart({ id: "old" })) + yield* seeder.publish(LLMEvent.textDelta({ id: "old", text: "Earlier answer. ".repeat(500) })) + yield* seeder.publish(LLMEvent.textEnd({ id: "old" })) + yield* seeder.publish(LLMEvent.stepFinish({ index: 0, reason: "stop" })) + + const result = yield* modelCall() + + // Guard against a vacuous pass: without compaction actually firing this proves nothing + // about the restart path. + const rows = yield* db + .select({ type: EventTable.type }) + .from(EventTable) + .where(eq(EventTable.aggregate_id, sessionID)) + .all() + .pipe(Effect.orDie) + expect(rows.filter((row) => row.type.includes("compaction")).length).toBeGreaterThan(0) + + // The point: whatever the attempt went through, dispatch is still the caller's. + expect(result.kind).toBe("called") + if (result.kind !== "called") return + expect(result.calls.map((call) => call.id)).toEqual(["call_probe"]) + expect(ran.write).toBe(0) + }), + ) +}) + +// A turn has to stop when the provider fails. Only the attempt knows that happened: the log shows a +// failed assistant with tool parts, which reads the same as a step that wants to keep going. If the +// seal re-derives the decision instead of being told, a hard provider failure loops on the durable +// path until a step ceiling catches it. +describe("SessionRunner provider failure in a stepped turn", () => { + runnerHarness(callsToolThenFails).effect("ends the turn instead of asking for another step", () => + Effect.gen(function* () { + const ran = yield* ready + const runner = yield* SessionRunner.Service + + const model = yield* deferStep + expect(model.needsContinuation).toBe(false) + // The attempt already failed them on the way out, so dispatching would only re-read settled + // parts. + expect(model.calls).toHaveLength(0) + + const result = yield* runner.sealStep({ + sessionID, + step: model.step, + settlement: model.settlement, + assistantMessageID: model.assistantMessageID, + needsContinuation: model.needsContinuation, + }) + + expect(result.continue).toBe(false) + expect(ran.write).toBe(0) + }), + ) +}) + +// A decline is the user stopping the turn, not one tool failing. If it reaches the dispatcher as an +// ordinary tool error it gets swallowed, the step seals, and the agent carries on past a refusal. +describe("SessionRunner declined permission in a stepped turn", () => { + runnerHarness(callsDecliningTool).effect("halts the turn rather than reporting a failed tool", () => + Effect.gen(function* () { + yield* ready + const call = yield* deferOneCall + const runner = yield* SessionRunner.Service + + const exit = yield* runner.runToolCall({ sessionID, call }).pipe(Effect.exit) + + // Named, not inferred: the activity boundary turns this error into a non-retryable halt, + // where a plain tool failure reads as one bad tool and the turn continues. Raising a bare + // interrupt instead would leave the boundary to guess the user's decision from the absence + // of a cancellation. + expect(Exit.isFailure(exit)).toBe(true) + if (!Exit.isFailure(exit)) return + expect(Cause.squash(exit.cause)).toBeInstanceOf(SessionRunDeclinedError) + }), + ) +}) + +// A step's writers share one owner token, so the fence does not separate two seal attempts. The +// projector is what holds the line, and step.ended was the one event in that window with no guard. +describe("SessionRunner duplicate seal", () => { + runnerHarness(callsTool).effect("keeps the first close when a late attempt lands", () => + Effect.gen(function* () { + yield* ready + const model = yield* deferStep + const runner = yield* SessionRunner.Service + const store = yield* SessionStore.Service + yield* runner.runToolCall({ sessionID, call: model.calls[0]! }) + yield* runner.sealStep({ + sessionID, + step: model.step, + settlement: model.settlement, + assistantMessageID: model.assistantMessageID, + needsContinuation: model.needsContinuation, + }) + const closedAt = assistant(yield* store.context(sessionID)) + const first = String(closedAt?.type === "assistant" ? closedAt.time.completed : undefined) + + // A zombie attempt publishing its own Step.Ended under the same token is admitted by the + // fence, so the projection has to reject it. + const events = yield* EventV2.Service + yield* events.publish(SessionEvent.Step.Ended, { + sessionID, + timestamp: yield* DateTime.now, + assistantMessageID: SessionMessage.ID.make(model.assistantMessageID!), + finish: "stop", + cost: 0, + tokens: { input: 99, output: 99, reasoning: 0, cache: { read: 0, write: 0 } }, + }) + + const after = assistant(yield* store.context(sessionID)) + expect(String(after?.type === "assistant" ? after.time.completed : undefined)).toBe(first) + expect(after?.type === "assistant" ? after.tokens?.input : undefined).not.toBe(99) + }), + ) +}) diff --git a/packages/temporal/README.md b/packages/temporal/README.md index 707f47649a4c..1a2a31ce9e82 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -58,6 +58,8 @@ session runs as the workflow `session-exec-`. | `OPENCODE_TEMPORAL_TASK_QUEUE` | `opencode-session-exec` | | | `OPENCODE_TEMPORAL_ROLE` | `both` | `both` hosts the worker and the client in one process. `client` drives workflows without a worker. `worker` runs activities with no HTTP surface. | | `OPENCODE_SESSION_IDLE_TIMEOUT` | | How long an idle session's workflow stays open. Local mode honors the same variable. | +| `OPENCODE_TEMPORAL_STEPPED` | off | `1` runs each step as a model call, one activity per tool call, and a seal. | +| `OPENCODE_TEMPORAL_SERIAL_TOOLS` | derived | Run a step's tool calls one at a time. On by default only where two hosts could write one step's tree: a shared store. | | `OPENCODE_EVENT_POLL_MS` | | How often a live subscriber re-reads the log for events another process appended. `0` turns the tick off. | | `OPENCODE_DB` | | One absolute path shared by every process on a host. | | `OPENCODE_DB_URL`, `OPENCODE_DB_AUTH_TOKEN` | | A libSQL URL for a store shared across hosts. Takes precedence over `OPENCODE_DB`. | @@ -92,6 +94,55 @@ tagged `RunError`, encoded through a `Schema.Union` of every member in `run-erro A user decline inside an activity is a non-retryable failure, so the workflow does not re-drive a turn the user stopped. +## A step as three activities + +`OPENCODE_TEMPORAL_STEPPED=1` splits a step into three activities instead of one: + +``` +runModelCall -> runToolCall (one per call, concurrent) -> sealStep +``` + +`SessionRunner.runModelCall` performs the attempt, records each call as `Tool.Called`, and hands the +calls back rather than running them. `runToolCall` settles one call. `sealStep` takes the end +snapshot, diffs it against the start, and publishes `Step.Ended`. The loop between them is workflow +code (`stepped-turn.ts`), so a retry policy, a timeout, an approval or a budget can sit between the +model asking for a tool and the tool running. Each activity carries its own bounds: sealing does not +inherit a turn-sized backstop, and one tool waiting on a human no longer holds the attempt and its +sibling tools under a single timeout. + +The supervisor is the same. Wake, interrupt, idle self-termination and continue-as-new only ever +called one `runTurnStep`, so the stepped mode supplies a different one. The mode rides the workflow +input, so a session that rolls over keeps it. + +Three rules carry the split: + +- **One owner token per step, not per activity.** The event log fences a publish behind the current + owner, so a step's writers share one token. Only `runModelCall` claims; the tool and seal + activities publish under the token it returns. +- **A call that already started is not silently repeated.** `runToolCall` publishes `Tool.Called` + before it runs the tool, so a call the log shows as running is one a dispatch was already inside. + On a second dispatch only a tool declaring `idempotent` runs again; anything else reaches the model + as an unknown outcome. The attempt number cannot stand in for this: it counts every way a dispatch + can die, including the ones that never reached the tool. +- **A stop closes the calls it cut short, and a hand-off does not.** A dispatch closes its own call + when the cancellation means the turn is over. When it means this attempt is being handed to another + (a worker shutting down, a pause, a heartbeat timeout), the call is left as it was found, or the + next attempt would read a closed call and never run the tool. + +A declined permission crosses the activity boundary as its own error type, so the workflow tells a +refusal apart from a tool that failed. The whole-step path still raises it as an interrupt, which is +the behaviour its own tests pin, so that drain says so at the boundary. + +Each tool call ships the tree from the host that ran it, because the seal can land on any worker and +a capture there would miss what the tool wrote. Where two hosts could run one step's tools, they run +one at a time (`OPENCODE_TEMPORAL_SERIAL_TOOLS`), or the second to ship would publish a tree without +the first's work. + +What the split costs is the overlap between the model's stream and its own tools: a whole-step +activity starts each tool the moment the model asks for it, while here the attempt returns first. +The tools of one step still run concurrently with each other. Against live providers that tail is +under a tenth of a second, and no model tested emitted text after asking for its first tool. + ## Running workers separately By default the serve process hosts both the worker and the client. To scale workers apart from the diff --git a/packages/temporal/src/activities.ts b/packages/temporal/src/activities.ts index 68a37c8d8b03..b085d3eb9b45 100644 --- a/packages/temporal/src/activities.ts +++ b/packages/temporal/src/activities.ts @@ -3,15 +3,17 @@ // crash is detected quickly, and forwards Temporal cancellation as an AbortSignal so an interrupt // turns into Effect fiber interruption inside the runner. -import { heartbeat, Context } from "@temporalio/activity" - -// The event-log owner token for one activity execution: run id + activity id + attempt. A Temporal -// retry of the SAME step gets a fresh attempt, so once the retry claims the log the previous attempt -// (if still running) is fenced out. The activity id is essential: activity attempt numbers restart -// at 1 for every step, so a run-id+attempt token alone would repeat across steps (step 1 attempt 1 -// and step 2 attempt 1 both mint `run#1`), letting a zombie attempt from an earlier step re-match -// the current owner and append stale events. Including the per-execution activity id keeps every -// step's tokens disjoint. +import { heartbeat, CancelledFailure, Context } from "@temporalio/activity" + +// The SDK's cancel reasons that mean nothing is coming back for this call: the workflow asked to +// stop, or it is gone. The rest (a worker shutting down, a pause, a heartbeat timeout, a reset) +// hand the same call to another attempt. They reach an activity as the reason on the abort, which +// every SDK version sets, and as `cancellationDetails`, which only a server that sends them does. +const turnEndingReasons = ["CANCELLED", "NOT_FOUND"] + +// The event-log owner token for one activity execution. A retry of the same step gets a fresh +// attempt, so once it claims the log the attempt before it is fenced out. The activity id is what +// keeps steps apart: attempt numbers restart at 1 on every step, so a token without it repeats. export function ownerTokenFrom(run: string, activityId: string, attempt: number): string { return `${run}:${activityId}:${attempt}` } @@ -28,29 +30,90 @@ function ownerToken(): string { // The step contract lives with the drain in core; re-exported here so the workflow and its tests // keep one import site inside this package. import type { StepDrainInput, StepDrainResult } from "./drain" +import type { + ModelCallDrainInput, + ModelCallDrainResult, + SealDrainInput, + ToolCallDrainInput, + ToolCallDrainResult, +} from "./stepped-drain" export type { StepDrainInput, StepDrainResult } export type StepActivities = { runTurnStep(input: StepDrainInput): Promise } +// Heartbeat while a drain runs so a dead worker is noticed quickly. This only proves the process is +// alive, not that the work is progressing. +const withHeartbeat = async (body: () => Promise): Promise => { + const beat = setInterval(() => { + try { + heartbeat() + } catch { + // heartbeat outside an activity context is a no-op for our purposes + } + }, 3000) + try { + return await body() + } finally { + clearInterval(beat) + } +} + export function makeStepActivities( stepDrain: (input: StepDrainInput, signal: AbortSignal) => Promise, ): StepActivities { return { async runTurnStep(input) { - const beat = setInterval(() => { - try { - heartbeat() - } catch { - // heartbeat outside an activity context is a no-op for our purposes - } - }, 3000) - try { - return await stepDrain({ ...input, owner: ownerToken() }, Context.current().cancellationSignal) - } finally { - clearInterval(beat) + return withHeartbeat(() => stepDrain({ ...input, owner: ownerToken() }, Context.current().cancellationSignal)) + }, + } +} + +export type SteppedTurnActivities = { + runModelCall(input: ModelCallDrainInput): Promise + runToolCall(input: ToolCallDrainInput): Promise + sealStep(input: SealDrainInput): Promise +} + +// The three activities of a stepped step. Only the model call mints and claims an owner token: it is +// the writer that supersedes the attempt before it, and the tool and seal activities publish under +// the token it returned. Giving each of them its own token would make a step's writers fence each +// other out of the log. +export function makeSteppedTurnActivities(drains: { + modelCallDrain: ( + input: ModelCallDrainInput & { readonly owner: string }, + signal: AbortSignal, + ) => Promise + toolCallDrain: ( + input: ToolCallDrainInput, + signal: AbortSignal, + turnEnded: () => boolean, + ) => Promise + sealDrain: (input: SealDrainInput, signal: AbortSignal) => Promise +}): SteppedTurnActivities { + return { + async runModelCall(input) { + return withHeartbeat(() => + drains.modelCallDrain({ ...input, owner: ownerToken() }, Context.current().cancellationSignal), + ) + }, + async runToolCall(input) { + const context = Context.current() + // An attempt that is being handed on must leave the call as it found it, or its successor + // reads a closed call and never runs the tool. A superseded attempt learns of it the same way + // a stopped one does; closing a call its successor is re-running costs the model that result, + // which is the smaller loss. + const turnEnded = () => { + const details = context.cancellationDetails + if (details) return details.cancelRequested || details.notFound + const reason: unknown = context.cancellationSignal.reason + return reason instanceof CancelledFailure && turnEndingReasons.includes(reason.message ?? "") } + return withHeartbeat(() => drains.toolCallDrain(input, context.cancellationSignal, turnEnded)) + }, + async sealStep(input) { + return withHeartbeat(() => drains.sealDrain(input, Context.current().cancellationSignal)) }, } } diff --git a/packages/temporal/src/boundary.ts b/packages/temporal/src/boundary.ts new file mode 100644 index 000000000000..e413da286394 --- /dev/null +++ b/packages/temporal/src/boundary.ts @@ -0,0 +1,82 @@ +// Crossing from Effect into an activity result. Every drain in this package ends here, so the way a +// failure is classified is decided in one place: +// - an interrupt caused by the driver's own cancellation rethrows the abort reason, so the +// attempt records Cancelled rather than Failed +// - a run the user refused (a declined permission) crosses non-retryable under its own type, or +// the supervisor re-drives a turn the user explicitly stopped +// - a genuine run error crosses non-retryable with the RunError encoded in `details`, so the +// caller reconstructs the exact typed error instead of a string +// Only crashes and task timeouts (never thrown here) go through the activity retry policy. + +import { Cause, Effect, Exit } from "effect" +import { ApplicationFailure } from "@temporalio/activity" +import { SessionSchema } from "@opencode-ai/core/session/schema" +import { SessionRunDeclinedError } from "@opencode-ai/core/session/error" +import { encodeRunError } from "./run-error-codec" +import { HALTED_FAILURE_TYPE } from "./protocol" + +export interface BoundaryOptions { + /** Whether an interrupt the body raises on its own, with nothing cancelling it, means the user + * refused the run. The whole-step path says yes: it turns a decline into an interrupt to halt + * the loop, which is the local behaviour V1 defined and its own tests pin. The stepped path says + * no, because its dispatch names the decline, so an interrupt there is the runner stopping for + * some other reason and reading it as a refusal would put words in the user's mouth. */ + readonly declineIsInterrupt?: boolean +} + +const halted = (sessionID: string, declined?: SessionRunDeclinedError) => { + const encoded = encodeRunError( + declined ?? new SessionRunDeclinedError({ sessionID: SessionSchema.ID.make(sessionID) }), + ) + return ApplicationFailure.create({ + message: "session run halted (user declined)", + type: HALTED_FAILURE_TYPE, + nonRetryable: true, + details: encoded === undefined ? undefined : [encoded], + }) +} + +// Failures that say something about the moment rather than about the work: storage that was not +// reachable, a defect from a database call that `orDie` turned into one. Everything else stays +// non-retryable, because re-running a step whose input the model already answered is worse than +// failing it. A libsql blip during a seal would otherwise fail the step for good rather than move +// it to another worker. +const TRANSIENT = new Set([ + "ToolOutputStore.StorageError", + "SqlError", + "SqliteError", + // A rebuild that did not finish. git and the filesystem fail for reasons that pass, and the + // alternative is a turn failing for good because one worker had a bad minute. + "WorktreeMaterializer.MaterializeError", +]) + +export const runAtBoundary = async ( + sessionID: string, + signal: AbortSignal, + body: Effect.Effect, + options?: BoundaryOptions, +): Promise => { + const exit = await Effect.runPromiseExit(body, { signal }) + if (Exit.isSuccess(exit)) return exit.value + const cause = exit.cause + if (Cause.hasInterruptsOnly(cause)) { + if (signal.aborted) throw signal.reason instanceof Error ? signal.reason : new Error("session run interrupted") + if (options?.declineIsInterrupt) throw halted(sessionID) + // Nothing cancelled this run and nothing named a reason: the runner stopped itself, which a + // session handed to a runner bound to another location does. + throw ApplicationFailure.create({ + message: "session run interrupted with no cancellation", + type: "SessionRunInterrupted", + nonRetryable: true, + }) + } + const squashed = Cause.squash(cause) as { _tag?: string; message?: string } + if (squashed instanceof SessionRunDeclinedError) throw halted(sessionID, squashed) + const encoded = encodeRunError(squashed) + throw ApplicationFailure.create({ + message: squashed?.message ?? Cause.pretty(cause), + type: squashed?._tag ?? "SessionRunError", + nonRetryable: !(squashed?._tag !== undefined && TRANSIENT.has(squashed._tag)), + details: encoded === undefined ? undefined : [encoded], + }) +} diff --git a/packages/temporal/src/config.ts b/packages/temporal/src/config.ts index 6d4c7c70de3e..6953e4c81f7b 100644 --- a/packages/temporal/src/config.ts +++ b/packages/temporal/src/config.ts @@ -18,14 +18,33 @@ export interface Interface { readonly role: Role /** Override for the supervisor's idle self-termination; local mode honors the same variable. */ readonly idleTimeout?: string + /** Drive each step as a provider attempt, one activity per tool call, and a seal. Off by default: + * the whole-step mode is what runs today, and this only changes how new sessions start. */ + readonly stepped?: boolean + /** Run a step's tool calls one at a time instead of together. Tools of one step write the same + * tree and each ships from the host that ran it, so two on two hosts each publish a tree without + * the other's work: the second is refused rather than reverting the first, which leaves its work + * stranded there. `OPENCODE_TEMPORAL_SERIAL_TOOLS=1` forces it on; it is on by default only where + * the store is shared, which is what lets one step's tools land on different workers. */ + readonly serialTools?: boolean } export class Service extends Context.Service()("@opencode/temporal/Config") {} -export const fromEnv = (): Interface => ({ - address: process.env.TEMPORAL_ADDRESS ?? DEFAULTS.address, - namespace: process.env.TEMPORAL_NAMESPACE ?? DEFAULTS.namespace, - taskQueue: process.env.OPENCODE_TEMPORAL_TASK_QUEUE ?? DEFAULTS.taskQueue, - role: (process.env.OPENCODE_TEMPORAL_ROLE as Role | undefined) ?? "both", - idleTimeout: process.env.OPENCODE_SESSION_IDLE_TIMEOUT, -}) +const given = (name: string) => process.env[name] !== undefined && process.env[name] !== "" +const onOff = (name: string, fallback: boolean) => (given(name) ? process.env[name] === "1" : fallback) + +export const fromEnv = (): Interface => { + const sharedStore = !!process.env.OPENCODE_DB_URL + return { + address: process.env.TEMPORAL_ADDRESS ?? DEFAULTS.address, + namespace: process.env.TEMPORAL_NAMESPACE ?? DEFAULTS.namespace, + taskQueue: process.env.OPENCODE_TEMPORAL_TASK_QUEUE ?? DEFAULTS.taskQueue, + role: (process.env.OPENCODE_TEMPORAL_ROLE as Role | undefined) ?? "both", + idleTimeout: process.env.OPENCODE_SESSION_IDLE_TIMEOUT, + stepped: onOff("OPENCODE_TEMPORAL_STEPPED", false), + // Serial by default only where two hosts can end up writing one step's tree: a shared store, + // with nothing keeping the step's tools on one worker. + serialTools: onOff("OPENCODE_TEMPORAL_SERIAL_TOOLS", sharedStore), + } +} diff --git a/packages/temporal/src/drain.ts b/packages/temporal/src/drain.ts index fa692c74e54d..ed96d6d806d2 100644 --- a/packages/temporal/src/drain.ts +++ b/packages/temporal/src/drain.ts @@ -1,11 +1,10 @@ -// The per-step drain body for the Temporal layer: it runs inside the runTurnStep activity. It wraps -// one SessionRunner.runStep, claims the event log for the attempt, ensures the worktree, and encodes -// the error for the activity boundary. Local mode does not use this: it runs whole turns through -// SessionRunner.run on the SessionRunCoordinator (execution/local.ts). Both modes go through the -// same SessionRunner and the same durable event log. +// The drain bodies the Temporal activities run: one whole step, or the three units a stepped turn +// is made of (the provider attempt, one tool call, the seal). Each wraps a SessionRunner call for +// the activity boundary: find the session, claim the event log for the attempt, bring the project +// tree forward, and run under the session's location. Local mode does not use these; it runs whole +// turns through SessionRunner.run on the coordinator. Both modes share SessionRunner and the log. -import { Cause, Context, Effect, Exit, type LayerMap } from "effect" -import { ApplicationFailure } from "@temporalio/activity" +import { Context, Effect, type LayerMap } from "effect" import type { LocationServiceMap } from "@opencode-ai/core/location-service-map" import type { Location } from "@opencode-ai/core/location" import type { LocationError, LocationServices } from "@opencode-ai/core/location-services" @@ -14,9 +13,9 @@ import { WorktreeMaterializer } from "@opencode-ai/core/session/execution/worktr import { SessionRunner } from "@opencode-ai/core/session/runner" import { SessionSchema } from "@opencode-ai/core/session/schema" import { SessionStore } from "@opencode-ai/core/session/store" -import { SessionRunDeclinedError } from "@opencode-ai/core/session/error" import type { SessionInput } from "@opencode-ai/core/session/input" -import { encodeRunError } from "./run-error-codec" +import { runAtBoundary } from "./boundary" + // One step of a turn, as any executor drives it. `promotion` is null (not undefined) so it // serializes cleanly across an executor's process boundary. export interface StepDrainInput { @@ -35,6 +34,13 @@ export interface StepDrainResult { promotion: string | null } +/** The step result as it crosses the activity boundary. A missing session reads as a step with + * nothing left to do, which is what a resume waiting on it should see. */ +export const toStepResult = (step: number, result?: SessionRunner.StepResult): StepDrainResult => + result === undefined + ? { continue: false, step, promotion: null } + : { continue: result.continue, step: result.step, promotion: result.promotion ?? null } + export interface DrainDeps { readonly store: SessionStore.Interface readonly locations: LayerMap.LayerMap @@ -47,60 +53,60 @@ export interface DrainDeps { readonly worktrees: WorktreeMaterializer.Interface } -export const makeDrains = ({ store, locations, ctx, events, worktrees }: DrainDeps) => { +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 +} + +/** Run `use` against a session's runner. Undefined when the session is gone, which is not a run + * error: deleting a session while its workflow is alive is allowed, and a resume waiting on it + * should resolve rather than reject. */ +export type InSession = ( + sessionID: string, + owner: string | undefined, + options: InSessionOptions, + use: (runner: SessionRunner.Interface, session: SessionSchema.Info) => Effect.Effect, +) => Effect.Effect + +export const makeInSession = + ({ store, locations, ctx, events, worktrees }: DrainDeps): InSession => + (sessionID, owner, options, use) => + Effect.gen(function* () { + const session = yield* store.get(SessionSchema.ID.make(sessionID)) + if (!session) return undefined + // 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) + return yield* SessionRunner.Service.use((runner) => use(runner, session)).pipe( + Effect.provide(locations.get(session.location)), + ) + }).pipe(Effect.provideService(EventV2.EventOwner, owner), Effect.provide(ctx), Effect.scoped) + +export const makeDrains = (deps: DrainDeps) => { + const inSession = makeInSession(deps) + // Run exactly one step of the turn (the supervisor loops it); returns the next loop state. - const stepDrain = async (input: StepDrainInput, signal: AbortSignal): Promise => { - const exit = await Effect.runPromiseExit( - Effect.gen(function* () { - const session = yield* store.get(SessionSchema.ID.make(input.sessionID)) - if (!session) return { continue: false, step: input.step, promotion: null } - // Take the event log before running so a superseded attempt's later appends are fenced. - if (input.owner) yield* events.claim(session.id, input.owner) - // A worker resuming on a host without the project tree rebuilds it from snapshot packs. - yield* worktrees.ensure(session.location.directory) - const r = yield* SessionRunner.Service.use((runner) => - runner.runStep({ + const stepDrain = async (input: StepDrainInput, signal: AbortSignal): Promise => + runAtBoundary( + input.sessionID, + signal, + inSession(input.sessionID, input.owner, { claim: input.owner !== undefined }, (runner, session) => + runner + .runStep({ sessionID: session.id, step: input.step, promotion: (input.promotion ?? undefined) as SessionInput.Delivery | undefined, first: input.first, force: input.force, - }), - ).pipe(Effect.provide(locations.get(session.location))) - return { continue: r.continue, step: r.step, promotion: r.promotion ?? null } - }).pipe(Effect.provideService(EventV2.EventOwner, input.owner), Effect.provide(ctx), Effect.scoped), - { signal }, + }) + .pipe(Effect.map((result) => toStepResult(input.step, result))), + ).pipe(Effect.map((result) => result ?? toStepResult(input.step))), + // A whole step turns a declined permission into an interrupt to halt its own loop, so an + // interrupt with nothing cancelling it is that refusal and nothing else. + { declineIsInterrupt: true }, ) - if (Exit.isSuccess(exit)) return exit.value - const cause = exit.cause - if (Cause.hasInterruptsOnly(cause)) { - // Two interrupt sources: driver cancellation (the AbortSignal fired; rethrow its reason so - // the attempt records Cancelled, not Failed) and an internal halt like a user declining a - // permission (the signal did NOT fire). The latter must be non-retryable, or the supervisor - // re-drives a turn the user explicitly stopped. - if (signal.aborted) throw signal.reason instanceof Error ? signal.reason : new Error("session run interrupted") - const declined = encodeRunError( - new SessionRunDeclinedError({ sessionID: SessionSchema.ID.make(input.sessionID) }), - ) - throw ApplicationFailure.create({ - message: "session run halted (user declined)", - type: "SessionRunDeclined", - nonRetryable: true, - details: declined === undefined ? undefined : [declined], - }) - } - // A genuine run error is thrown non-retryable so Temporal surfaces it (to resume) rather than - // retrying; only crashes / task timeouts (never thrown here) go through the retry policy. The - // error is encoded faithfully in `details` so the caller can reconstruct the exact RunError. - const squashed = Cause.squash(cause) as { _tag?: string; message?: string } - const encoded = encodeRunError(squashed) - throw ApplicationFailure.create({ - message: squashed?.message ?? Cause.pretty(cause), - type: squashed?._tag ?? "SessionRunError", - nonRetryable: true, - details: encoded === undefined ? undefined : [encoded], - }) - } - return { stepDrain } + return { inSession, stepDrain } } diff --git a/packages/temporal/src/executor.ts b/packages/temporal/src/executor.ts index 69acc86ad57d..416f5b5aaa25 100644 --- a/packages/temporal/src/executor.ts +++ b/packages/temporal/src/executor.ts @@ -13,8 +13,9 @@ import { makeGlobalNode } from "@opencode-ai/core/effect/app-node" import { SessionSchema } from "@opencode-ai/core/session/schema" import { SessionStore } from "@opencode-ai/core/session/store" import { SessionExecution } from "@opencode-ai/core/session/execution" -import { makeStepActivities } from "./activities" +import { makeStepActivities, makeSteppedTurnActivities } from "./activities" import { makeDrains } from "./drain" +import { makeSteppedDrains } from "./stepped-drain" import { WorktreeMaterializer } from "@opencode-ai/core/session/execution/worktree" import { toRunError } from "./run-error-codec" import * as WF from "./workflow" @@ -58,12 +59,19 @@ const layer = Layer.effect( // Same knob local mode honors; the workflow sandbox cannot read env, so the client forwards the // override as a workflow argument. const IDLE_TIMEOUT = config.idleTimeout + const STEPPED = config.stepped === true + // Only the client can read whether the store is shared, so whether a step's tools may overlap + // is decided here and rides the workflow input. + const SERIAL_TOOLS = config.serialTools === true const events = yield* EventV2.Service const worktrees = yield* WorktreeMaterializer.Service // The per-step drain (drain.ts) wraps SessionRunner.runStep for the activity boundary. Local // mode runs whole turns through SessionRunner.run on the coordinator; both share SessionRunner. - const { stepDrain } = makeDrains({ store, locations, ctx, events, worktrees }) + const drains = makeDrains({ store, locations, ctx, events, worktrees }) + // The stepped mode's three drains. Registered unconditionally: which mode a session runs is a + // property of its workflow input, so a worker has to be able to serve either. + const stepped = makeSteppedDrains({ inSession: drains.inSession }) // Worker connection (native) hosts the runTurnStep activity + the workflow. Skipped in // client-only role so serve can run without an embedded worker. @@ -86,7 +94,7 @@ const layer = Layer.effect( namespace: NAMESPACE, taskQueue: TASK_QUEUE, workflowsPath: fileURLToPath(new URL("./workflow.ts", import.meta.url)), - activities: makeStepActivities(stepDrain), + activities: { ...makeStepActivities(drains.stepDrain), ...makeSteppedTurnActivities(stepped) }, }), ) const runHandle = worker.run() @@ -126,7 +134,15 @@ const layer = Layer.effect( client.workflow.signalWithStart(WORKFLOW_TYPE, { taskQueue: TASK_QUEUE, workflowId: workflowId(id), - args: [id, { startWithWake: true, idleTimeout: IDLE_TIMEOUT } satisfies WF.SessionTurnOptions], + args: [ + id, + { + startWithWake: true, + idleTimeout: IDLE_TIMEOUT, + stepped: STEPPED, + serialTools: SERIAL_TOOLS, + } satisfies WF.SessionTurnOptions, + ], signal: WF.wake, signalArgs: [], }), @@ -177,7 +193,15 @@ const layer = Layer.effect( // startWithWake=false: a fresh resume-with-start must not manufacture a wake drain; // its forced drain comes from the resume update. Ignored when USE_EXISTING joins a // running workflow (which keeps its own state). - args: [id, { startWithWake: false, idleTimeout: IDLE_TIMEOUT } satisfies WF.SessionTurnOptions], + args: [ + id, + { + startWithWake: false, + idleTimeout: IDLE_TIMEOUT, + stepped: STEPPED, + serialTools: SERIAL_TOOLS, + } satisfies WF.SessionTurnOptions, + ], workflowIdConflictPolicy: "USE_EXISTING", }) return client.workflow.executeUpdateWithStart(WF.resume, { diff --git a/packages/temporal/src/protocol.ts b/packages/temporal/src/protocol.ts index 3f7807fef94f..b1ed3af1560e 100644 --- a/packages/temporal/src/protocol.ts +++ b/packages/temporal/src/protocol.ts @@ -8,6 +8,11 @@ export const WORKFLOW_TYPE = "sessionTurn" export const SIGNALS = { wake: "wake", interrupt: "interrupt" } as const export const RESUME_UPDATE = "resume" +/** The failure type a drain raises when the user stopped the turn (a declined permission, a + * rejected question). The workflow has to tell it apart from a failed tool, so the name is shared + * rather than spelled twice. */ +export const HALTED_FAILURE_TYPE = "SessionRunDeclined" + export const WORKFLOW_ID_PREFIX = "session-exec-" export const workflowId = (sessionID: string) => `${WORKFLOW_ID_PREFIX}${sessionID}` diff --git a/packages/temporal/src/run-error-codec.ts b/packages/temporal/src/run-error-codec.ts index c7534a25a8b6..fd82dfdce41a 100644 --- a/packages/temporal/src/run-error-codec.ts +++ b/packages/temporal/src/run-error-codec.ts @@ -1,7 +1,3 @@ -// Faithful round-trip of a SessionRunner.RunError across the Temporal boundary. Every member of the -// union is a Schema.TaggedErrorClass, so we can encode the error to JSON in the activity and decode -// it back into the exact tagged instance in the layer, instead of flattening it to a carrier. - import { Schema } from "effect" import { LLMError } from "@opencode-ai/llm" import { Integration } from "@opencode-ai/core/integration" diff --git a/packages/temporal/src/stepped-drain.ts b/packages/temporal/src/stepped-drain.ts new file mode 100644 index 000000000000..2b81a8daca8b --- /dev/null +++ b/packages/temporal/src/stepped-drain.ts @@ -0,0 +1,129 @@ +// The three drain bodies a stepped turn is made of: the provider attempt, one tool call, and the +// seal. A whole-step drain runs one attempt plus all of its tools in a single activity, so nothing +// can sit between the model asking for a tool and the tool running. Splitting them is what gives +// each tool call its own retry policy, timeout and approval window. +// +// All three write to the same session log, so they publish under one owner token: the model call +// claims it and hands it back, the other two inherit it. A token per activity execution, which is +// right when the activity is the whole step, would make a step's writers fence each other out. + +import { Effect } from "effect" +import type { DeferredToolCall, ToolCallOutcome } from "@opencode-ai/core/session/runner" +import type { StepSettlement } from "@opencode-ai/core/session/runner/publish-llm-event" +import type { SessionInput } from "@opencode-ai/core/session/input" +import { runAtBoundary } from "./boundary" +import { type InSession, type StepDrainInput, type StepDrainResult, toStepResult } from "./drain" + +/** The provider attempt of one step. Same shape as a whole-step drain: the difference is what it + * does with the tool calls, not what it needs to start. */ +export type ModelCallDrainInput = StepDrainInput + +export type ModelCallDrainResult = + | { readonly kind: "settled"; readonly result: StepDrainResult } + | { + readonly kind: "called" + readonly step: number + readonly calls: ReadonlyArray + readonly settlement?: StepSettlement + readonly assistantMessageID?: string + readonly needsContinuation?: boolean + /** The event-log token this attempt claimed. The tool and seal activities of this step must + * publish under it, so it travels with the calls instead of being minted again. */ + readonly owner: string + } + +export interface ToolCallDrainInput { + readonly sessionID: string + readonly call: DeferredToolCall + readonly owner: string +} + +export interface ToolCallDrainResult { + readonly outcome: ToolCallOutcome +} + +export interface SealDrainInput { + readonly sessionID: string + readonly step: number + readonly settlement?: StepSettlement + readonly assistantMessageID?: string + readonly needsContinuation?: boolean + readonly owner: string +} + +export interface SteppedDrainDeps { + readonly inSession: InSession +} + +export const makeSteppedDrains = ({ inSession }: SteppedDrainDeps) => { + // Only the model call claims the log: it is the writer that supersedes a previous attempt, and + // the rest of the step rides its token. + const modelCallDrain = async ( + input: ModelCallDrainInput & { readonly owner: string }, + signal: AbortSignal, + ): Promise => + runAtBoundary( + input.sessionID, + signal, + inSession(input.sessionID, input.owner, { claim: true }, (runner, session) => + runner.runModelCall({ + sessionID: session.id, + step: input.step, + promotion: (input.promotion ?? undefined) as SessionInput.Delivery | undefined, + first: input.first, + force: input.force, + }), + ).pipe( + Effect.map( + (result): ModelCallDrainResult => + result === undefined || result.kind === "settled" + ? { kind: "settled", result: toStepResult(input.step, result?.result) } + : { ...result, owner: input.owner }, + ), + ), + ) + + const toolCallDrain = async ( + input: ToolCallDrainInput, + signal: AbortSignal, + /** Whether a cancellation means the turn is over rather than this attempt being handed on. + * Only the first closes the call: a worker shutting down leaves it for the next attempt, which + * has to be free to decide whether the tool may run again. */ + turnEnded: () => boolean = () => false, + ): Promise => + runAtBoundary( + input.sessionID, + signal, + inSession(input.sessionID, input.owner, {}, (runner, session) => + runner.runToolCall({ sessionID: session.id, call: input.call }).pipe( + // A stop landing mid-tool leaves the call recorded as running, where a whole step closes + // the tools it opened before it returns. Nothing else closes it until the next turn's + // entry check, so a transcript would show the call still going long after the stop. + Effect.onInterrupt(() => + turnEnded() + ? runner.failToolCall({ sessionID: session.id, call: input.call }).pipe(Effect.ignore) + : Effect.void, + ), + ), + ).pipe(Effect.map((result) => result ?? { outcome: "already-settled" as const })), + ) + + const sealDrain = async (input: SealDrainInput, signal: AbortSignal): Promise => + runAtBoundary( + input.sessionID, + signal, + inSession(input.sessionID, input.owner, {}, (runner, session) => + runner + .sealStep({ + sessionID: session.id, + step: input.step, + settlement: input.settlement, + assistantMessageID: input.assistantMessageID, + needsContinuation: input.needsContinuation, + }) + .pipe(Effect.map((result) => toStepResult(input.step, result))), + ).pipe(Effect.map((result) => result ?? toStepResult(input.step))), + ) + + return { modelCallDrain, toolCallDrain, sealDrain } +} diff --git a/packages/temporal/src/stepped-turn.ts b/packages/temporal/src/stepped-turn.ts new file mode 100644 index 000000000000..6a575366f029 --- /dev/null +++ b/packages/temporal/src/stepped-turn.ts @@ -0,0 +1,148 @@ +// One step as three units of work instead of one: the provider attempt, each tool call it asks for, +// and the seal that closes it. That is the whole point of the split, and it is workflow code, so the +// model-to-tools loop lives where retries, timers, approvals and budgets can sit between the two. +// +// MUST stay pure, like `supervisor.ts`: this is bundled into the workflow sandbox, so no `effect`, +// no `@opencode-ai/core` runtime imports, no Node builtins. Type-only imports are erased and safe. +// +// What this costs, stated plainly: a whole-step activity starts each tool the moment the model asks +// for it, while the stream is still going. Here the attempt has to return before any tool starts, +// because a workflow cannot consume a stream. The tools of one step still run concurrently with +// each other; what is lost is the overlap between the model and its own tools. + +import { ActivityFailure, type ApplicationFailure } from "@temporalio/workflow" +import { HALTED_FAILURE_TYPE } from "./protocol" +import type { StepDrainInput, StepDrainResult } from "./drain" +import type { + ModelCallDrainInput, + ModelCallDrainResult, + SealDrainInput, + ToolCallDrainInput, + ToolCallDrainResult, +} from "./stepped-drain" + +/** + * A halt the user asked for, as it looks once it has crossed an activity boundary. The runner raises + * it as an interrupt, `boundary.ts` throws it as an `ApplicationFailure`, and the SDK wraps that in + * one `ActivityFailure`. It is not a cancellation, so `isCancellation` says no, and a dispatcher + * that checks only that would treat a refusal as one failed tool and carry on. + * + * A `TimeoutFailure` or a `CancelledFailure` cause has no `type` field, so neither matches. + */ +export const isHaltFailure = (error: unknown) => + error instanceof ActivityFailure && (error.cause as ApplicationFailure | undefined)?.type === HALTED_FAILURE_TYPE + +/** The three activities a stepped turn drives. */ +export interface SteppedActivities { + readonly runModelCall: (input: ModelCallDrainInput) => Promise + readonly runToolCall: (input: ToolCallDrainInput) => Promise + readonly sealStep: (input: SealDrainInput) => Promise +} + +export interface SteppedTurnDeps { + readonly activities: SteppedActivities + /** Whether an error is the driver's cancellation. An interrupt has to end the turn, so it must not + * be swallowed the way a failed tool is. */ + readonly isCancellation: (error: unknown) => boolean + /** Whether an error is the user stopping the turn, like a declined permission. It arrives as an + * ordinary activity failure, so without this it reads as one bad tool and the turn carries on + * past the refusal. */ + readonly isHalt: (error: unknown) => boolean + /** Where a step says what became of the calls it dispatched. A dispatch that decided not to run + * a tool, or could not keep its result, is a step's most surprising outcome and the least + * visible: it reads as an ordinary success everywhere else. */ + readonly log?: (message: string, attributes: Record) => void + /** Run the calls one at a time. Each tool ships the tree from the host that ran it, so two on two + * hosts each publish a tree without the other's work and the second is refused, leaving its work + * stranded there. Serial is what moving files between hosts costs. */ + readonly serial?: boolean + /** Run something where the driver's cancellation cannot reach it. An interrupt landing during the + * tool phase otherwise leaves the step with no ending published at all, so a follower waiting on + * the turn never hears it stop. */ + readonly nonCancellable?: (fn: () => Promise) => Promise +} + +/** + * Drives one step and reports the next loop state, so it drops straight into the supervisor in place + * of a whole-step activity. + */ +export const makeSteppedTurn = + ({ activities, isCancellation, isHalt, log, serial, nonCancellable }: SteppedTurnDeps) => + async (input: StepDrainInput): Promise => { + const model = await activities.runModelCall(input) + // A crashed step finalized from the log, or the recovery gate finding no work: the step is over + // and there is nothing to dispatch or seal. + if (model.kind === "settled") return model.result + + // Each call is its own unit of work. A tool that fails outright does not take the turn with it: + // the seal closes its call as an error and the model gets to react, which is better than losing + // the step. A cancel and a user halt are different, and both have to propagate. + const dispatch = (call: (typeof model.calls)[number]) => + activities.runToolCall({ sessionID: input.sessionID, call, owner: model.owner }) + const dispatched: PromiseSettledResult[] = [] + if (serial) { + // One at a time, and still settled rather than thrown, so a tool that fails does not take the + // rest of the batch with it. The loop keeps going: the seal closes each call and the model + // reacts to what it is told. + for (const call of model.calls) { + dispatched.push( + await dispatch(call).then( + (value) => ({ status: "fulfilled", value }) as const, + (reason) => ({ status: "rejected", reason }) as const, + ), + ) + } + } else { + dispatched.push(...(await Promise.allSettled(model.calls.map(dispatch)))) + } + + const sealing = (stopped: boolean): SealDrainInput => ({ + sessionID: input.sessionID, + step: model.step, + // A stopped step is not one that continues. The settlement carries the model's own finish + // reason, and for a step that asked for tools that is `tool-calls`, which every follower + // reads as "another step follows". Passing it through on the way out would record a turn the + // user stopped as a turn still going. + settlement: stopped && model.settlement ? { ...model.settlement, finish: "stop" } : model.settlement, + assistantMessageID: model.assistantMessageID, + needsContinuation: stopped ? false : model.needsContinuation, + owner: model.owner, + }) + const seal = (stopped: boolean) => activities.sealStep(sealing(stopped)) + + for (const outcome of dispatched) { + if (outcome.status !== "rejected") continue + if (isCancellation(outcome.reason) || isHalt(outcome.reason)) { + // Seal on the way out, out of reach of the cancellation, so the calls that did return keep + // their results and the step is recorded as ended. Without it a stop landing during the + // tools publishes no step event at all: the interrupt is only visible during the model + // call, and a follower waiting on the turn hangs. + // + // Explicitly with no continuation. A seal left to decide re-derives "keep going" from the + // tool parts and would carry the agent on past a declined permission. The reason the turn + // is stopping is rethrown either way, and a seal that fails here must not replace it. + await (nonCancellable ?? ((fn: () => Promise) => fn()))(() => seal(true)).catch((err) => + log?.("could not seal an interrupted step", { step: model.step, error: String(err) }), + ) + throw outcome.reason + } + } + + // A dispatch that settled its call needs no telling. The rest are what an operator is looking + // for when a turn did something unexpected: a tool reported as unknown ran or did not, and + // nothing else in this workflow's history says which call that was. + const unsettled = model.calls + .map((call, index) => { + const result = dispatched[index] + return { + call: call.id, + tool: call.name, + outcome: result?.status === "fulfilled" ? result.value.outcome : "errored", + } + }) + .filter((entry) => entry.outcome !== "settled") + if (unsettled.length > 0) + log?.("step did not settle every call it dispatched", { step: model.step, calls: unsettled }) + + return seal(false) + } diff --git a/packages/temporal/src/workflow.ts b/packages/temporal/src/workflow.ts index f18ac6f93868..badbee6d23a4 100644 --- a/packages/temporal/src/workflow.ts +++ b/packages/temporal/src/workflow.ts @@ -21,7 +21,8 @@ import { log, workflowInfo, } from "@temporalio/workflow" -import type { StepActivities } from "./activities" +import type { StepActivities, SteppedTurnActivities } from "./activities" +import { isHaltFailure, makeSteppedTurn } from "./stepped-turn" import { SIGNALS, RESUME_UPDATE } from "./protocol" import { makeSupervisor, type SupervisorRuntime } from "./supervisor" @@ -38,6 +39,13 @@ const activityOptions = { const { runTurnStep } = proxyActivities(activityOptions) +// The stepped mode's activities. Each is its own activity, so a tool that hangs, or that waits on +// a human, no longer holds the provider attempt and its sibling tools under one shared timeout. +const { runModelCall, runToolCall } = proxyActivities(activityOptions) +// Sealing is a snapshot, a diff and one event. It should not inherit a turn-sized backstop. +const sealOptions = { ...activityOptions, startToCloseTimeout: "10 minutes" } as const +const { sealStep } = proxyActivities(sealOptions) + export const wake = defineSignal(SIGNALS.wake) export const interrupt = defineSignal(SIGNALS.interrupt) export const resume = defineUpdate(RESUME_UPDATE) @@ -93,6 +101,22 @@ const runtime: SupervisorRuntime = { warn: (message, attributes) => log.warn(message, attributes), } +// Same supervisor, different step body: wake, interrupt, idle timeout and continue-as-new are +// unchanged, and only what "one step" means differs. Built per run rather than once, because +// whether a step's tools may overlap rides the workflow input: the sandbox cannot read env. +const steppedRuntime = (serial: boolean): SupervisorRuntime => ({ + ...runtime, + runTurnStep: makeSteppedTurn({ + activities: { runModelCall, runToolCall, sealStep }, + isCancellation, + isHalt: isHaltFailure, + serial, + nonCancellable: (fn) => CancellationScope.nonCancellable(fn), + // The SDK's logger, so a line carries its workflow and run id and is suppressed on replay. + log: (message, attributes) => log.info(message, attributes), + }), +}) + // The scope of the drain currently running, so an interrupt signal can cancel exactly that turn. let activeDrainScope: CancellationScope | undefined // The workflow's root scope, captured at entry, to detect a whole-run cancellation. @@ -107,18 +131,30 @@ const workflows = makeSupervisor(runtime) export interface SessionTurnOptions { readonly startWithWake?: boolean readonly idleTimeout?: string + /** Drive each step as a provider attempt, one activity per tool call, and a seal, instead of one + * activity for the whole step. Off by default: the whole-step mode is what runs today. */ + readonly stepped?: boolean + /** Run a step's tool calls one at a time. Each ships the tree from the host that ran it, so two + * on two hosts each publish a tree without the other's work. The client decides, because only it + * can read whether the store is shared. */ + readonly serialTools?: boolean } export async function sessionTurn(sessionID: string, options?: SessionTurnOptions): Promise { rootScope = CancellationScope.current() const startWithWake = options?.startWithWake ?? true const idleTimeout = options?.idleTimeout - if (!idleTimeout) return workflows.sessionTurn(sessionID, startWithWake) + const stepped = options?.stepped === true + const serialTools = options?.serialTools === true + if (!idleTimeout && !stepped) return workflows.sessionTurn(sessionID, startWithWake) return makeSupervisor( { - ...runtime, - continueAsNew: (id, wake) => continueAsNew(id, { startWithWake: wake, idleTimeout }), + ...(stepped ? steppedRuntime(serialTools) : runtime), + // The mode has to survive the boundary, or a long session silently reverts to whole-step + // activities the first time it rolls over. + continueAsNew: (id, wake) => + continueAsNew(id, { startWithWake: wake, idleTimeout, stepped, serialTools }), }, - { idleTimeout }, + idleTimeout ? { idleTimeout } : undefined, ).sessionTurn(sessionID, startWithWake) } diff --git a/packages/temporal/test/session-execution-temporal-contract.test.ts b/packages/temporal/test/session-execution-temporal-contract.test.ts index f87d5fa50236..d933f85bc006 100644 --- a/packages/temporal/test/session-execution-temporal-contract.test.ts +++ b/packages/temporal/test/session-execution-temporal-contract.test.ts @@ -15,7 +15,17 @@ if (process.env.OPENCODE_CONTRACT_TEMPORAL === "1") { // One task queue per run: a stale worker from an earlier run against the same dev server would // otherwise steal activities and answer with its own (differently mocked) graph. process.env.OPENCODE_TEMPORAL_TASK_QUEUE ??= `contract-${crypto.randomUUID()}` + // The stepped mode has to satisfy the SAME contract, or the two Temporal modes drift and the + // parity story only holds for the one that happens to be tested. Which mode this run covers comes + // from the same env the executor reads, so the label never disagrees with what actually ran: + // + // OPENCODE_CONTRACT_TEMPORAL=1 -> whole step per activity + // OPENCODE_CONTRACT_TEMPORAL=1 OPENCODE_TEMPORAL_STEPPED=1 -> model call, per tool, seal + const stepped = process.env.OPENCODE_TEMPORAL_STEPPED === "1" // Imported inside the opt-in so a plain `bun test` never loads the Temporal SDK. const { SessionExecutionTemporal } = await import("@opencode-ai/temporal/executor") - runContract("temporal driver", makeExecutionFor(SessionExecutionTemporal.node)) + runContract( + stepped ? "temporal driver (stepped)" : "temporal driver", + makeExecutionFor(SessionExecutionTemporal.node), + ) } diff --git a/packages/temporal/test/stepped-turn.test.ts b/packages/temporal/test/stepped-turn.test.ts new file mode 100644 index 000000000000..2a256910405c --- /dev/null +++ b/packages/temporal/test/stepped-turn.test.ts @@ -0,0 +1,249 @@ +// Deterministic unit tests for the stepped turn body (src/stepped-turn.ts) driven by fake activities -- +// no Temporal, no DB. This is the piece that turns one step into three units of work, so what is +// pinned here is the orchestration: the owner token reaches every writer, a settled step dispatches +// nothing, a failed tool still lets the step close, and an interrupt is not swallowed. +import { describe, it, expect } from "bun:test" +import { isHaltFailure, makeSteppedTurn, type SteppedActivities } from "../src/stepped-turn" +import { runAtBoundary } from "../src/boundary" +import { SessionRunDeclinedError } from "@opencode-ai/core/session/error" +import { SessionSchema } from "@opencode-ai/core/session/schema" +import { Effect } from "effect" +import { ActivityFailure, ApplicationFailure, CancelledFailure, TimeoutFailure } from "@temporalio/workflow" +import type { StepDrainInput, StepDrainResult } from "../src/activities" +import type { + ModelCallDrainResult, + SealDrainInput, + ToolCallDrainInput, + ToolCallDrainResult, +} from "../src/stepped-drain" + +class FakeCancel extends Error {} +class FakeHalt extends Error {} +const isCancellation = (e: unknown) => e instanceof FakeCancel +const isHalt = (e: unknown) => e instanceof FakeHalt + +const INPUT: StepDrainInput = { + sessionID: "ses_1", + step: 2, + promotion: null, + first: false, + force: false, +} +const SEALED: StepDrainResult = { continue: true, step: 3, promotion: "steer" } +const call = (id: string, name = "probe_write") => ({ id, name, assistantMessageID: "msg_1" }) + +const fakes = ( + model: ModelCallDrainResult, + onTool: (input: ToolCallDrainInput) => Promise = async () => ({ + outcome: "settled", + }), +) => { + const tools: ToolCallDrainInput[] = [] + const seals: SealDrainInput[] = [] + const activities: SteppedActivities = { + runModelCall: async () => model, + runToolCall: async (input) => { + tools.push(input) + return onTool(input) + }, + sealStep: async (input) => { + seals.push(input) + return SEALED + }, + } + return { activities, tools, seals } +} + +describe("stepped turn", () => { + it("dispatches nothing and does not seal when the step is already settled", async () => { + const settled: StepDrainResult = { continue: false, step: 2, promotion: null } + const { activities, tools, seals } = fakes({ kind: "settled", result: settled }) + + const result = await makeSteppedTurn({ activities, isCancellation, isHalt })(INPUT) + + // A crashed step finalized from the log, or the recovery gate finding no work: there is nothing + // to run and nothing to close. + expect(result).toEqual(settled) + expect(tools).toHaveLength(0) + expect(seals).toHaveLength(0) + }) + + it("runs every call under the attempt's owner and then seals", async () => { + const { activities, tools, seals } = fakes({ + kind: "called", + step: 2, + calls: [call("call_a"), call("call_b")], + settlement: { finish: "tool-calls", tokens: { input: 1, output: 2, reasoning: 0, cache: { read: 0, write: 0 } } }, + owner: "run-1:model-1:1", + }) + + const result = await makeSteppedTurn({ activities, isCancellation, isHalt })(INPUT) + + expect(tools.map((t) => t.call.id)).toEqual(["call_a", "call_b"]) + // Every writer in the step publishes under the token the attempt claimed. A token minted per + // activity would make them fence each other out of the log. + expect(tools.map((t) => t.owner)).toEqual(["run-1:model-1:1", "run-1:model-1:1"]) + expect(seals).toHaveLength(1) + expect(seals[0]?.owner).toBe("run-1:model-1:1") + // The finish reason lives only in the attempt's memory, so the seal has to be handed it. + expect(seals[0]?.settlement?.finish).toBe("tool-calls") + expect(result).toEqual(SEALED) + }) + + it("reports the calls it did not settle, and says nothing when they all settled", async () => { + const lines: Array<{ message: string; attributes: Record }> = [] + const log = (message: string, attributes: Record) => lines.push({ message, attributes }) + const model: ModelCallDrainResult = { + kind: "called", + step: 2, + calls: [call("call_a"), call("call_b", "probe_read")], + owner: "own", + } + const settling = fakes(model) + await makeSteppedTurn({ ...settling, isCancellation, isHalt, log })(INPUT) + // Nothing surprising happened, so nothing is said about it. + expect(lines).toHaveLength(0) + + const skipping = fakes(model, async (input) => ({ + outcome: input.call.id === "call_a" ? "unknown" : "settled", + })) + await makeSteppedTurn({ ...skipping, isCancellation, isHalt, log })(INPUT) + + // Which call was skipped, and why, is the dispatch's own knowledge: the log records what the + // model was told, and the workflow's history records an activity that succeeded. + expect(lines).toHaveLength(1) + expect(lines[0]?.attributes).toEqual({ + step: 2, + calls: [{ call: "call_a", tool: "probe_write", outcome: "unknown" }], + }) + }) + + it("seals even when a tool call fails outright", async () => { + const { activities, tools, seals } = fakes( + { kind: "called", step: 2, calls: [call("call_a"), call("call_b")], owner: "own" }, + async (input) => { + if (input.call.id === "call_a") throw new Error("activity exhausted its retries") + return { outcome: "settled" } + }, + ) + + const result = await makeSteppedTurn({ activities, isCancellation, isHalt })(INPUT) + + // One broken tool must not take the turn with it: the seal closes that call as an error and the + // model gets to react, which beats losing the step. + expect(tools).toHaveLength(2) + expect(seals).toHaveLength(1) + expect(result).toEqual(SEALED) + }) + + // Each tool ships the project tree from the host that ran it, so two on two hosts each publish a + // tree without the other's work. Serial is what moving files between hosts costs; when nothing is + // moving, the calls still overlap, which is the case the split was measured on. + for (const serial of [true, false]) { + it(`runs a step's tools ${serial ? "one at a time when told to" : "together when it is not"}`, async () => { + let inFlight = 0 + let overlapped = false + const { activities, tools } = fakes( + { kind: "called", step: 2, calls: [call("a"), call("b"), call("c")], owner: "own" }, + async () => { + inFlight += 1 + if (inFlight > 1) overlapped = true + await new Promise((resolve) => setTimeout(resolve, 5)) + inFlight -= 1 + return { outcome: "settled" } + }, + ) + + await makeSteppedTurn({ activities, isCancellation, isHalt, serial })(INPUT) + expect(tools).toHaveLength(3) + expect(overlapped).toBe(!serial) + }) + } + + it("closes an interrupted step without letting it ask for another", async () => { + const { activities, seals } = fakes( + { kind: "called", step: 2, calls: [call("call_a")], owner: "own" }, + async () => { + throw new FakeCancel("interrupted") + }, + ) + + const run = makeSteppedTurn({ activities, isCancellation, isHalt })(INPUT) + + // A cancellation is not a failed tool, so it still ends the turn. The step is closed on the way + // out all the same, or a stop landing here publishes no step event and a follower waiting on the + // turn hangs. What the seal must not do is decide the turn keeps going. + await expect(run).rejects.toBeInstanceOf(FakeCancel) + expect(seals).toHaveLength(1) + expect(seals[0]?.needsContinuation).toBe(false) + }) + + it("closes a halted step without carrying on past the refusal", async () => { + const { activities, seals } = fakes( + { kind: "called", step: 2, calls: [call("call_a")], owner: "own" }, + async () => { + throw new FakeHalt("declined") + }, + ) + + const run = makeSteppedTurn({ activities, isCancellation, isHalt })(INPUT) + + // A decline crosses the activity boundary as an ordinary failure, not a cancel. The halt is + // still what ends the turn, and the seal is told not to continue, which would otherwise let the + // agent run on past the user's refusal. + await expect(run).rejects.toBeInstanceOf(FakeHalt) + expect(seals).toHaveLength(1) + expect(seals[0]?.needsContinuation).toBe(false) + }) +}) + +// The bug this predicate exists for was a mismatch between what `boundary.ts` throws and what the +// dispatcher recognises. Injecting a fake predicate cannot catch that, so match against the real +// failure shapes. The negative cases are the point: a predicate that answered true for everything +// would pass the positive one alone. +describe("halt predicate", () => { + const wrap = (cause?: Error) => + new ActivityFailure("activity failed", "runToolCall", "1", 1 as never, undefined, cause) + + const throwsFrom = (body: Effect.Effect, options?: { declineIsInterrupt: true }) => + runAtBoundary("ses_1", new AbortController().signal, body, options).then( + () => undefined, + (error: unknown) => error, + ) + + it("recognises what the activity boundary throws for a named refusal", async () => { + // Built by running the boundary rather than by hand, so the two sides cannot drift: a dispatch + // reports a decline as this error, and whatever the boundary makes of it is what a dispatcher + // has to recognise. + const thrown = await throwsFrom( + Effect.fail(new SessionRunDeclinedError({ sessionID: SessionSchema.ID.make("ses_1") })), + ) + expect(thrown).toBeInstanceOf(ApplicationFailure) + expect(isHaltFailure(wrap(thrown as Error))).toBe(true) + }) + + it("recognises a whole step's refusal, which arrives as an interrupt", async () => { + const thrown = await throwsFrom(Effect.interrupt, { declineIsInterrupt: true }) + expect(isHaltFailure(wrap(thrown as Error))).toBe(true) + }) + + it("does not read an unexplained interrupt as a refusal", async () => { + // The stepped path names its refusals, so an interrupt with nothing cancelling it is the + // runner stopping for another reason. Calling that a decline would report a user decision + // nobody made. + const thrown = await throwsFrom(Effect.interrupt) + expect(thrown).toBeInstanceOf(ApplicationFailure) + expect((thrown as ApplicationFailure).type).toBe("SessionRunInterrupted") + expect(isHaltFailure(wrap(thrown as Error))).toBe(false) + }) + + it("says no to everything else that can come back from an activity", () => { + expect(isHaltFailure(wrap(new CancelledFailure("cancelled")))).toBe(false) + expect(isHaltFailure(wrap(new TimeoutFailure("timed out", undefined, 1 as never)))).toBe(false) + expect(isHaltFailure(wrap(ApplicationFailure.create({ type: "SessionRunError" })))).toBe(false) + expect(isHaltFailure(wrap())).toBe(false) + // Unwrapped, so not what a dispatcher ever sees. + expect(isHaltFailure(ApplicationFailure.create({ type: "SessionRunDeclined" }))).toBe(false) + expect(isHaltFailure(new Error("plain"))).toBe(false) + }) +})