diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index 9d3145ae96d4..b7829e3cd01e 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -4,7 +4,7 @@ import { Cause, Context, Duration, Effect, Layer, Option, PubSub, Queue, Schema import { Stream } from "effect" import { Event } from "@opencode-ai/schema/event" import type { Data, Definition, Payload } from "@opencode-ai/schema/event" -import { and, asc, eq, gt, inArray } from "drizzle-orm" +import { and, asc, eq, gt, inArray, isNull } from "drizzle-orm" import { Database } from "./database/database" import { EventSequenceTable, EventTable } from "./event/sql" import { Flag } from "./flag/flag" @@ -188,6 +188,41 @@ export interface LayerOptions { /** Chosen to be well under what a person notices in a transcript while staying one cheap indexed * read per subscribed session. In-process commits still wake instantly; this only catches what the * wake cannot see, so it is worth its cost only where another process writes: see `pollingNode`. */ +// Whether the token already on the row is a later attempt of the same activity execution than the +// one claiming. Tokens are `run:activityId:attempt`, so only the attempt is comparable: two +// different activity ids are two different units of work and neither supersedes the other. +const supersededBy = (held: string, claimer: string): boolean => { + const split = (token: string) => { + const cut = token.lastIndexOf(":") + const head = token.slice(0, cut) + const idAt = head.lastIndexOf(":") + const id = head.slice(idAt + 1) + return { + run: head.slice(0, idAt), + id, + // Temporal hands out activity ids as an increasing sequence within a run, so they order the + // units of work. A token from an earlier step is a zombie, whatever its attempt number says. + activity: Number(id), + attempt: Number(token.slice(cut + 1)), + } + } + const a = split(held) + const b = split(claimer) + // Different runs cannot be ordered from the tokens alone, and a continue-as-new legitimately + // starts a new one, so those are allowed through. A zombie from a run that rolled over is the + // case this does not cover. + if (a.run !== b.run) return false + const ordered = Number.isInteger(a.activity) && Number.isInteger(b.activity) + // Activity ids are an increasing sequence when Temporal assigns them, but a caller may set its + // own. Without numbers to compare, two different units of work cannot be ordered, and only two + // attempts of the same one can. + if (ordered && a.activity !== b.activity) return a.activity > b.activity + // Compared as written, not as parsed: two ids that are not numbers both parse to NaN, and NaN + // read as equal made every later activity look like a retry of the one before it. + if (!ordered && a.id !== b.id) return false + return Number.isInteger(a.attempt) && Number.isInteger(b.attempt) && a.attempt > b.attempt +} + const DEFAULT_LIVE_POLL = Duration.seconds(1) // An operator's override, in milliseconds, for either node. Read at layer build rather than at @@ -570,13 +605,64 @@ export const layerWith = (options?: LayerOptions) => .pipe(Effect.orDie) } + // A compare and set, not a write. Two attempts of one activity can be alive at once and they + // do not arrive in order: a paused attempt 1 that resumes after attempt 2 has claimed used to + // take the log back, and then every publish from attempt 2's tool activities died on the + // fence for a step that was going fine. function claim(aggregateID: string, ownerID: string) { - return db - .update(EventSequenceTable) - .set({ owner_id: ownerID }) - .where(eq(EventSequenceTable.aggregate_id, aggregateID)) - .run() - .pipe(Effect.orDie) + return Effect.gen(function* () { + const row = yield* db + .select({ ownerID: EventSequenceTable.owner_id }) + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, aggregateID)) + .get() + .pipe(Effect.orDie) + // No sequence row yet, so there is nothing to fence and nothing to lose a race to: the + // first publish inserts the row with this owner on it. + if (row === undefined) return + if (row.ownerID != null && supersededBy(row.ownerID, ownerID)) { + yield* Effect.die( + new InvalidDurableEventError({ + type: "session.claim", + message: `Stale claim for aggregate ${aggregateID}: held by ${row.ownerID}, claimer ${ownerID}`, + }), + ) + } + // Conditional on what was just read, because the read and the write are two statements + // and over a network store they are two requests. Two attempts of one activity reaching + // here together both passed the check above, and an unconditional write let the loser + // land last and fence out the winner's tools. + yield* db + .update(EventSequenceTable) + .set({ owner_id: ownerID }) + .where( + and( + eq(EventSequenceTable.aggregate_id, aggregateID), + row.ownerID == null + ? isNull(EventSequenceTable.owner_id) + : eq(EventSequenceTable.owner_id, row.ownerID), + ), + ) + .run() + .pipe(Effect.orDie) + // Read back rather than trusting a driver-specific affected-row count. Losing means + // somebody claimed between the two statements, and a loser that carried on would publish + // under a token the fence rejects. + const after = yield* db + .select({ ownerID: EventSequenceTable.owner_id }) + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, aggregateID)) + .get() + .pipe(Effect.orDie) + if (after?.ownerID !== ownerID) { + yield* Effect.die( + new InvalidDurableEventError({ + type: "session.claim", + message: `Lost the claim for aggregate ${aggregateID}: held by ${after?.ownerID}, claimer ${ownerID}`, + }), + ) + } + }) } const subscribe = (definition: D): Stream.Stream> => diff --git a/packages/core/src/session/execution/worktree.ts b/packages/core/src/session/execution/worktree.ts index 9d3532d95b8d..a4f2a9b5e0f7 100644 --- a/packages/core/src/session/execution/worktree.ts +++ b/packages/core/src/session/execution/worktree.ts @@ -12,9 +12,9 @@ export * as WorktreeMaterializer from "./worktree" // hosts still cannot see each other's writes, because those are not captured until the step is // sealed. One worker per worktree is what makes a step's tools share a tree. -import { rm, writeFile } from "node:fs/promises" +import { readdir, rm, writeFile } from "node:fs/promises" import path from "path" -import { Cause, Context, Effect, Layer } from "effect" +import { Cause, Context, Effect, Layer, Schema } from "effect" import { ChildProcess } from "effect/unstable/process" import { and, asc, desc, eq } from "drizzle-orm" import { Database } from "../../database/database" @@ -26,6 +26,7 @@ import { Global } from "../../global" import { AppProcess } from "../../process" import { AbsolutePath } from "../../schema" import { SnapshotPackTable } from "../../snapshot/sql" +import { chainHead, isBehind, orderChain } from "../../snapshot/chain" import { readWorktreeTip, writeWorktreeTip } from "../../snapshot/tip" export interface Interface { @@ -33,18 +34,46 @@ export interface Interface { * Make sure the session's directory holds the newest state the shared store has for it, * rebuilding its worktree from stored snapshot packs when it is missing or behind. A directory * with no stored packs, and a tree this host has neither built nor captured from, are left - * alone. Never fails the caller. + * alone. + * + * A rebuild that fails dies with `WorktreeMaterializeError` rather than returning: running the + * step against whatever is in the directory tells the model those files are the project, and for + * a fresh host that is nothing at all. Tagged so the activity boundary retries it elsewhere. + * + * `pauseBeforeLock` waits between reading the directory and taking the lock. Zero everywhere but + * the check that reproduces what a concurrent drain does in that gap: nothing outside this module + * can hold a caller there, and what the check asserts is the real outcome, whether a failed + * rebuild removes a directory this call did not create. */ - readonly ensure: (directory: string) => Effect.Effect + readonly ensure: ( + directory: string, + options?: { readonly pauseBeforeLock?: number }, + ) => Effect.Effect } export class Service extends Context.Service()( "@opencode/v2/WorktreeMaterializer", ) {} -// HEAD of a rebuilt tree, which doubles as the mark that says the tree is ours to move. +// HEAD of a rebuilt tree, so the rebuilt repo reads as a clean checkout rather than an unborn +// branch over a full untracked tree. const RESTORED = "refs/heads/opencode-restore" +/** A rebuild that did not finish. Tagged so the boundary can tell it from a refusal and retry it. */ +export class WorktreeMaterializeError extends Schema.TaggedErrorClass()( + "WorktreeMaterializer.MaterializeError", + { message: Schema.String }, +) {} + +// Nothing in it at all, so there is no work to protect and nothing to lose by checking a tree out +// over it. Unreadable counts as not empty: a directory we cannot look into is not one to overwrite. +const isEmptyDir = (dir: string) => + Effect.promise(() => + readdir(dir) + .then((entries) => entries.length === 0) + .catch(() => false), + ) + const layer = Layer.effect( Service, Effect.gen(function* () { @@ -61,13 +90,15 @@ const layer = Layer.effect( .create({ worktree, gitDirectory: AbsolutePath.make(path.join(worktree, ".git")) }) .pipe(Effect.orDie) // Index every pack shipped for this worktree; objects accumulate, the newest tree wins. - const rows = yield* db + const stored = yield* db .select() .from(SnapshotPackTable) .where(eq(SnapshotPackTable.worktree, tip.worktree)) - .orderBy(asc(SnapshotPackTable.time_created)) .all() .pipe(Effect.orDie) + // A pack cannot be indexed before the one it was built on, and the write clock does not order + // them: two hosts disagree about the time, and one behind puts its pack first. + const rows = orderChain(stored) const packDirectory = path.join(repository.gitDirectory, "objects", "pack") yield* fs.ensureDir(packDirectory).pipe(Effect.orDie) for (const row of rows) { @@ -120,68 +151,54 @@ const layer = Layer.effect( const behind = Effect.fnUntraced(function* (tip: typeof SnapshotPackTable.$inferSelect) { const held = yield* readWorktreeTip(global.data, tip.worktree) if (!held || held === tip.tree) return false - const shipped = yield* db - .select({ time: SnapshotPackTable.time_created }) + const rows = yield* db + .select() .from(SnapshotPackTable) - .where(and(eq(SnapshotPackTable.worktree, tip.worktree), eq(SnapshotPackTable.tree, held))) - .orderBy(desc(SnapshotPackTable.time_created)) - .limit(1) - .get() + .where(eq(SnapshotPackTable.worktree, tip.worktree)) + .all() .pipe(Effect.orDie) - return shipped !== undefined && shipped.time < tip.time_created + return isBehind(rows, held) }) - // Whether this tree is one we built from packs. A checkout the host already had is somebody's - // working copy: reading its captures is fine, but checking a stored tree out over it would - // rewrite files and HEAD under whoever owns it. - const rebuilt = (worktree: string) => - proc - .run( - ChildProcess.make( - "git", - [ - "--git-dir", - path.join(worktree, ".git"), - "rev-parse", - "--verify", - "--quiet", - RESTORED, - ], - { cwd: worktree, extendEnv: true }, - ), - ) - .pipe( - Effect.map((result) => result.exitCode === 0), - Effect.catchCause(() => Effect.succeed(false)), - ) - - const ensure = Effect.fn("WorktreeMaterializer.ensure")(function* (directory: string) { + const ensure = Effect.fn("WorktreeMaterializer.ensure")(function* ( + directory: string, + options?: { readonly pauseBeforeLock?: number }, + ) { // The newest capture whose session ran in this directory decides which worktree to rebuild, // and which state a tree that is already here has to be brought to. - const tip = yield* db - .select() - .from(SnapshotPackTable) - .where(eq(SnapshotPackTable.directory, directory)) - .orderBy(desc(SnapshotPackTable.time_created)) - .limit(1) - .get() - .pipe(Effect.orDie) + const tip = chainHead( + yield* db + .select() + .from(SnapshotPackTable) + .where(eq(SnapshotPackTable.directory, directory)) + .all() + .pipe(Effect.orDie), + ) if (!tip) return - const present = yield* fs.existsSafe(tip.worktree) - if (present) { - if (!(yield* behind(tip))) return - if (!(yield* rebuilt(tip.worktree))) { - yield* Effect.logWarning("worktree is behind the store and was not built from it", { - worktree: tip.worktree, - tree: tip.tree, - }) - return - } - } + // An empty directory is not somebody's working copy, so the rule that protects one does not + // apply to it. Treating it as present is what stops a fresh host from ever building the tree: + // it has no tip note, so `behind` says no, and the tools then run against nothing. A mounted + // path that exists but holds nothing is the ordinary shape of a host that has never seen this + // project, which is exactly the case the packs are for. + const present = (yield* fs.existsSafe(tip.worktree)) && !(yield* isEmptyDir(tip.worktree)) + // `behind` is already the whole rule. It is false unless this host has a note of its own, and + // a note means this host agreed to that state: either it built the tree from packs or it + // captured the tree from here. Moving it forward from a state it agreed to loses nothing. + // + // What used to gate this as well was whether the tree carried the marker `materialize` + // writes. Only a rebuilt tree ever has that, so a host that seeded the session from its own + // checkout never did, and once any other host shipped, every activity that host drew died + // here. A developer's checkout is protected by having no note at all, not by the marker. + if (present && !(yield* behind(tip))) return + if (options?.pauseBeforeLock) yield* Effect.sleep(options.pauseBeforeLock) yield* locks.withLock(tip.worktree)( Effect.gen(function* () { - // Re-check inside the lock: a concurrent drain may have done this already. - if ((yield* fs.existsSafe(tip.worktree)) && !(yield* behind(tip))) return + // Re-check inside the lock: a concurrent drain may have done this already. Same notion of + // present as above, or an empty directory bails out here instead and the tree that the + // outer check just decided to build never gets built. + const here = + (yield* fs.existsSafe(tip.worktree)) && !(yield* isEmptyDir(tip.worktree)) + if (here && !(yield* behind(tip))) return yield* materialize(tip).pipe( Effect.catchCauseIf( (cause) => !Cause.hasInterrupts(cause), @@ -189,13 +206,25 @@ const layer = Layer.effect( Effect.gen(function* () { // A half-built tree would pass the exists check forever, so what we created is // removed. A tree that was already here is not ours to remove: a failed refresh - // leaves it as stale as it was. - if (!present) + // leaves it as stale as it was. Asked of the reading taken inside the lock, which + // is the only one that describes the directory this attempt started from: the + // outer one is why the re-check exists, and a drain that materialized while this + // one waited makes it name a directory that no longer exists. + if (!here) yield* Effect.promise(() => rm(tip.worktree, { recursive: true, force: true })) - yield* Effect.logWarning("failed to materialize worktree", { + yield* Effect.logError("failed to materialize worktree", { worktree: tip.worktree, cause, }) + // Swallowing this ran the step against whatever was in the directory, which for + // a fresh host is nothing at all. Tagged, so the activity boundary can retry it: + // git and the filesystem fail for reasons that pass, and the alternative is a + // turn that fails for good because one worker had a bad minute. + return yield* Effect.die( + new WorktreeMaterializeError({ + message: `could not materialize ${tip.worktree} at ${tip.tree}`, + }), + ) }), ), ) diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index 2a63cf5ae9db..79cf3611e946 100644 --- a/packages/core/src/session/runner/index.ts +++ b/packages/core/src/session/runner/index.ts @@ -39,10 +39,21 @@ export interface StepResult { /** A tool call the provider asked for, recorded but not run, handed to the caller to dispatch. * Every id comes from the provider or the publisher and is carried, never regenerated: a second run * of the same step would mint different ones and the results would not match the log. */ +/** A call the model asked for, handed to whoever will run it. + * + * It names the call rather than carrying it. The arguments are already in the log when this is + * handed over: the streaming path ends the input fragment before it defers, so `Tool.Input.Ended` + * lands on the deferred path too and the dispatcher reads them off the pending call. Carrying them + * as well put them across a durable boundary twice, once as the model call's result and once as the + * tool call's input, so a step with large `write` bodies wrote them into history twice and a big + * enough one passed the payload limit. + * + * Taking them off failed once, and the reason is worth keeping: what the record holds is the raw + * JSON *string*, and handing that straight to a tool gave every one of them a string where its + * schema wanted an object. The dispatcher parses it. */ export interface DeferredToolCall { readonly id: string readonly name: string - readonly input: unknown readonly assistantMessageID: string } diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index df79bfc3a1cc..d41031ddc6ee 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -52,6 +52,7 @@ import { DEFAULT_MAX_STEPS, REPEAT_LIMIT, REPEATED_CALLS_PROMPT, trailingIdentic 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" /** @@ -108,6 +109,13 @@ import { llmClient } from "../../effect/app-node-platform" * 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* () { @@ -312,12 +320,7 @@ const layer = Layer.effect( // the // stream does and the overlap between the model and its tools is lost. if (deferTools) { - deferred.push({ - id: event.id, - name: event.name, - input: event.input, - assistantMessageID, - }) + deferred.push({ id: event.id, name: event.name, assistantMessageID }) return } yield* Effect.uninterruptibleMask((restore) => @@ -681,6 +684,18 @@ const layer = Layer.effect( } const moreQueue = yield* SessionInput.hasPending(db, sessionID, "queue") if (moreQueue) return { ran: true, continue: true, step: 1, promotion: "queue" as SessionInput.Delivery } + // The turn is over, and this is the only place that knows it: a step ending is not a turn + // ending, because a steer or a queued prompt continues the same turn through another step. + // Everything watching from outside had to infer it from a finish reason and a silence. Said + // once, here, for both modes, since both come through this function. + // + // Only the ordinary ending. A turn the user stopped, or one a provider error ended, does not + // reach here, so a follower still needs its other reasons to stop waiting. + yield* events.publish(SessionEvent.Turn.Ended, { + sessionID, + timestamp: yield* DateTime.now, + finish: "stop", + }) return { ran: true, continue: false, step: step + 1, promotion: undefined } }) @@ -802,6 +817,24 @@ const layer = Layer.effect( }) return { outcome: "unknown" } as 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. @@ -811,7 +844,7 @@ const layer = Layer.effect( assistantMessageID, callID: input.call.id, tool: input.call.name, - input: record(input.call.input), + input: record(args), // Deferred calls are never provider-executed: those are filtered out before the hand-off. provider: { executed: false }, }) @@ -823,7 +856,7 @@ const layer = Layer.effect( call: LLMEvent.toolCall({ id: input.call.id, name: input.call.name, - input: input.call.input, + input: args, }), }) .pipe( @@ -870,6 +903,16 @@ const layer = Layer.effect( // 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 }) diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index d634b86e8771..fd345cccf47b 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -114,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 @@ -225,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, @@ -354,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, @@ -370,6 +375,11 @@ 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}`) diff --git a/packages/core/src/snapshot-sync.ts b/packages/core/src/snapshot-sync.ts index ede416f8300a..c9bcdbd1f24f 100644 --- a/packages/core/src/snapshot-sync.ts +++ b/packages/core/src/snapshot-sync.ts @@ -21,7 +21,8 @@ import { AppProcess } from "./process" import { AbsolutePath } from "./schema" import type { Snapshot } from "./snapshot" import { SnapshotPackTable } from "./snapshot/sql" -import { writeWorktreeTip } from "./snapshot/tip" +import { chainHead } from "./snapshot/chain" +import { readWorktreeTip, writeWorktreeTip } from "./snapshot/tip" import { Hash } from "./util/hash" export interface Interface { @@ -62,23 +63,45 @@ const layer = Layer.effect( { stdin }, ) + // The newest state the store holds for this worktree, read off the chain the packs form rather + // than off `time_created`, which is whichever host wrote the row. + const newest = () => + db + .select() + .from(SnapshotPackTable) + .where(eq(SnapshotPackTable.worktree, worktree)) + .all() + .pipe(Effect.orDie, Effect.map(chainHead)) + const push = Effect.fn("SnapshotSync.push")(function* (tree: Snapshot.ID) { - // Noted before the packing, which is best-effort: what this host holds is true whether or not - // the pack reaches the store, and a note left behind would let a later drain check out an - // older tree over work only this host has. - if (source) yield* writeWorktreeTip(global.data, worktree, tree) + // Only a host standing on the store's newest state may add to it. One that never caught up + // packs its older files, becomes the newest by time, and every other host then checks that + // out over the work they were shipped to carry. + // + // Ahead of the note and outside the packing below, both deliberately. The note must not be + // moved for a ship that is not allowed, and the packing swallows its failures on purpose: a + // pack that does not reach the store costs the next host a rebuild from further back, where + // this is a host saying something untrue about the project. + if (source) { + const stoodOn = yield* readWorktreeTip(global.data, worktree) + const ahead = yield* newest() + if (ahead && ahead.tree !== tree && stoodOn !== ahead.tree) { + yield* Effect.die( + new Error( + `refusing to ship ${worktree}: this host stood on ${stoodOn ?? "nothing"}, ` + + `and the store is at ${ahead.tree}`, + ), + ) + } + } yield* Effect.gen(function* () { if (!source) return - const latest = yield* db - .select() - .from(SnapshotPackTable) - .where(eq(SnapshotPackTable.worktree, worktree)) - .orderBy(desc(SnapshotPackTable.time_created)) - .limit(1) - .get() - .pipe(Effect.orDie) - // The newest shipped state already is this tree: nothing to pack. - if (latest?.tree === tree) return + const latest = yield* newest() + // The newest shipped state already is this tree: nothing to pack, and the note is true. + if (latest?.tree === tree) { + yield* writeWorktreeTip(global.data, worktree, tree) + return + } // Chain onto the previous sync commit only when this host has it; a base absent locally // would produce a delta pack the pack builder cannot compute. const base = @@ -115,6 +138,12 @@ const layer = Layer.effect( .onConflictDoNothing() .run() .pipe(Effect.orDie) + // After the insert, never before it. The packing below swallows its failures, so a note + // written first and an insert that then failed named a tree the store never saw: `isBehind` + // finds no row for it and leaves the host where it is, while the ship guard compares that + // note with a head it can never match, so every later push from this host dies. Left at the + // last state the store agreed on, both keep working and the next push chains from there. + yield* writeWorktreeTip(global.data, worktree, tree) }).pipe( Effect.catchCauseIf( (cause) => !Cause.hasInterrupts(cause), diff --git a/packages/core/src/snapshot/chain.ts b/packages/core/src/snapshot/chain.ts new file mode 100644 index 000000000000..9173a7a8e0a7 --- /dev/null +++ b/packages/core/src/snapshot/chain.ts @@ -0,0 +1,81 @@ +// The order snapshot packs go in, decided by the packs themselves rather than by a clock. +// +// Each push chains onto the one before it, so `base` already records the order. `time_created` is +// whichever host wrote the row, and hosts do not agree on the time: a worker five minutes behind +// makes its older tree look like the newest one, and every other host then checks that out over +// the work they were shipped to carry. The chain has no such failure, because a host cannot invent +// a parent it has not seen. +// +// Forks should not happen: only a host standing on the newest state may add to it. They are still +// handled rather than assumed away, because a store written before that rule existed can hold one. +// Depth decides, and the write clock is only the tiebreak between two rows at the same depth. + +export interface ChainRow { + readonly id: string + readonly base: string | null + readonly time_created: number +} + +const depths = (rows: readonly T[]): Map => { + const byID = new Map(rows.map((row) => [row.id, row])) + const depth = new Map() + // Iterative, because the chain is one link per capture and nothing prunes it: a long session + // would put a stack frame per tool call that changed a file. + for (const start of rows) { + if (depth.has(start.id)) continue + const pending: T[] = [] + const seen = new Set() + let at: T | undefined = start + while (at && !depth.has(at.id) && !seen.has(at.id)) { + seen.add(at.id) + pending.push(at) + at = at.base ? (byID.get(at.base) as T | undefined) : undefined + } + // A root, a row whose base is not in the store, or a cycle: all start the count at zero. + let below = at && depth.has(at.id) ? depth.get(at.id)! : -1 + for (const row of pending.reverse()) depth.set(row.id, ++below) + } + return depth +} + +/** Packs in an order where a pack's base always comes before it, which is what indexing them needs. */ +export const orderChain = (rows: readonly T[]): T[] => { + const depth = depths(rows) + return [...rows].sort( + (a, b) => (depth.get(a.id) ?? 0) - (depth.get(b.id) ?? 0) || a.time_created - b.time_created, + ) +} + +/** The newest state the store holds, which is the deepest link in the chain. */ +export const chainHead = (rows: readonly T[]): T | undefined => { + const depth = depths(rows) + let head: T | undefined + for (const row of rows) { + if (!head) { + head = row + continue + } + const here = depth.get(row.id) ?? 0 + const best = depth.get(head.id) ?? 0 + if (here > best || (here === best && row.time_created > head.time_created)) head = row + } + return head +} + +/** + * Whether `tree` is an earlier state than the head, as opposed to one the store has never seen. + * A tree the store does not hold is this host's own uncaptured work, and moving off it would drop + * work nothing else has. + */ +export const isBehind = ( + rows: readonly T[], + tree: string, +): boolean => { + const head = chainHead(rows) + if (!head || head.tree === tree) return false + const depth = depths(rows) + const mine = rows.filter((row) => row.tree === tree) + if (mine.length === 0) return false + const deepest = Math.max(...mine.map((row) => depth.get(row.id) ?? 0)) + return deepest < (depth.get(head.id) ?? 0) +} 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 d45b2311faca..e788d2344bfc 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -775,6 +775,73 @@ 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 used to take the log back, which fenced out the + // tool activities of the step that was actually 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) + }), + ) + + // What this pins is the outcome, not the race: whoever the row names is the one that was told it + // won. It does NOT pin the compare-and-set that makes that true under a real interleaving. Two + // claims started together here run to completion one after the other, so this passes with the + // condition on the write removed. `event-claim.test.ts` forces that interleaving, with the seam + // at the database rather than in `claim`. Named for what it does. + it.effect("two claims for one log leave a single owner, and it is one that was told so", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = Session.ID.create() + yield* events.publish(DurableMessage, durableData(aggregateID, "seed")) + + const outcomes = yield* Effect.all( + ["run:11:1", "run:11:2"].map((token) => events.claim(aggregateID, token).pipe(Effect.exit)), + { concurrency: "unbounded" }, + ) + const won = outcomes.filter(Exit.isSuccess).length + + const { db } = yield* Database.Service + const row = yield* db + .select({ ownerID: EventSequenceTable.owner_id }) + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, aggregateID)) + .get() + + // Whoever the row names is the one that must have been told it won. Any other pairing means a + // claimer carried on believing it held a log it does not. + expect(won).toBeGreaterThanOrEqual(1) + expect(["run:11:1", "run:11:2"]).toContain(row?.ownerID ?? "") + if (won === 2) expect(row?.ownerID).toBe("run:11:2") + }), + ) + it.effect("claim fences replay owners", () => Effect.gen(function* () { const events = yield* EventV2.Service diff --git a/packages/core/test/session-runner-model-call.test.ts b/packages/core/test/session-runner-model-call.test.ts index b5527599eca1..2490d8599404 100644 --- a/packages/core/test/session-runner-model-call.test.ts +++ b/packages/core/test/session-runner-model-call.test.ts @@ -52,7 +52,7 @@ import { ReferenceGuidance } from "@opencode-ai/core/reference/guidance" import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat" import { Auth } from "@opencode-ai/llm/route" import { describe, expect } from "bun:test" -import { Cause, Effect, Exit, Layer, Schema, Stream } from "effect" +import { Cause, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect" import { testEffect } from "./lib/effect" const model = OpenAIChat.route @@ -124,6 +124,27 @@ const callsCrashingIdempotentTool: LLMClientShape["stream"] = () => 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" }), + ]) // A provider turn that publishes nothing at all: no text, no reasoning, no tool call. The publisher // mints the assistant message lazily on first content, so after this 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 @@ -221,9 +242,22 @@ const seedSession = Effect.gen(function* () { // 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 }) => +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({}), @@ -282,7 +316,7 @@ const registerProbes = (ran: { write: number; read: number }) => }) }) -const counters = () => ({ write: 0, read: 0 }) +const counters = () => ({ write: 0, read: 0 }) as { write: number; read: number; echoed?: string } const toolPart = (messages: ReadonlyArray, callID: string) => { for (const message of messages) { @@ -389,6 +423,26 @@ describe("SessionRunner model-only attempt", () => { expect(message?.type === "assistant" ? Boolean(message.time.completed) : false).toBe(true) }), ) + + // The turn saying it is over, as opposed to a step saying it is. Everything watching a session + // from outside used to infer the difference from a finish reason and then a silence, because a + // steer or a queued prompt continues the same turn through another step. + harness(textOnly).effect("says the turn ended, once, when nothing follows it", () => + Effect.gen(function* () { + yield* seedSession + const events = yield* EventV2.Service + const ended = yield* events + .subscribe(SessionEvent.Turn.Ended) + .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + const runner = yield* SessionRunner.Service + + yield* runner.runStep({ sessionID, step: 2, promotion: undefined, first: false, force: false }) + + const seen = yield* Fiber.join(ended) + expect(seen.length).toBe(1) + expect(seen[0]?.data.sessionID).toBe(sessionID) + }), + ) }) // Dispatching one recorded call on its own. The policy under test is what happens when a dispatch @@ -427,6 +481,42 @@ describe("SessionRunner tool dispatch", () => { }), ) + // What the hand-off no longer carries. The arguments come off the recorded call, 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. + harness(callsEchoWhole).effect("hands the tool the arguments the model asked with", () => + Effect.gen(function* () { + yield* seedSession + 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. + harness(callsEchoStreamed).effect("and the same when the provider streamed them", () => + Effect.gen(function* () { + yield* seedSession + 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") + }), + ) + harness(callsTool).effect("does nothing when the call already has a result", () => Effect.gen(function* () { yield* seedSession diff --git a/packages/core/test/snapshot-chain.test.ts b/packages/core/test/snapshot-chain.test.ts new file mode 100644 index 000000000000..37412fd0023c --- /dev/null +++ b/packages/core/test/snapshot-chain.test.ts @@ -0,0 +1,58 @@ +// The packs form a chain, and the chain is what orders them. `time_created` is whichever host +// wrote the row, and hosts do not agree on the time, so a worker whose clock is behind used to make +// its older tree the newest one that every other host then checked out. +import { describe, expect, test } from "bun:test" +import { chainHead, isBehind, orderChain } from "@opencode-ai/core/snapshot/chain" + +const row = (id: string, base: string | null, time: number, tree = `tree-${id}`) => ({ + id, + base, + time_created: time, + tree, +}) + +describe("snapshot chain", () => { + test("orders a chain by its links, not by the write clock", () => { + // Written by a host five minutes behind, so `b` claims an earlier time than its own parent. + const rows = [row("b", "a", 1_000), row("a", null, 300_000), row("c", "b", 2_000)] + expect(orderChain(rows).map((r) => r.id)).toEqual(["a", "b", "c"]) + }) + + test("the head is the deepest link, whatever the clock says", () => { + const rows = [row("a", null, 300_000), row("b", "a", 1_000), row("c", "b", 2_000)] + expect(chainHead(rows)?.id).toBe("c") + }) + + test("an empty store has no head", () => { + expect(chainHead([])).toBeUndefined() + }) + + test("a fork is decided by depth, and the clock only breaks a tie", () => { + // `x` and `y` both build on `a`. `y` is deeper, so it wins even though `x` was written later. + const rows = [row("a", null, 1), row("x", "a", 99_000), row("y", "a", 2), row("z", "y", 3)] + expect(chainHead(rows)?.id).toBe("z") + }) + + test("a row whose base is not in the store is treated as a root", () => { + const rows = [row("b", "missing", 5), row("c", "b", 6)] + expect(orderChain(rows).map((r) => r.id)).toEqual(["b", "c"]) + expect(chainHead(rows)?.id).toBe("c") + }) + + test("behind means earlier in the chain, not earlier on a clock", () => { + const rows = [row("a", null, 300_000), row("b", "a", 1_000)] + expect(isBehind(rows, "tree-a")).toBe(true) + expect(isBehind(rows, "tree-b")).toBe(false) + }) + + test("a tree the store has never seen is this host's own work, not a state behind", () => { + const rows = [row("a", null, 1), row("b", "a", 2)] + expect(isBehind(rows, "tree-never-shipped")).toBe(false) + }) + + test("a row that names itself as its base does not run the stack out", () => { + const rows = [row("a", "a", 1), row("b", "a", 2)] + expect(() => chainHead(rows)).not.toThrow() + expect(chainHead(rows)?.id).toBe("b") + }) +}) diff --git a/packages/core/test/worktree-materialize.test.ts b/packages/core/test/worktree-materialize.test.ts index beb0eba19186..c20318e1c345 100644 --- a/packages/core/test/worktree-materialize.test.ts +++ b/packages/core/test/worktree-materialize.test.ts @@ -5,10 +5,10 @@ import { describe, expect } from "bun:test" import { $ } from "bun" import { realpathSync } from "node:fs" -import { mkdir, readFile, rm, writeFile } from "node:fs/promises" +import { mkdir, readdir, readFile, rm, writeFile } from "node:fs/promises" import path from "path" import { asc } from "drizzle-orm" -import { Effect, Layer } from "effect" +import { Effect, Fiber, Layer } from "effect" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Database } from "@opencode-ai/core/database/database" @@ -18,6 +18,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema" import { Snapshot } from "@opencode-ai/core/snapshot" import { SnapshotSync } from "@opencode-ai/core/snapshot-sync" import { SnapshotPackTable } from "@opencode-ai/core/snapshot/sql" +import { writeWorktreeTip } from "@opencode-ai/core/snapshot/tip" import { WorktreeMaterializer } from "@opencode-ai/core/session/execution/worktree" import { testEffect } from "./lib/effect" import { tmpdir } from "./fixture/tmpdir" @@ -109,7 +110,130 @@ describe("WorktreeMaterializer", () => { }), ) - it.live("moves a rebuilt tree forward, and leaves a tree it did not build alone", () => + // A rebuild that fails removes what it created, and only that. The reading it asks is the one + // taken inside the lock: another drain can fill the directory while this one waits for it, and + // the reading from before the wait then names a directory that no longer exists. What that costs + // is not the rebuild, which retries, but the files git ignores in what it removed: an install, a + // build, a `.env`. `pauseBeforeLock` is the wait, and it is the only thing invented here. + it.live("keeps a directory another drain filled while a failed rebuild waited for the lock", () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + const root = realpathSync(tmp.path) + const worktree = path.join(root, "project") + const file = path.join(root, "shared.db") + const data = path.join(root, "host-b-data") + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await $`git init -q ${worktree}`.quiet() + await $`git -C ${worktree} config user.email t@t`.quiet() + await $`git -C ${worktree} config user.name t`.quiet() + await writeFile(path.join(worktree, "tracked.txt"), "v1\n") + await $`git -C ${worktree} add .`.quiet() + await $`git -C ${worktree} commit -qm seed`.quiet() + }) + + const A = yield* Layer.build(captureStack(file, worktree, path.join(root, "host-a-data"))) + const first = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + if (!first) throw new Error("expected a capture") + yield* SnapshotSync.Service.use((s) => s.push(first)).pipe(Effect.provide(A)) + const stored = yield* Database.Service.use(({ db }) => + db.select().from(SnapshotPackTable).all(), + ).pipe(Effect.orDie, Effect.provide(Database.layerFromPath(file)), Effect.scoped) + + // The newest state in the store, and a pack that is not a pack: indexing it is how a rebuild + // fails for reasons the store cannot rule out. + yield* Effect.sleep(10) + yield* Database.Service.use(({ db }) => + db + .insert(SnapshotPackTable) + .values([ + { + id: "f".repeat(40), + directory: worktree, + worktree, + tree: "e".repeat(40), + base: stored[0]!.id, + pack: Buffer.from([0x50, 0x41, 0x43, 0x4b]), + }, + ]) + .run(), + ).pipe(Effect.orDie, Effect.provide(Database.layerFromPath(file)), Effect.scoped) + + // This host is empty and behind, which is the state that decides to rebuild. + yield* Effect.promise(() => rm(worktree, { recursive: true, force: true })) + const B = yield* Layer.build(materializeStack(file, data)) + const rebuilding = yield* WorktreeMaterializer.Service.use((w) => + w.ensure(worktree, { pauseBeforeLock: 400 }), + ).pipe(Effect.provide(B), Effect.exit, Effect.forkChild) + + // What another drain leaves behind while this one waits: a checkout, the files git ignores, + // and the note saying this host agreed to that state. + yield* Effect.sleep(150) + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await writeFile(path.join(worktree, "tracked.txt"), "v1\n") + await writeFile(path.join(worktree, ".env"), "SECRET=1\n") + }) + yield* writeWorktreeTip(data, worktree, stored[0]!.tree) + + const outcome = yield* Fiber.join(rebuilding) + // The rebuild really did fail, which is the premise: a check where it succeeded would say + // nothing about what a failure removes. + expect(outcome._tag).toBe("Failure") + + // The rebuild failed on the bad pack. The packs would restore `tracked.txt` on a retry; the + // ignored file is in no pack and nothing else has a copy. + const left = yield* Effect.promise(() => readdir(worktree).catch(() => [] as string[])) + // A `.git` the failed rebuild made on its way is fine; what must survive is the other drain's + // work, and above all the file no pack carries. + expect(left).toContain("tracked.txt") + expect(left).toContain(".env") + + yield* Effect.promise(() => tmp[Symbol.asyncDispose]()) + }), + ) + + it.live("rebuilds into a directory that exists but is empty", () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + const root = realpathSync(tmp.path) + const worktree = path.join(root, "project") + const file = path.join(root, "shared.db") + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await $`git init -q ${worktree}`.quiet() + await $`git -C ${worktree} config user.email t@t`.quiet() + await $`git -C ${worktree} config user.name t`.quiet() + await writeFile(path.join(worktree, "note.txt"), "travelled\n") + await $`git -C ${worktree} add .`.quiet() + await $`git -C ${worktree} commit -qm seed`.quiet() + }) + + const A = yield* Layer.build(captureStack(file, worktree, path.join(root, "host-a-data"))) + const captured = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + if (!captured) throw new Error("expected a capture") + yield* SnapshotSync.Service.use((s) => s.push(captured)).pipe(Effect.provide(A)) + + // The shape a container gives a fresh host: the path is there because something mounted it, + // and there is nothing in it. Deleting the directory instead is the case already covered, + // and it is the easy one: an absent tree is obviously safe to build. + yield* Effect.promise(async () => { + await rm(worktree, { recursive: true, force: true }) + await mkdir(worktree, { recursive: true }) + }) + + const B = yield* Layer.build(materializeStack(file, path.join(root, "host-b-data"))) + yield* WorktreeMaterializer.Service.use((w) => w.ensure(worktree)).pipe(Effect.provide(B)) + + expect(yield* Effect.promise(() => readFile(path.join(worktree, "note.txt"), "utf8"))).toBe( + "travelled\n", + ) + + yield* Effect.promise(() => tmp[Symbol.asyncDispose]()) + }), + ) + + it.live("moves a tree that is behind forward, and leaves one already at the tip alone", () => Effect.gen(function* () { const tmp = yield* Effect.promise(() => tmpdir()) const root = realpathSync(tmp.path) @@ -176,6 +300,135 @@ describe("WorktreeMaterializer", () => { }), ) + // The write direction. A host the store has moved past used to pack its older files, become the + // newest by time, and every other host then checked that out over the work they were shipped to + // carry. This is the same rule the read direction already had, in the direction nothing checked. + it.live("refuses to ship from a host the store has moved past", () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + const root = realpathSync(tmp.path) + const worktree = path.join(root, "project") + const file = path.join(root, "shared.db") + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await $`git init -q ${worktree}`.quiet() + await $`git -C ${worktree} config user.email t@t`.quiet() + await $`git -C ${worktree} config user.name t`.quiet() + await writeFile(path.join(worktree, "f.txt"), "v1\n") + await $`git -C ${worktree} add .`.quiet() + await $`git -C ${worktree} commit -qm seed`.quiet() + }) + + const A = yield* Layer.build(captureStack(file, worktree, path.join(root, "a-data"))) + const first = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + yield* SnapshotSync.Service.use((s) => s.push(first!)).pipe(Effect.provide(A)) + + // Another host ships while this one is not looking. Written straight into the store, because + // two capture stacks for one worktree resolve to the same host: the node builder keys them by + // location, so the second host has to be the row rather than a second stack. + const elsewhere = "e".repeat(40) + yield* Effect.sleep(10) + yield* Database.Service.use(({ db }) => + db + .insert(SnapshotPackTable) + .values([ + { + id: "d".repeat(40), + directory: worktree, + worktree, + tree: elsewhere, + pack: Buffer.from([0x50, 0x41, 0x43, 0x4b]), + }, + ]) + .run(), + ).pipe(Effect.orDie, Effect.provide(Database.layerFromPath(file)), Effect.scoped) + + // This host is still standing on `first`, so what it holds is not built on what the store now + // says the project is. Shipping it would revert the other host. + yield* Effect.promise(() => writeFile(path.join(worktree, "f.txt"), "stale\n")) + const stale = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + const exit = yield* SnapshotSync.Service.use((s) => s.push(stale!)).pipe( + Effect.provide(A), + Effect.exit, + ) + expect(exit._tag).toBe("Failure") + + // Nothing was added, and the note was not moved either: a refused ship must leave this host + // saying what it actually holds. + const rows = yield* Database.Service.use(({ db }) => + db.select().from(SnapshotPackTable).orderBy(asc(SnapshotPackTable.time_created)).all(), + ).pipe(Effect.orDie, Effect.provide(Database.layerFromPath(file)), Effect.scoped) + expect(rows).toHaveLength(2) + expect(rows[1]?.tree).toBe(elsewhere) + + yield* Effect.promise(() => tmp[Symbol.asyncDispose]()) + }), + ) + + // A host that seeded the session from its own checkout has a note but no rebuild marker, because + // only a rebuild writes one. Gating the move on that marker meant every activity such a host drew + // died as soon as any other host shipped, and the comment said it would be sent elsewhere when the + // boundary marks it non-retryable. The note is the rule: a host that agreed to a state may be + // moved off it. + it.live("moves a tree the host captured rather than rebuilt", () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + const root = realpathSync(tmp.path) + const worktree = path.join(root, "project") + const file = path.join(root, "shared.db") + const data = path.join(root, "seed-host-data") + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await $`git init -q ${worktree}`.quiet() + await $`git -C ${worktree} config user.email t@t`.quiet() + await $`git -C ${worktree} config user.name t`.quiet() + await writeFile(path.join(worktree, "f.txt"), "seeded\n") + await $`git -C ${worktree} add .`.quiet() + await $`git -C ${worktree} commit -qm seed`.quiet() + }) + + // This host captures from its own checkout, so it gets a note and no rebuild marker. + const A = yield* Layer.build(captureStack(file, worktree, data)) + const first = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + yield* SnapshotSync.Service.use((s) => s.push(first!)).pipe(Effect.provide(A)) + const packs = yield* Database.Service.use(({ db }) => + db.select().from(SnapshotPackTable).all(), + ).pipe(Effect.orDie, Effect.provide(Database.layerFromPath(file)), Effect.scoped) + + // Another host ships on top, so this one is behind. + yield* Effect.sleep(10) + yield* Database.Service.use(({ db }) => + db + .insert(SnapshotPackTable) + .values([ + { + id: "d".repeat(40), + directory: worktree, + worktree, + tree: "e".repeat(40), + base: packs[0]!.id, + pack: Buffer.from([0x50, 0x41, 0x43, 0x4b]), + }, + ]) + .run(), + ).pipe(Effect.orDie, Effect.provide(Database.layerFromPath(file)), Effect.scoped) + + const B = yield* Layer.build(materializeStack(file, data)) + const exit = yield* WorktreeMaterializer.Service.use((w) => w.ensure(worktree)).pipe( + Effect.provide(B), + Effect.exit, + ) + + // The pack above is not a real one, so the rebuild itself cannot succeed here. What this pins + // is which failure: a rebuild that was attempted and failed, not a refusal to try. + const why = String(exit) + expect(why).not.toContain("was not built") + expect(why).toContain("could not materialize") + + yield* Effect.promise(() => tmp[Symbol.asyncDispose]()) + }), + ) + // The shared-store deployment uses the libsql backend, so the pack blob has to survive that // driver's parameter path too, not only bun's. it.live("round-trips a pack blob through the libsql backend", () => diff --git a/packages/opencode/src/cli/cmd/detached.ts b/packages/opencode/src/cli/cmd/detached.ts new file mode 100644 index 000000000000..b6888e4a9075 --- /dev/null +++ b/packages/opencode/src/cli/cmd/detached.ts @@ -0,0 +1,397 @@ +// Commands for a session nobody is sitting in front of: start one and walk away, ask the +// deployment what it is still running, and follow one from a machine that never had it. +// +// These are thin HTTP clients on purpose. In a durable deployment the serve processes are +// interchangeable (any of them reads the shared store and signals the same workflows), so a client +// needs an endpoint and a session id, never a particular host. That is the whole reason a session +// can outlive the process that started it, and it is why nothing here talks to Temporal. The one +// exception is `doctor`, which reads the driver's own configuration module: it answers what this +// deployment resolved, and a second copy of those rules living here is how the two would disagree. + +import type { Argv } from "yargs" +import { cmd } from "./cmd" +import { UI } from "../ui" +import { ServerAuth } from "@/server/auth" +import { TemporalConfig } from "@opencode-ai/temporal/config" +// Type-only: the SDK types a duration as a template literal, and this takes one from a person. +import type { Duration } from "@temporalio/common" + +const DEFAULT_URL = "http://127.0.0.1:4096" + +type Remote = { readonly url: string; readonly headers: Record } + +function remote(args: { attach?: string; password?: string; username?: string }): Remote { + const url = (args.attach ?? process.env["OPENCODE_SERVER"] ?? DEFAULT_URL).replace(/\/+$/, "") + // No password configured is a valid deployment, so absent auth is absent headers, not an error. + return { url, headers: ServerAuth.headers({ password: args.password, username: args.username }) ?? {} } +} + +async function call(r: Remote, path: string, init?: RequestInit): Promise { + const response = await fetch(`${r.url}/api${path}`, { + ...init, + headers: { ...r.headers, ...(init?.body ? { "content-type": "application/json" } : {}), ...init?.headers }, + }) + if (!response.ok) { + const detail = await response.text().catch(() => "") + throw new Error(`${init?.method ?? "GET"} /api${path} failed: ${response.status} ${detail.slice(0, 200)}`) + } + if (response.status === 204) return undefined as T + const body = (await response.json()) as { data: T } + return body.data +} + +// The remote-facing options every command here shares. Kept in one builder so a second endpoint +// flag can never drift between them. +function remoteOptions(yargs: Argv) { + return yargs + .option("attach", { + type: "string", + describe: `server to talk to (default ${DEFAULT_URL}, or $OPENCODE_SERVER)`, + }) + .option("password", { alias: "p", type: "string", describe: "basic auth password" }) + .option("username", { alias: "u", type: "string", describe: "basic auth username" }) + .option("json", { type: "boolean", describe: "print machine-readable output", default: false }) +} + +interface SessionInfo { + id: string + title?: string + time?: { created?: number; updated?: number } + location?: { directory?: string } +} + +const stamp = (ms?: number) => (ms ? new Date(ms).toISOString().replace("T", " ").slice(0, 19) : "") + +// UI.println writes to stderr, which is right for a person and wrong for a pipe. Anything a script +// is meant to read goes to stdout instead. +const emit = (line: string) => process.stdout.write(line + "\n") + +export const SessionStartCommand = cmd({ + command: "start ", + describe: "start a session, hand it a prompt, and return without waiting for it", + builder: (yargs: Argv) => + remoteOptions(yargs) + .positional("prompt", { type: "string", describe: "what the agent should do", demandOption: true }) + .option("dir", { type: "string", describe: "session working directory (default: this one)" }) + .option("model", { type: "string", describe: "provider/model, e.g. openai/gpt-5-mini" }), + handler: async (args) => { + const r = remote(args) + try { + const directory = args.dir ?? process.cwd() + const session = await call(r, "/session", { + method: "POST", + body: JSON.stringify({ directory }), + }) + if (args.model) { + const slash = args.model.indexOf("/") + if (slash < 1) throw new Error(`--model wants provider/model, got ${args.model}`) + const model = { providerID: args.model.slice(0, slash), id: args.model.slice(slash + 1) } + await call(r, `/session/${session.id}/model`, { method: "POST", body: JSON.stringify({ model }) }) + } + // The prompt is admitted, not awaited. Whoever is polling the task queue runs the turn, and + // this process has nothing left to do with it. + await call(r, `/session/${session.id}/prompt`, { + method: "POST", + body: JSON.stringify({ prompt: { text: args.prompt } }), + }) + if (args.json) { + emit(JSON.stringify({ id: session.id, directory, url: r.url })) + return + } + emit(session.id) + UI.println(` follow it with: opencode session watch ${session.id}`) + } catch (error) { + UI.error(error instanceof Error ? error.message : String(error)) + process.exitCode = 1 + } + }, +}) + +// What this process resolved, and what is wrong with it. Deploying was a handful of variables that +// have to agree, with no way to ask whether they did: every mistake in them fails as something +// else, hours later, on whoever prompted the session rather than on whoever deployed it. +export const SessionDoctorCommand = cmd({ + command: "doctor", + describe: "what this deployment resolved, and what is wrong with it", + builder: (yargs: Argv) => yargs, + handler: async () => { + const config = TemporalConfig.fromEnv() + UI.println("opencode, temporal execution") + for (const [name, value] of Object.entries(TemporalConfig.describe(config))) { + UI.println(` ${name}: ${value}`) + } + for (const note of TemporalConfig.notes(config)) UI.println(`note: ${note}`) + const problems = TemporalConfig.preflight(config) + for (const problem of problems) UI.println(`problem: ${problem}`) + if (problems.length > 0) { + process.exitCode = 1 + return + } + UI.println("this deployment looks consistent") + }, +}) + +// A turn nobody starts. `start` still needs something running to hand the prompt to; a schedule +// does not, which is the difference between a session you can walk away from and one that runs +// without you. The session is created once, here, over HTTP like everything else in this file; the +// firing itself reaches only Temporal, and a deployment with no serve process at all still runs it. +export const SessionScheduleCommand = cmd({ + command: "schedule ", + describe: "run a prompt on a schedule, with no client at firing time", + builder: (yargs: Argv) => + remoteOptions(yargs) + .positional("prompt", { type: "string", describe: "what the agent should do", demandOption: true }) + .option("every", { type: "string", describe: "interval, e.g. 1h" }) + .option("cron", { type: "string", describe: "cron expression, e.g. '0 9 * * *'" }) + .option("id", { type: "string", describe: "schedule id (default: generated)" }) + .option("session", { type: "string", describe: "an existing session to prompt (default: a new one)" }) + .option("dir", { type: "string", describe: "session working directory (default: this one)" }), + handler: async (args) => { + const r = remote(args) + try { + if (!args.every && !args.cron) throw new Error("schedule wants --every= or --cron=") + const sessionID = + args.session ?? + ( + await call(r, "/session", { + method: "POST", + body: JSON.stringify({ directory: args.dir ?? process.cwd() }), + }) + ).id + const config = TemporalConfig.fromEnv() + const { Client, Connection, ScheduleOverlapPolicy } = await import("@temporalio/client") + const connection = await Connection.connect(TemporalConfig.connectionOptions(config)) + try { + const client = new Client({ connection, namespace: config.namespace }) + const scheduleId = args.id ?? `opencode-${sessionID}` + await client.schedule.create({ + scheduleId, + spec: { + ...(args.cron ? { cronExpressions: [args.cron] } : {}), + ...(args.every ? { intervals: [{ every: args.every as Duration }] } : {}), + }, + // A firing that lands while the last one is still working is skipped rather than queued. + // An agent task is not a metrics scrape: two of them on one project is a bad day. + policies: { overlap: ScheduleOverlapPolicy.SKIP }, + action: { + type: "startWorkflow", + workflowType: "scheduledPrompt", + taskQueue: config.taskQueue, + args: [ + { + sessionID, + text: args.prompt, + session: { idleTimeout: config.idleTimeout, stepped: config.stepped === true }, + }, + ], + }, + }) + if (args.json) { + emit(JSON.stringify({ schedule: scheduleId, session: sessionID })) + return + } + emit(scheduleId) + UI.println(` every firing prompts ${sessionID}; follow it with: opencode session watch ${sessionID}`) + } finally { + await connection.close() + } + } catch (error) { + UI.error(error instanceof Error ? error.message : String(error)) + process.exitCode = 1 + } + }, +}) + +export const SessionRunningCommand = cmd({ + command: "running", + describe: "list the sessions this deployment is executing right now", + builder: (yargs: Argv) => remoteOptions(yargs), + handler: async (args) => { + const r = remote(args) + try { + // Which sessions are running is the executor's answer, not a guess from the transcript: a + // durable deployment reads it from the running workflows, so it survives a restart of + // whichever process happens to be answering this call. + const active = await call>(r, "/session/active") + const ids = Object.keys(active) + if (args.json) { + emit(JSON.stringify(ids.map((id) => ({ id, status: active[id]?.type })))) + return + } + if (ids.length === 0) { + UI.println("nothing running") + return + } + const all = await call(r, "/session").catch(() => [] as SessionInfo[]) + const byId = new Map(all.map((s) => [s.id, s])) + for (const id of ids) { + const session = byId.get(id) + const cells = [id, active[id]?.type ?? "?", stamp(session?.time?.updated), session?.title ?? ""] + emit(cells.join(" ")) + } + } catch (error) { + UI.error(error instanceof Error ? error.message : String(error)) + process.exitCode = 1 + } + }, +}) + +// What a follower prints. The stream carries far more than a person watching wants to read, so this +// keeps the events that say the work moved and drops the token-level ones. +const INTERESTING: Record string | undefined> = { + "session.next.prompted": () => "prompted", + "session.next.step.started": () => "step", + "session.next.tool.called": (d) => `tool ${d.tool}: ${JSON.stringify(d.input ?? {}).slice(0, 120)}`, + "session.next.tool.success": (d) => `tool ok ${firstText(d.content).slice(0, 200)}`, + "session.next.tool.failed": (d) => `tool failed ${firstText(d.content).slice(0, 200)}`, + "session.next.text.ended": (d) => (d.text ? `said: ${String(d.text).slice(0, 400)}` : undefined), + "session.next.step.failed": (d) => `step failed: ${d.error?.message ?? ""}`, +} + +function firstText(content: unknown): string { + if (!Array.isArray(content)) return "" + const part = content.find((c) => c && typeof c === "object" && (c as any).type === "text") as any + return part?.text ? String(part.text).trim() : "" +} + +export const SessionWatchCommand = cmd({ + command: "watch ", + describe: "follow a running session from anywhere, and exit when it goes idle", + builder: (yargs: Argv) => + remoteOptions(yargs) + .positional("sessionID", { type: "string", describe: "session to follow", demandOption: true }) + .option("wait", { type: "boolean", default: true, describe: "keep following until the session is idle" }), + handler: async (args) => { + const r = remote(args) + const sessionID = args.sessionID + + // The turn saying so itself, which is the only thing that knows: a step ending is not a turn + // ending, because a steer or a queued prompt continues the same turn. It covers the ordinary + // ending only, so the reasons below are still what a stopped or failed turn ends this on. + const turnEnded = (event: { type?: string }) => event.type === "session.next.turn.ended" + + // A step ending is where a turn usually ends, from the model's own finish reason: `tool-calls` + // is the one that means another step follows. It is not on its own proof the turn is over, + // because a steer or a queued prompt continues it, so the executor's own answer decides. + const looksDone = (event: { type?: string; data?: any }) => + event.type === "session.next.step.failed" || + (event.type === "session.next.step.ended" && event.data?.finish !== "tool-calls") + + // What says the turn did NOT end after all: a steer or a queued prompt continues the same turn + // through `stepContinuation`, and the next thing on the wire is another step starting. + const carriesOn = (event: { type?: string }) => + event.type === "session.next.step.started" || event.type === "session.next.prompted" + + // The running set cannot end a `watch` on its own. It holds a session for the whole idle + // timeout after the work is done, so gating the exit on it means never exiting. Nothing + // publishes a turn-level ending either, so what usually ends this is the wire going quiet: a + // terminal step, then no continuation within a grace window. A steer arrives in milliseconds, + // so the window only has to outlast the hop between two activities. + const GRACE_MS = Number(process.env.OPENCODE_WATCH_GRACE_MS ?? 5_000) + // The wire cannot answer the other case. A turn that ends while this is reconnecting publishes + // its last step into a gap, and the stream has no replay, so nothing arrives afterwards and the + // quiet means nothing. Absence from the running set is slow but certain, and it is the one + // reading that only ever says "finished", so it can end a watch without being able to hang one. + const POLL_MS = Number(process.env.OPENCODE_WATCH_POLL_MS ?? 30_000) + const inactive = async () => { + const active = await call>(r, "/session/active").catch(() => undefined) + // Unreachable is not finished, which is the whole point of this command. + return active !== undefined && !(sessionID in active) + } + + // Kept across reconnects. The terminal step lands on one connection and the quiet that follows + // it on the next, and starting this again per connection is what followed a finished turn for + // as long as the terminal stayed open. + let settleAt: number | undefined + + // The stream ends when the serve this is attached to restarts, which is the event this command + // exists for. Exiting 0 there reports a turn that is still running as done. + const attempts = 30 + try { + for (let attempt = 0; ; attempt++) { + const response = await fetch(`${r.url}/api/session/${sessionID}/event`, { + headers: r.headers, + }).catch(() => undefined) + if (!response?.ok || !response.body) { + if (attempt >= attempts) + throw new Error(`cannot follow ${sessionID}: ${response?.status ?? "unreachable"}`) + await new Promise((resolve) => setTimeout(resolve, 1000)) + continue + } + + const reader = response.body.getReader() + const decoder = new TextDecoder() + let buffer = "" + let ended = false + // Held across iterations rather than started fresh each time: a read that loses the race is + // still queued on the stream, and dropping it drops whatever it goes on to deliver. + let pending: ReturnType | undefined + for (;;) { + const next = (pending ??= reader.read()) + let timer: ReturnType | undefined + const chunk = args.wait + ? await Promise.race([ + next, + new Promise<"quiet">((resolve) => { + const wait = settleAt ? Math.max(0, settleAt - Date.now()) : POLL_MS + timer = setTimeout(() => resolve("quiet"), wait) + }), + ]) + : await next + if (timer) clearTimeout(timer) + if (chunk === "quiet") { + // A terminal step and then nothing: the turn is over. Otherwise this is the periodic + // ask, and only the running set can end the wait. + if ((settleAt && Date.now() >= settleAt) || (await inactive())) { + ended = true + break + } + continue + } + pending = undefined + const { done, value } = chunk + if (done) break + buffer += decoder.decode(value, { stream: true }) + const lines = buffer.split("\n") + buffer = lines.pop() ?? "" + for (const raw of lines) { + const line = raw.startsWith("data:") ? raw.slice(5).trim() : raw.trim() + if (!line.startsWith("{")) continue + let event: { type?: string; data?: any } + try { + event = JSON.parse(line) + } catch { + continue + } + if (args.json) { + emit(line) + } else { + const render = event.type ? INTERESTING[event.type] : undefined + const text = render?.(event.data ?? {}) + if (text) UI.println(`${stamp(event.data?.timestamp)} ${text}`) + } + if (args.wait) { + // The turn saying it is over ends this now: there is nothing to wait out, and the + // grace window exists only because nothing used to say it. + if (turnEnded(event)) { + ended = true + break + } + if (carriesOn(event)) settleAt = undefined + else if (looksDone(event)) settleAt = Date.now() + GRACE_MS + } + } + } + await reader.cancel().catch(() => {}) + // Without `--wait` the stream itself is the whole command, so its end is this one's too. + if (ended || !args.wait) return + // The stream dropped with the turn still going. Reconnect and keep following. + if (attempt >= attempts) throw new Error(`lost the stream for ${sessionID}`) + await new Promise((resolve) => setTimeout(resolve, 1000)) + } + } catch (error) { + UI.error(error instanceof Error ? error.message : String(error)) + process.exitCode = 1 + } + }, +}) diff --git a/packages/opencode/src/cli/cmd/session.ts b/packages/opencode/src/cli/cmd/session.ts index 9e6ddda9d2d8..94b746b63b72 100644 --- a/packages/opencode/src/cli/cmd/session.ts +++ b/packages/opencode/src/cli/cmd/session.ts @@ -1,6 +1,13 @@ import type { Argv } from "yargs" import { Effect } from "effect" import { cmd } from "./cmd" +import { + SessionDoctorCommand, + SessionRunningCommand, + SessionScheduleCommand, + SessionStartCommand, + SessionWatchCommand, +} from "./detached" import { effectCmd, fail } from "../effect-cmd" import { Session } from "@/session/session" import { SessionID } from "../../session/schema" @@ -44,7 +51,16 @@ function pagerCmd(): string[] { export const SessionCommand = cmd({ command: "session", describe: "manage sessions", - builder: (yargs: Argv) => yargs.command(SessionListCommand).command(SessionDeleteCommand).demandCommand(), + builder: (yargs: Argv) => + yargs + .command(SessionListCommand) + .command(SessionDeleteCommand) + .command(SessionStartCommand) + .command(SessionRunningCommand) + .command(SessionWatchCommand) + .command(SessionScheduleCommand) + .command(SessionDoctorCommand) + .demandCommand(), async handler() {}, }) diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 3a559c3e38a4..4b72989e5608 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -145,6 +145,24 @@ export namespace Shell { export type Ended = typeof Ended.Type } +// The turn, as opposed to the steps it was made of. A step ending is not a turn ending: a steer or +// a queued prompt continues the same turn through another step, and everything watching a session +// from outside had to guess at the difference from a finish reason plus a silence. Live-only, and +// deliberately: it says nothing the durable events do not already say, and the record of what a +// turn did is those events. What it adds is a boundary, published by the one thing that knows it. +export namespace Turn { + export const Ended = Event.define({ + type: "session.next.turn.ended", + schema: { + ...Base, + // What the last step of the turn came to, so a follower can say why it stopped rather than + // only that it did. + finish: Schema.String, + }, + }) + export type Ended = typeof Ended.Type +} + export namespace Step { export const Started = Event.define({ type: "session.next.step.started", @@ -458,6 +476,7 @@ export const DurableDefinitions = Event.inventory( Step.Started, Step.Ended, Step.Failed, + Turn.Ended, Text.Started, Text.Ended, Tool.Input.Started, diff --git a/packages/temporal/README.md b/packages/temporal/README.md index d2d8ac5a6f95..498c128df62f 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -340,6 +340,12 @@ what makes any worker able to serve any session, and turning affinity on is choo Two consequences to plan for, both silent: +- **The key is the directory, not the host.** In a container fleet where every worker's project is + the same path, this affinity is a no-op: they all poll the same queue and a step's tools still + land wherever. It is a real routing decision only where hosts serve genuinely different paths. + What keeps a step's writes together in a container fleet is step affinity below, which is keyed by + host as well as by path. + - **A worker serves one tree.** In the default `role=both` deployment the embedded worker polls the queue for the process directory, so a session in another project has no poller. Point `OPENCODE_TEMPORAL_WORKTREE` at the project root, not at a subfolder, since the key is the project @@ -451,6 +457,58 @@ later. Verified by `packages/core/test/session-runner-resume.test.ts`. Resume is verified end to end: it resolves on a healthy session and rejects on a failing one with the original tagged error (`LLM.Error`) reconstructed across the boundary. +### A turn nobody starts + +`session start` still needs something running to hand the prompt to. A schedule does not: it is a +Temporal object, and what it fires is a workflow that admits the prompt itself and then starts the +session's own supervisor. At firing time there is no client and no serve process, only workers. + +The session is created once, when the schedule is made, because a session is a row in the store +before it is anything else. After that the firing reaches only Temporal. The prompt is admitted as +queued rather than delivered, so a firing that lands while the last turn is still working is not +lost: it is drained when that turn ends, and overlapping firings are skipped rather than stacked. + +### Picking a deployment rather than assembling one + +The settings below are not independent, and getting them wrong fails as something else later: a +store only one process can see reads as a worker that never picks anything up. `OPENCODE_TEMPORAL_PROFILE` +picks one deployment and the rest follow. + +| | `local` (default) | `fleet` | +|---|---|---| +| what it is | one serve, worker inside it | serve processes and workers, separate | +| store | this process only | **you set** `OPENCODE_DB_URL` | +| role | `both` | `client` for serve, `worker` for workers | +| unit of work | a whole step | the model call, each tool call, the seal | + +Anything can still be set on its own; the profile decides only what it is when you do not. A fleet +cannot be talked out of the two that make it one, and a process that fails preflight refuses to +build rather than accepting work it cannot do. + +Reaching a server that is not the dev server: + +```bash +TEMPORAL_ADDRESS=your-ns.a1b2c.tmprl.cloud:7233 TEMPORAL_NAMESPACE=your-ns.a1b2c \ + OPENCODE_TEMPORAL_API_KEY_FILE=/run/secrets/temporal-key # Temporal Cloud +TEMPORAL_ADDRESS=temporal.internal:7233 \ + OPENCODE_TEMPORAL_TLS_CERT=/run/secrets/tls.crt \ + OPENCODE_TEMPORAL_TLS_KEY=/run/secrets/tls.key # a cluster with mTLS +``` + +The key comes from a file rather than from argv, and nothing prints it. Both halves build the +connection from one function, so a client and a worker cannot disagree about how the cluster is +reached. + +Ask before deploying rather than after: + +```bash +opencode session doctor +``` + +It prints what this process resolved and names what is wrong: an API key against a dev server, a +Cloud key with the `default` namespace, an address that is not loopback with no credentials, half a +certificate pair, a fleet with a store nobody else can read. + ### Running workers separately By default the serve process hosts both the Temporal activity worker and the workflow client @@ -569,13 +627,43 @@ and dependencies are not captured, so a rebuilt tree may need an install step be identically. Worker affinity (below) or a shared volume skips the materialization latency on warm paths; the packs are the portable baseline that works with neither. -Two rules bound what that refresh may touch, because checking a stored tree out over the wrong one -destroys work. A tree is moved only when a host-local note (`snapshot/tip.ts`) says this host is -behind the store, so a host holding a capture that never shipped is left as it is. And it is moved -only when this host built the tree from packs, so a checkout the host already had, a developer's own -working copy, is never rewritten: that case is logged and left alone. What stays open is the tools -of ONE step running on two hosts, since nothing captures their writes until the step is sealed. -Affinity is what keeps a step's tools on one tree. +The rules that bound it, because checking a stored tree out over the wrong one destroys work: + +- **Newest is decided by the chain, not by a clock.** Each pack names the one it was built on, and + that order no host can get wrong. `time_created` is whichever host wrote the row, so a worker + five minutes behind used to make its older tree the newest one that everybody else checked out. +- **Only a host standing on the newest state may add to it.** A host that never caught up used to + pack its older files, become the newest by time, and revert everyone. It refuses now, ahead of + its own tip note and outside the packing (which swallows its failures on purpose, so a guard + inside it would only have logged). +- **A tree is moved only when this host has a note for it**, which means this host agreed to that + state: it either built the tree from packs or captured the tree from there. A developer's own + checkout has no note, so it is never rewritten. Gating this on whether the tree carried the marker + a rebuild writes was wrong in the other direction: a host that seeded the session from its own + checkout never has that marker, so once anyone else shipped, every activity that host drew failed + for good. +- **The tip note is written after the insert, never before.** The packing swallows its own failures, + so a note written first and an insert that then failed named a tree the store never saw: the host + was behind nothing it could see, and every later ship from it was refused. +- **A tool ships from the host that ran it.** The seal can land anywhere, and it used to be the only + thing that captured, so a tool's writes reached the store only when the seal happened to be on + the same host. +- **A step stays on the worker that ran its model call** (`OPENCODE_TEMPORAL_STEP_AFFINITY`, on by + default). Every worker polls a second queue of its own, keyed by host and directory, and the model + call reports it; the tools and the seal are addressed there. That worker is standing in the tree + the tools are about to write, so they see each other through the filesystem instead of shipping + the tree to each other, which is what lets them run at once again. What keeps this from being a + worse kind of stuck than the shared queue: the pinned dispatch carries a 30 second + `scheduleToStartTimeout`, and that failure means the activity never started, so the work moves to + the shared queue with nothing run twice. Whatever is left of that step then goes one at a time, + because on the shared queue it can land on two hosts again. +- **A step's tools otherwise run one at a time wherever the store is shared** + (`OPENCODE_TEMPORAL_SERIAL_TOOLS=1`, and the default only when step affinity is off). Two on two + hosts each publish a tree without the other's work, and the second is refused rather than + reverting the first, which leaves its work stranded there. +- **Capturing and shipping the tree is one at a time per directory.** Two tools of one step now run + at once on one host, 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. Host-local state that does NOT ride the DB, so it is not reconstructed on a different host: @@ -588,6 +676,92 @@ Host-local state that does NOT ride the DB, so it is not reconstructed on a diff `${data}` (the XDG data dir) at shared storage to make them portable. +## A session that outlives its client + +Everything above makes a session survive a worker. Together the same pieces make it survive the +*client*, which is the part a user can feel: start something, close the laptop, and pick it up from +a machine that has never seen it. + +Nothing new is needed underneath. A session is already a workflow rather than a process, the +running set already comes from Temporal visibility, the store is already shared, and a live tail +already re-reads so a subscriber sees work another process is doing. What was missing was a way to +say so from a command line, which is these three: + +```bash +# hand over a prompt and walk away; prints the session id and exits +opencode session start "port the auth module to the new API" --attach http://gateway:4096 + +# what is this deployment running right now, across every client that ever connected +opencode session running --attach http://gateway:4096 + +# follow one from anywhere, and stop when the turn stops +opencode session watch ses_abc123 --attach http://gateway:4096 + +# a turn nobody starts: the firing needs no client and no serve process +opencode session schedule "review yesterday.s merges" --cron "0 9 * * *" --attach http://gateway:4096 +``` + +`--attach` takes any serve in the deployment, because they are interchangeable: each one reads the +same store and signals the same workflows. There is no "the server that owns this session". That is +the property, and it is why these commands are plain HTTP clients with no Temporal dependency. +`$OPENCODE_SERVER` sets the endpoint once. For an interactive terminal instead of a follower, +`opencode attach --session ` already puts the TUI on a remote session. + +To run it as a deployment rather than a laptop: + +```bash +export OPENCODE_SESSION_EXECUTION=temporal +export OPENCODE_DB_URL=libsql://... # one store, so any worker resumes any session +export TEMPORAL_ADDRESS=... + +OPENCODE_TEMPORAL_ROLE=worker bun run packages/server/src/worker.ts # as many as you want +OPENCODE_TEMPORAL_ROLE=client opencode serve --port 4096 # as many as you want +``` + +### Verified + +`packages/temporal/scripts/detached-session-check.sh` runs the whole claim against real processes: +serve A starts a turn and is killed with a tool still running, the turn finishes on a standalone +worker, and serve B (which never saw the session) reports it running and replays the transcript. +Then `session start` returns without waiting, `session running` lists it, and `session watch` +follows it live from a cold client and exits when the turn ends. `session schedule` then creates a +schedule and the check waits for a firing to run a turn with no client involved at all. + +How it decides that has been wrong in both directions, so it is worth stating. Nothing publishes a +turn-level ending, and the running set holds a session for the supervisor's whole idle period, so +neither answers the question on its own. What ends a watch is a terminal step and then a quiet wire, +with the settling state kept across reconnects: the stream has no replay, and a turn that ends while +the client is reconnecting publishes into a gap. Absence from the running set, asked periodically, +is the backstop for that gap, and it is only ever allowed to end the wait, never to prolong it. + +The shared store is load-bearing, and the check proves it rather than assuming it: give serve B its +own `OPENCODE_DB` and the three cross-process assertions fail (`active` returns `{}`, the replay is +empty, the follower hangs) while the serve-A-and-worker ones still pass. + +### Across two machines + +`packages/temporal/scripts/cross-host-check.sh` runs the claim against containers, where each worker +has its own filesystem and hostname and the store is a real libSQL server. A session writes a file +on worker A, worker A's host is killed, and worker B, whose project volume is empty, continues the +same session and reads that file back. + +That check found a bug a single host cannot show. `WorktreeMaterializer.ensure` treated any existing +directory as somebody's working copy, and a fresh host has no tip note, so `behind` said no and the +tree was never built. The tools then ran against an empty directory and the model was told a wrong +answer, which is worse than a failure. On one host the case never appears: worker B either has the +project already or has no directory at all, and an absent directory materializes fine. A mounted +empty directory is the shape of a machine that has never seen the session, and it now materializes +too (`packages/core/test/worktree-materialize.test.ts` covers it). + +The compose file mounts the engine's source over the image, so a code change does not need a new +image. One libSQL server, so this shows a shared store over a network rather than one that survives +losing a node. + +Still to do. A turn started from a schedule or a webhook needs an entry point of its own; +`session start` is a command, so something has to run it. And the deployment above is a set of +environment variables rather than a supported mode, so defaults, migration-on-deploy, and +credential distribution are still the operator's problem. + ## Porting this pattern The shape transfers to any agent engine; Temporal is one executor behind a seam the engine owns. diff --git a/packages/temporal/docker/Dockerfile b/packages/temporal/docker/Dockerfile new file mode 100644 index 000000000000..48f699b5829f --- /dev/null +++ b/packages/temporal/docker/Dockerfile @@ -0,0 +1,31 @@ +# A worker (or a serve) as its own machine. Running this in containers is what turns "any worker +# resumes any session" from a claim about processes into a claim about hosts: each of these has its +# own filesystem, its own hostname, and nothing of the session on disk. What they share is the +# Temporal cluster and one libSQL store, which is exactly what the README asks an operator to set up. + +FROM oven/bun:1.3.14 + +# python3 and a compiler are here for one dependency: a tree-sitter grammar builds from source at +# install time. git is not incidental either: the snapshot packs a worker rebuilds a worktree from are git packs, so a +# host that has never seen the project needs it to materialize the tree. +RUN apt-get update \ + && apt-get install -y --no-install-recommends git ca-certificates procps curl \ + python3 make g++ \ + && rm -rf /var/lib/apt/lists/* \ + && git config --system user.email opencode@example.com \ + && git config --system user.name opencode \ + && git config --system init.defaultBranch main \ + && git config --system --add safe.directory '*' + +WORKDIR /app + +COPY package.json bun.lock bunfig.toml tsconfig.json* ./ +COPY patches ./patches +COPY packages ./packages +RUN bun install --frozen-lockfile + +ENV OPENCODE_SESSION_EXECUTION=temporal + +# The worker by default. The serve role overrides this in compose; both build the same application +# context, so the only difference is whether an HTTP surface comes with it. +CMD ["bun", "run", "packages/server/src/worker.ts"] diff --git a/packages/temporal/docker/compose.yml b/packages/temporal/docker/compose.yml new file mode 100644 index 000000000000..a063433fa795 --- /dev/null +++ b/packages/temporal/docker/compose.yml @@ -0,0 +1,103 @@ +# Two workers that are two machines, not two processes on one. +# +# What they share is what an operator is told to share: one Temporal cluster and one libSQL store. +# What they do not share is the session's working tree. `worker-a` has the project, `worker-b` gets +# an empty volume, so a session that moves between them has to rebuild the tree from the snapshot +# packs in the store. That is the part a single host can never really test, because there the tree +# is already sitting on the disk the other process is reading. +# +# docker compose -f packages/temporal/docker/compose.yml up -d temporal sqld serve worker-a +# +# Not covered: one libSQL server, so this shows a shared store over a network rather than a store +# that survives losing a node. + +name: opencode-l3 + +# A mapping rather than a list, because a list cannot be merged: a service that adds one variable +# would otherwise replace the whole set and silently lose the store. +x-env: &env + OPENCODE_SESSION_EXECUTION: temporal + TEMPORAL_ADDRESS: temporal:7233 + # One store for every host. Without it a session belongs to whichever machine holds its file. + OPENCODE_DB_URL: http://sqld:8080 + OPENCODE_TEMPORAL_STEPPED: "1" + OPENCODE_SERVER_PASSWORD: ${OPENCODE_SERVER_PASSWORD:-l3-check} + OPENAI_API_KEY: ${OPENAI_API_KEY:?set OPENAI_API_KEY} + +x-app: &app + image: opencode-temporal:l3 + # The engine's own source, over the copy baked into the image. bun runs TypeScript directly, so + # this is the same code the image would have had; mounting it keeps a one-file change from + # costing a full dependency install, which is most of the build. + volumes: + - ../../core/src:/app/packages/core/src:ro + - ../src:/app/packages/temporal/src:ro + depends_on: + temporal: + condition: service_healthy + sqld: + condition: service_started + +services: + temporal: + image: temporalio/admin-tools:1.29 + # The image's own entrypoint is `sleep infinity`, so a command alone becomes arguments to sleep. + entrypoint: ["temporal"] + command: ["server", "start-dev", "--ip", "0.0.0.0", "--log-level", "warn"] + ports: + - "7243:7233" + healthcheck: + test: ["CMD", "temporal", "operator", "cluster", "health", "--address", "127.0.0.1:7233"] + interval: 5s + timeout: 5s + retries: 40 + + sqld: + image: ghcr.io/tursodatabase/libsql-server:latest + environment: + - SQLD_NODE=primary + ports: + - "8081:8080" + + # Drives workflows, hosts no worker, and is the only thing with an HTTP surface. + serve: + <<: *app + environment: + <<: *env + OPENCODE_TEMPORAL_ROLE: client + # Absolute, because working_dir is the project rather than the checkout: a relative entry path + # would be looked for inside the session's tree. + command: ["bun", "run", "/app/packages/cli/src/index.ts", "serve", "--port", "4096", "--hostname", "0.0.0.0"] + working_dir: /project + ports: + - "4096:4096" + volumes: + - ../../core/src:/app/packages/core/src:ro + - ../src:/app/packages/temporal/src:ro + - project-a:/project + + worker-a: + <<: *app + environment: + <<: *env + OPENCODE_TEMPORAL_ROLE: worker + volumes: + - ../../core/src:/app/packages/core/src:ro + - ../src:/app/packages/temporal/src:ro + - project-a:/project + + # No project volume of its own that has ever seen this session: an empty tree, so the worktree has + # to come from the packs in the store. + worker-b: + <<: *app + environment: + <<: *env + OPENCODE_TEMPORAL_ROLE: worker + volumes: + - ../../core/src:/app/packages/core/src:ro + - ../src:/app/packages/temporal/src:ro + - project-b:/project + +volumes: + project-a: + project-b: diff --git a/packages/temporal/scripts/cross-host-check.sh b/packages/temporal/scripts/cross-host-check.sh new file mode 100755 index 000000000000..9670e5a4018d --- /dev/null +++ b/packages/temporal/scripts/cross-host-check.sh @@ -0,0 +1,129 @@ +#!/usr/bin/env bash +# Any worker resumes any session, across machines rather than across processes. +# +# On one host the second worker already has the project on disk, so the interesting half of the +# claim is never exercised: the tree is there whether or not anything shipped it. Here worker B is a +# container with an empty project volume, so a session that moves to it has to bring its worktree +# along, out of the snapshot packs in the shared store. +# +# Usage: OPENAI_API_KEY=... packages/temporal/scripts/cross-host-check.sh +# +# Not covered: one libSQL server, so this shows a shared store over a network rather than one that +# survives losing a node. + +set -uo pipefail + +cd "$(dirname "${BASH_SOURCE[0]}")/../../.." +COMPOSE="docker compose -f packages/temporal/docker/compose.yml" +MODEL_ID="${MODEL_ID:-gpt-5-mini}" + +fails=0 +ok() { printf 'PASS %s\n' "$1"; } +bad() { printf 'FAIL %s (%s)\n' "$1" "${2:-}"; fails=$((fails + 1)); } + +# KEEP=1 leaves the stack up, which is the difference between reading a failure and guessing at it. +cleanup() { [ -n "${KEEP:-}" ] || $COMPOSE down -v >/dev/null 2>&1; } +trap cleanup EXIT + +[ -n "${OPENAI_API_KEY:-}" ] || { echo "set OPENAI_API_KEY"; exit 1; } + +$COMPOSE down -v >/dev/null 2>&1 +# Only when the image is missing. The compose file mounts the engine's source over the image, so a +# code change does not need a new one, and the dependency install is most of the build. +if ! docker image inspect opencode-temporal:l3 >/dev/null 2>&1; then + docker build -f packages/temporal/docker/Dockerfile -t opencode-temporal:l3 . >/dev/null \ + || { echo "build failed"; exit 1; } +fi +$COMPOSE up -d temporal sqld serve worker-a >/dev/null 2>&1 || { echo "stack failed"; exit 1; } + +api() { curl -s -u "opencode:$PW" "$@"; } + +# The serve generates its own password on first boot and prefers it over the environment, so ask it +# rather than tell it. +PW="" +for _ in $(seq 1 60); do + PW=$($COMPOSE exec -T serve sh -c 'cat /root/.local/state/opencode/password 2>/dev/null' 2>/dev/null | tr -d '\r\n') + [ -n "$PW" ] && break + sleep 3 +done +[ -n "$PW" ] && ok "serve is up" || { bad "serve never came up"; exit 1; } + +hostA=$($COMPOSE exec -T worker-a hostname 2>/dev/null | tr -d '\r') +[ -n "$hostA" ] && ok "worker A is a host of its own ($hostA)" || bad "worker A came up" + +# A project only worker A and serve can see. +$COMPOSE exec -T serve sh -c \ + 'cd /project && git init -q 2>/dev/null; echo hello > README.md; git add -A; git commit -qm init' \ + >/dev/null 2>&1 + +new_session() { + api -X POST http://127.0.0.1:4096/api/session -H 'content-type: application/json' \ + -d '{"directory":"/project"}' | sed -n 's/^{"data":{"id":"\([^"]*\)".*/\1/p' +} +prompt() { + api -o /dev/null -X POST "http://127.0.0.1:4096/api/session/$1/prompt" \ + -H 'content-type: application/json' -d "{\"prompt\":{\"text\":$2}}" +} +# A turn is over when a step of it ends on "stop", which is not the same as the session leaving the +# running set: the supervisor stays open for its idle timeout with nothing left to do. Counted +# rather than matched, because the history of a second turn still contains the first one's ending, +# and matching would call every later turn finished before it started. +stops() { + local body + body=$(api "http://127.0.0.1:4096/api/session/$1/history?limit=100" 2>/dev/null) + case "$body" in *InvalidRequestError*) echo " history rejected: $body" >&2; echo -1; return ;; esac + printf '%s' "$body" | grep -o '"finish":"stop"' | wc -l | tr -d ' ' +} +await_turn() { + local before=$2 + for _ in $(seq 1 90); do + [ "$(stops "$1")" -gt "$before" ] && return 0 + sleep 4 + done + return 1 +} + +sid=$(new_session) +[ -n "$sid" ] && ok "a session was created ($sid)" || { bad "no session"; exit 1; } +api -o /dev/null -X POST "http://127.0.0.1:4096/api/session/$sid/model" \ + -H 'content-type: application/json' -d "{\"model\":{\"id\":\"$MODEL_ID\",\"providerID\":\"openai\"}}" + +# --- turn 1 on worker A: writes a file, so a snapshot of the tree is captured and shipped +before=$(stops "$sid") +prompt "$sid" '"Use the bash tool to run exactly: echo TRAVELLED > /project/note.txt && cat /project/note.txt. Report the output."' +await_turn "$sid" "$before" && ok "turn 1 finished on worker A" || bad "turn 1 never finished" + +packs=$(curl -s http://127.0.0.1:8081/v2/pipeline -H 'content-type: application/json' \ + -d '{"requests":[{"type":"execute","stmt":{"sql":"select count(*) from snapshot_pack"}},{"type":"close"}]}' \ + 2>/dev/null | grep -o '"value":"[0-9]*"' | head -1 | grep -o '[0-9]*') +[ "${packs:-0}" -gt 0 ] && ok "the tree was shipped to the shared store ($packs packs)" \ + || bad "no snapshot packs reached the store" "$packs" + +# --- worker A's host goes away, and a host that has never seen this project takes over +docker kill "$($COMPOSE ps -q worker-a)" >/dev/null 2>&1 +sleep 2 +[ -z "$($COMPOSE ps -q --status running worker-a)" ] && ok "worker A's host is gone" || bad "worker A's host is gone" + +$COMPOSE up -d worker-b >/dev/null 2>&1 +sleep 8 +hostB=$($COMPOSE exec -T worker-b hostname 2>/dev/null | tr -d '\r') +[ "$hostB" != "$hostA" ] && ok "worker B is a different host ($hostB)" || bad "worker B is a different host" +empty=$($COMPOSE exec -T worker-b sh -c 'ls -A /project | wc -l' 2>/dev/null | tr -d '\r ') +[ "${empty:-1}" = "0" ] && ok "worker B's project is empty before the turn" || bad "worker B's project was not empty" "$empty" + +# --- turn 2 on worker B: the file only exists there if the worktree travelled +before=$(stops "$sid") +prompt "$sid" '"Use the bash tool to run exactly: cat /project/note.txt. Report exactly what it printed."' +await_turn "$sid" "$before" && ok "turn 2 finished on worker B" || bad "turn 2 never finished" + +# Asked of worker B's own disk rather than of the transcript. The transcript still holds turn 1, +# where the file did exist, so anything matched across the whole of it proves nothing about B. +landed=$($COMPOSE exec -T worker-b sh -c 'cat /project/note.txt 2>&1' 2>/dev/null | tr -d '\r') +case "$landed" in + TRAVELLED*) ok "the worktree travelled to worker B" ;; + *) bad "the worktree travelled to worker B" "$landed" ;; +esac + +echo +[ "$fails" -eq 0 ] && echo "cross-host-check: OK" || echo "cross-host-check: $fails failed" +exit $([ "$fails" -eq 0 ] && echo 0 || echo 1) diff --git a/packages/temporal/scripts/detached-session-check.sh b/packages/temporal/scripts/detached-session-check.sh new file mode 100755 index 000000000000..5d72794c0c1d --- /dev/null +++ b/packages/temporal/scripts/detached-session-check.sh @@ -0,0 +1,177 @@ +#!/usr/bin/env bash +# Proves the claim a durable session is supposed to make: it belongs to the deployment, not to +# whoever started it. One worker, two serve processes, one shared store, and a client that is only +# ever a client. +# +# 1. serve A starts a turn, then A is killed while a tool is still running +# 2. the turn finishes anyway, on a worker that is a separate process +# 3. serve B, which never saw the session, reports it running and replays the whole transcript +# 4. `session start` hands over a prompt and returns, holding no terminal +# 5. `session watch` follows that turn live from a cold client and stops when the turn stops +# +# Needs: bun, the temporal CLI, and an OpenAI key. Nothing here is a unit test; it is the evidence +# for a claim that only shows up across processes. +# +# Usage: OPENAI_API_KEY=... packages/temporal/scripts/detached-session-check.sh + +set -uo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +OC="$ROOT/packages/opencode/src/index.ts" +RUN="${RUN_DIR:-/private/tmp/opencode-l3}" +PORT_TEMPORAL="${PORT_TEMPORAL:-7240}" +PORT_A="${PORT_A:-4610}" +PORT_B="${PORT_B:-4611}" +MODEL="${MODEL:-openai/gpt-5-mini}" + +fails=0 +ok() { printf 'PASS %s\n' "$1"; } +bad() { printf 'FAIL %s (%s)\n' "$1" "${2:-}"; fails=$((fails + 1)); } + +pids=() +cleanup() { + for pid in "${pids[@]:-}"; do + [ -n "$pid" ] || continue + kill -9 $(pgrep -P "$pid" 2>/dev/null) "$pid" 2>/dev/null + done +} +trap cleanup EXIT + +[ -n "${OPENAI_API_KEY:-}" ] || { echo "set OPENAI_API_KEY"; exit 1; } + +rm -rf "$RUN"; mkdir -p "$RUN/proj" "$RUN/logs" +git -C "$RUN/proj" init -q +echo hello > "$RUN/proj/README.md" +git -C "$RUN/proj" add -A +git -C "$RUN/proj" -c user.email=a@b.c -c user.name=t commit -qm init + +export OPENCODE_SESSION_EXECUTION=temporal +export TEMPORAL_ADDRESS="127.0.0.1:$PORT_TEMPORAL" +# One store both serves and the worker read. This is what makes any process able to answer for any +# session; without it a session belongs to the host holding its file. +export OPENCODE_DB="$RUN/shared.db" +export OPENCODE_TEMPORAL_STEPPED=1 +# A stored password wins over the environment for the v2 serve, so a script that invents one gets +# 401 on every call. Take what the server will actually be asking for. +STORED="${XDG_STATE_HOME:-$HOME/.local/state}/opencode/password" +if [ -f "$STORED" ]; then + OPENCODE_SERVER_PASSWORD="$(cat "$STORED")" +else + OPENCODE_SERVER_PASSWORD="${OPENCODE_SERVER_PASSWORD:-l3-check}" +fi +export OPENCODE_SERVER_PASSWORD + +temporal server start-dev --port "$PORT_TEMPORAL" --ui-port $((PORT_TEMPORAL + 1000)) --log-level warn \ + > "$RUN/logs/temporal.log" 2>&1 & +pids+=($!) +sleep 6 + +OPENCODE_TEMPORAL_ROLE=worker bun run "$ROOT/packages/server/src/worker.ts" > "$RUN/logs/worker.log" 2>&1 & +worker=$!; pids+=($worker) + +cd "$RUN/proj" +OPENCODE_TEMPORAL_ROLE=client bun run "$ROOT/packages/cli/src/index.ts" serve --port "$PORT_A" \ + > "$RUN/logs/serveA.log" 2>&1 & +serveA=$!; pids+=($serveA) +OPENCODE_TEMPORAL_ROLE=client bun run "$ROOT/packages/cli/src/index.ts" serve --port "$PORT_B" \ + > "$RUN/logs/serveB.log" 2>&1 & +pids+=($!) + +A="http://127.0.0.1:$PORT_A" +B="http://127.0.0.1:$PORT_B" +AUTH="opencode:$OPENCODE_SERVER_PASSWORD" + +# Bounded, because a fixed sleep is either a slow script or a flaky one. Both serves boot a whole +# application context, which on a cold module cache is not quick. +# Answering at all is not enough: an unauthorized answer is still an answer, and treating it as +# ready turns a credentials problem into a confusing timeout later. +wait_for() { + for _ in $(seq 1 60); do + [ "$(curl -s -o /dev/null -w '%{http_code}' -u "$AUTH" "$1/api/session")" = "200" ] && return 0 + sleep 2 + done + return 1 +} +wait_for "$A" && wait_for "$B" || { echo "serves never came up; see $RUN/logs"; exit 1; } + +# The id of the session, not of anything nested in it: the field is read off the first line of the +# document, so a later `"id"` (a model, a message) cannot be picked up instead. +session_id() { sed -n 's/^{"data":{"id":"\([^"]*\)".*/\1/p' | head -1; } + +# --- 1. a turn started on serve A, long enough to still be running when A dies +created=$(curl -s -u "$AUTH" -X POST "$A/api/session" -H 'content-type: application/json' \ + -d "{\"directory\":\"$RUN/proj\"}") +sid=$(printf '%s' "$created" | session_id) +[ -n "$sid" ] && ok "serve A created a session" || { bad "serve A created a session" "$created"; exit 1; } + +provider=${MODEL%%/*}; model=${MODEL#*/} +curl -s -o /dev/null -u "$AUTH" -X POST "$A/api/session/$sid/model" -H 'content-type: application/json' \ + -d "{\"model\":{\"id\":\"$model\",\"providerID\":\"$provider\"}}" +curl -s -o /dev/null -u "$AUTH" -X POST "$A/api/session/$sid/prompt" -H 'content-type: application/json' \ + -d '{"prompt":{"text":"Use the bash tool to run exactly: sleep 40 && echo SURVIVED. Then report the output."}}' +sleep 18 +pgrep -f "sleep 40 && echo SURVIVED" > /dev/null && ok "the tool is running on the worker" \ + || bad "the tool is running on the worker" "it never started" + +# --- 2. kill the process that started it, mid-tool +kill -9 $(pgrep -P $serveA 2>/dev/null) $serveA 2>/dev/null +sleep 3 +[ -z "$(lsof -nP -iTCP:$PORT_A -sTCP:LISTEN 2>/dev/null)" ] && ok "serve A is gone" || bad "serve A is gone" +pgrep -f "sleep 40 && echo SURVIVED" > /dev/null && ok "the turn outlived the client that started it" \ + || bad "the turn outlived the client that started it" "the tool died with serve A" + +# --- 3. serve B, which never saw this session, knows it and can replay it +running=$(curl -s -u "$AUTH" "$B/api/session/active") +case "$running" in *"$sid"*) ok "serve B reports it running" ;; *) bad "serve B reports it running" "$running" ;; esac + +sleep 35 +timeout 30 curl -s -N -u "$AUTH" "$B/api/session/$sid/event" > "$RUN/logs/replay.txt" 2>&1 +grep -q "SURVIVED" "$RUN/logs/replay.txt" && ok "serve B replays work done while no client existed" \ + || bad "serve B replays work done while no client existed" + +# --- 4. start a turn and walk away +started=$(timeout 90 bun run "$OC" session start \ + "Use the bash tool to run exactly: sleep 20 && echo WATCHED. Then report the output." \ + --attach "$B" --model "$MODEL" --dir "$RUN/proj" --json 2>/dev/null) +sid2=$(printf '%s' "$started" | sed -n 's/.*"id":"\([^"]*\)".*/\1/p') +[ -n "$sid2" ] && ok "session start returned an id without waiting" || bad "session start returned an id" "$started" + +listed=$(timeout 60 bun run "$OC" session running --attach "$B" --json 2>/dev/null) +case "$listed" in *"$sid2"*) ok "session running lists it" ;; *) bad "session running lists it" "$listed" ;; esac + +# --- 5. follow it live from a client that has never seen it, and stop when the turn stops +began=$(date +%s) +timeout 120 bun run "$OC" session watch "$sid2" --attach "$B" > "$RUN/logs/watch.txt" 2>&1 +took=$(( $(date +%s) - began )) +grep -q "WATCHED" "$RUN/logs/watch.txt" && ok "session watch followed the turn" \ + || bad "session watch followed the turn" "$(tail -3 "$RUN/logs/watch.txt")" +[ "$took" -lt 100 ] && ok "session watch stopped when the turn did (${took}s)" \ + || bad "session watch stopped when the turn did" "${took}s, so it hung" + +# --- 6. a turn nobody starts. The schedule is a Temporal object, so at firing time there is no +# client and no HTTP call: the workflow admits the prompt itself and starts the session's own +# supervisor. Prompted into a fresh session so what arrives can only have come from the firing. +sched=$(timeout 90 bun run "$OC" session schedule \ + "Use the bash tool to run exactly: echo SCHEDULED. Then report the output." \ + --every 10s --attach "$B" --dir "$RUN/proj" --json 2>/dev/null) +sid3=$(printf '%s' "$sched" | sed -n 's/.*"session":"\([^"]*\)".*/\1/p') +scheduleId=$(printf '%s' "$sched" | sed -n 's/.*"schedule":"\([^"]*\)".*/\1/p') +[ -n "$sid3" ] && ok "session schedule created one" || bad "session schedule created one" "$sched" + +# Long enough for a firing plus a turn, and nothing here prompts it. +answered="" +for _ in $(seq 1 24); do + sleep 5 + answered=$(curl -s -u "$AUTH" "$B/api/session/$sid3/message" 2>/dev/null || true) + case "$answered" in *SCHEDULED*) break ;; esac +done +case "$answered" in + *SCHEDULED*) ok "a firing ran a turn with no client involved" ;; + *) bad "a firing ran a turn with no client involved" "$(printf '%s' "$answered" | head -c 200)" ;; +esac +[ -n "$scheduleId" ] && temporal schedule delete --schedule-id "$scheduleId" \ + --address "127.0.0.1:$PORT_TEMPORAL" >/dev/null 2>&1 + +echo +[ "$fails" -eq 0 ] && echo "detached-session-check: OK" || echo "detached-session-check: $fails failed" +exit $([ "$fails" -eq 0 ] && echo 0 || echo 1) diff --git a/packages/temporal/src/boundary.ts b/packages/temporal/src/boundary.ts index 579f52024096..a70710d10f0f 100644 --- a/packages/temporal/src/boundary.ts +++ b/packages/temporal/src/boundary.ts @@ -36,6 +36,20 @@ const halted = (sessionID: string, declined?: SessionRunDeclinedError) => { }) } +// 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. Without this a libsql blip during a seal failed the step for good rather than moving +// 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, @@ -63,7 +77,7 @@ export const runAtBoundary = async ( throw ApplicationFailure.create({ message: squashed?.message ?? Cause.pretty(cause), type: squashed?._tag ?? "SessionRunError", - nonRetryable: true, + 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 0fbe624b414c..93adfef85194 100644 --- a/packages/temporal/src/config.ts +++ b/packages/temporal/src/config.ts @@ -3,6 +3,7 @@ export * as TemporalConfig from "./config" // Connection and behavior settings for the Temporal executor. The executor reads them at layer // build: an embedder or a test provides the service to override, and absent that the values come // from env. Nothing reads env at module load, so import order carries no configuration. +import { readFileSync } from "node:fs" import { Context } from "effect" import { DEFAULTS } from "./protocol" @@ -11,11 +12,22 @@ import { DEFAULTS } from "./protocol" // worker's bundler); `worker` runs a standalone activity worker with no HTTP surface. export type Role = "both" | "client" | "worker" +// Which deployment this is. The settings below are not independent: a fleet whose store is not +// shared is a set of workers that cannot see each other's sessions, and finding that out takes a +// session that answers with the wrong files. `fleet` sets what has to agree, and `preflight` +// refuses what cannot. +export type Profile = "local" | "fleet" + export interface Interface { + readonly profile: Profile readonly address: string readonly namespace: string readonly taskQueue: string readonly role: Role + /** How a server that is not the dev server is reached: an API key for Cloud, a certificate pair + * for a cluster with mTLS. Read from files, never from argv, and never logged. */ + readonly apiKey?: string + readonly tls?: { readonly cert: string; readonly key: string; readonly ca?: string } | true /** 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: @@ -29,17 +41,134 @@ export interface Interface { /** The worktree this worker serves, when affinity is on. Defaults to the process directory, which * is what a serve process with an embedded worker is already sitting in. */ readonly worktree?: string + /** 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 not needed while a step + * is pinned to one worker, which is the default. */ + readonly serialTools?: boolean + /** Send the tools and the seal of a step back to the worker that made its model call, on a queue + * that worker polls on its own. That worker is standing in the tree the tools are about to write, + * so the step's tools see each other's writes through the filesystem and can run at once. On by + * default: a pin nobody answers falls back to the shared queue after a schedule-to-start bound, + * so the worst it costs is that wait. `OPENCODE_TEMPORAL_STEP_AFFINITY=0` turns it off. */ + readonly stepAffinity?: boolean } export class Service extends Context.Service()("@opencode/temporal/Config") {} +const read = (path: string | undefined) => (path ? readFileSync(path, "utf8") : undefined) +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 => ({ + profile: process.env.OPENCODE_TEMPORAL_PROFILE === "fleet" ? "fleet" : "local", 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", + apiKey: process.env.OPENCODE_TEMPORAL_API_KEY ?? read(process.env.OPENCODE_TEMPORAL_API_KEY_FILE), + tls: + process.env.OPENCODE_TEMPORAL_TLS_CERT && process.env.OPENCODE_TEMPORAL_TLS_KEY + ? { + cert: readFileSync(process.env.OPENCODE_TEMPORAL_TLS_CERT, "utf8"), + key: readFileSync(process.env.OPENCODE_TEMPORAL_TLS_KEY, "utf8"), + ca: read(process.env.OPENCODE_TEMPORAL_TLS_CA), + } + : process.env.OPENCODE_TEMPORAL_TLS === "1" + ? true + : undefined, idleTimeout: process.env.OPENCODE_SESSION_IDLE_TIMEOUT, - stepped: process.env.OPENCODE_TEMPORAL_STEPPED === "1", + // A fleet's unit of work is the smaller one: a worker dying takes one tool call with it rather + // than a whole step, and a tool call is where the retry policy and the approval belong. + stepped: onOff("OPENCODE_TEMPORAL_STEPPED", process.env.OPENCODE_TEMPORAL_PROFILE === "fleet"), worktreeAffinity: process.env.OPENCODE_TEMPORAL_WORKTREE_AFFINITY === "1", worktree: process.env.OPENCODE_TEMPORAL_WORKTREE, + stepAffinity: process.env.OPENCODE_TEMPORAL_STEP_AFFINITY !== "0", + serialTools: + process.env.OPENCODE_TEMPORAL_SERIAL_TOOLS === "1" || + (process.env.OPENCODE_TEMPORAL_SERIAL_TOOLS !== "0" && + process.env.OPENCODE_TEMPORAL_STEP_AFFINITY === "0" && + process.env.OPENCODE_TEMPORAL_WORKTREE_AFFINITY !== "1" && + !!process.env.OPENCODE_DB_URL), +}) + +/** What `Connection.connect` and `NativeConnection.connect` both take, built once so a client and a + * worker in different processes cannot disagree about how the cluster is reached. */ +export const connectionOptions = (config: Interface) => { + const tls = + config.tls === true || (config.apiKey && config.tls === undefined) + ? true + : config.tls + ? { + clientCertPair: { crt: Buffer.from(config.tls.cert), key: Buffer.from(config.tls.key) }, + ...(config.tls.ca ? { serverRootCACertificate: Buffer.from(config.tls.ca) } : {}), + } + : undefined + return { + address: config.address, + ...(tls ? { tls } : {}), + ...(config.apiKey ? { apiKey: config.apiKey } : {}), + } +} + +const LOOPBACK = /^(127\.0\.0\.1|localhost|\[::1\]|0\.0\.0\.0)(:|$)/ + +/** + * What is wrong with this deployment, said before it takes work rather than after. Each of these + * fails as something else: a store only one process can see reads as a worker that never picks + * anything up, and a client with no worker anywhere reads as a session that accepts a prompt and + * never answers it. + */ +export const preflight = (config: Interface): string[] => { + const problems: string[] = [] + const shared = !!process.env.OPENCODE_DB_URL + if (config.profile === "fleet") { + if (!shared) + problems.push( + "the fleet profile needs OPENCODE_DB_URL: the store is the record, and workers that do " + + "not share it cannot serve each other's sessions", + ) + if (config.role === "both") + problems.push( + "OPENCODE_TEMPORAL_ROLE is `both` in a fleet: a serve that also polls is a laptop " + + "deployment. Run `client` next to standalone `worker` processes", + ) + } + if (config.apiKey && LOOPBACK.test(config.address)) + problems.push(`an API key is set but TEMPORAL_ADDRESS is ${config.address}, which is a dev server`) + if (config.apiKey && config.namespace === "default") + problems.push("an API key is set but TEMPORAL_NAMESPACE is `default`, which is not a Cloud namespace") + if (!!process.env.OPENCODE_TEMPORAL_TLS_CERT !== !!process.env.OPENCODE_TEMPORAL_TLS_KEY) + problems.push("OPENCODE_TEMPORAL_TLS_CERT and OPENCODE_TEMPORAL_TLS_KEY come as a pair") + return problems +} + +/** + * Worth saying, not worth refusing. Only what cannot work belongs in `preflight`, because a process + * that exits takes a deployment with it, and plaintext to an address that is not loopback is a + * private network in most deployments and a mistake in some. Nothing here can tell which. + */ +export const notes = (config: Interface): string[] => { + const said: string[] = [] + if (!LOOPBACK.test(config.address) && !config.apiKey && !config.tls) + said.push( + `reaching ${config.address} in plaintext. For Temporal Cloud set OPENCODE_TEMPORAL_API_KEY; ` + + "for a cluster with mTLS set OPENCODE_TEMPORAL_TLS_CERT and OPENCODE_TEMPORAL_TLS_KEY", + ) + return said +} + +/** Every setting that decides how this process behaves, and nothing that is a credential. */ +export const describe = (config: Interface): Record => ({ + profile: config.profile, + address: config.address, + namespace: config.namespace, + taskQueue: config.taskQueue, + role: config.role, + store: process.env.OPENCODE_DB_URL ? "shared (OPENCODE_DB_URL)" : "this process only", + stepped: String(config.stepped === true), + stepAffinity: String(config.stepAffinity !== false), + serialTools: String(config.serialTools === true), + credentials: config.apiKey ? "api key" : config.tls ? "certificate pair" : "none (plaintext)", }) diff --git a/packages/temporal/src/executor.ts b/packages/temporal/src/executor.ts index 94987e603aa5..eede29325a7f 100644 --- a/packages/temporal/src/executor.ts +++ b/packages/temporal/src/executor.ts @@ -1,6 +1,7 @@ export * as SessionExecutionTemporal from "./executor" import { fileURLToPath } from "node:url" +import { hostname } from "node:os" import { Effect, Layer, Option } from "effect" import { Client, Connection, WithStartWorkflowOperation } from "@temporalio/client" // Imported lazily inside the worker branch: the worker package drags webpack and swc (it bundles @@ -15,8 +16,8 @@ import { SessionStore } from "@opencode-ai/core/session/store" import { SessionExecution } from "@opencode-ai/core/session/execution" import { makeStepActivities, makeSteppedTurnActivities } from "./activities" import { makeDrains } from "./drain" -import { makeL2Drains } from "./l2-drain" -import { queueForWorktree } from "./queue" +import { makeL2Drains, makeScheduleDrains } from "./l2-drain" +import { queueForWorktree, queueForWorker } from "./queue" import { Database } from "@opencode-ai/core/database/database" import { ProjectTable } from "@opencode-ai/core/project/sql" import { eq } from "drizzle-orm" @@ -66,6 +67,9 @@ const layer = Layer.effect( // 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 AFFINITY = config.worktreeAffinity === true // The tree this process serves when affinity is on. A serve process with an embedded worker is // already sitting in it, so the process directory is the right default. @@ -73,6 +77,13 @@ const layer = Layer.effect( // Which queue a worker polls. With affinity off this is the one shared queue and any worker can // draw any session, rebuilding the tree if it has to. const POLL_QUEUE = AFFINITY ? queueForWorktree(TASK_QUEUE, SERVED_WORKTREE) : TASK_QUEUE + // The queue this worker polls on its own, so a step can be sent back to it. Keyed by host as + // well as directory: two containers serve `/project` and share none of it. Only workers have + // one, and only they report it, so a client-only process never pins a step to itself. + const STEP_QUEUE = + HOST_WORKER && config.stepAffinity !== false + ? queueForWorker(TASK_QUEUE, hostname(), SERVED_WORKTREE) + : undefined // Which queue a session's workflow runs on. Keyed on the PROJECT worktree, not the session's // directory: `worktrees.ensure` rebuilds the project tree, so keying on the directory a session // happened to start in would split one physical tree across a queue per subdirectory, and a @@ -91,6 +102,13 @@ const layer = Layer.effect( return project ? queueForWorktree(TASK_QUEUE, project.worktree) : TASK_QUEUE }) : Effect.succeed(TASK_QUEUE) + // Before anything is accepted, not after: every one of these fails as something else later, and + // the failure lands on whoever prompted the session rather than on whoever deployed it. + for (const note of TemporalConfig.notes(config)) yield* Effect.logInfo(`configuration: ${note}`) + const problems = TemporalConfig.preflight(config) + for (const problem of problems) yield* Effect.logError(`configuration: ${problem}`) + if (problems.length > 0) yield* Effect.die(`this deployment cannot serve sessions: ${problems[0]}`) + const events = yield* EventV2.Service const worktrees = yield* WorktreeMaterializer.Service @@ -99,7 +117,10 @@ const layer = Layer.effect( const { stepDrain } = 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 l2 = makeL2Drains({ store, locations, ctx, events, worktrees }) + const l2 = makeL2Drains({ store, locations, ctx, events, worktrees, stepQueue: STEP_QUEUE }) + // What a schedule fires into: admitting a prompt is a row in the store, and a workflow cannot + // write one. Registered on every worker, because a firing lands wherever one is polling. + const schedules = makeScheduleDrains({ db, events, ctx }) // Worker connection (native) hosts the runTurnStep activity + the workflow. Skipped in // client-only role so serve can run without an embedded worker. @@ -115,7 +136,7 @@ const layer = Layer.effect( ), ) const nativeConn = yield* Effect.acquireRelease( - Effect.promise(() => NativeConnection.connect({ address: ADDRESS })), + Effect.promise(() => NativeConnection.connect(TemporalConfig.connectionOptions(config))), (conn) => Effect.promise(() => conn.close().catch(() => {})), ) const worker = yield* Effect.promise(() => @@ -124,7 +145,11 @@ const layer = Layer.effect( namespace: NAMESPACE, taskQueue: POLL_QUEUE, workflowsPath: fileURLToPath(new URL("./workflow.ts", import.meta.url)), - activities: { ...makeStepActivities(stepDrain), ...makeSteppedTurnActivities(l2) }, + activities: { + ...makeStepActivities(stepDrain), + ...makeSteppedTurnActivities(l2), + promptSession: schedules.promptDrain, + }, }), ) const runHandle = worker.run() @@ -135,6 +160,33 @@ const layer = Layer.effect( await runHandle.catch(() => {}) }), ) + + // A second poller, on this worker's own queue, for the steps pinned to it. Activities only: + // the workflow runs wherever it was started, and only the work that has to come back here is + // addressed here. Without it a pin has nobody to answer it and every step pays the + // schedule-to-start wait before falling back. + if (STEP_QUEUE) { + const pinnedWorker = yield* Effect.promise(() => + Worker.create({ + connection: nativeConn, + namespace: NAMESPACE, + taskQueue: STEP_QUEUE, + activities: { + ...makeStepActivities(stepDrain), + ...makeSteppedTurnActivities(l2), + promptSession: schedules.promptDrain, + }, + }), + ) + const pinnedHandle = pinnedWorker.run() + pinnedHandle.catch(() => {}) + yield* Effect.addFinalizer(() => + Effect.promise(async () => { + pinnedWorker.shutdown() + await pinnedHandle.catch(() => {}) + }), + ) + } } // Worker-only process: it hosts activities but drives no workflows, so the client methods are @@ -161,7 +213,7 @@ const layer = Layer.effect( // Client connection drives the per-session workflows. const clientConn = yield* Effect.acquireRelease( - Effect.promise(() => Connection.connect({ address: ADDRESS })), + Effect.promise(() => Connection.connect(TemporalConfig.connectionOptions(config))), (conn) => Effect.promise(() => conn.close().catch(() => {})), ) const client = new Client({ connection: clientConn, namespace: NAMESPACE }) @@ -178,6 +230,7 @@ const layer = Layer.effect( startWithWake: true, idleTimeout: IDLE_TIMEOUT, stepped: STEPPED, + serialTools: SERIAL_TOOLS, } satisfies WF.SessionTurnOptions, ], signal: WF.wake, @@ -247,6 +300,7 @@ const layer = Layer.effect( startWithWake: false, idleTimeout: IDLE_TIMEOUT, stepped: STEPPED, + serialTools: SERIAL_TOOLS, } satisfies WF.SessionTurnOptions, ], workflowIdConflictPolicy: "USE_EXISTING", diff --git a/packages/temporal/src/l2-drain.ts b/packages/temporal/src/l2-drain.ts index b61d721eb3b9..f2e278ab6d1e 100644 --- a/packages/temporal/src/l2-drain.ts +++ b/packages/temporal/src/l2-drain.ts @@ -19,7 +19,10 @@ import type { DeferredToolCall, ToolCallOutcome } from "@opencode-ai/core/sessio import type { StepSettlement } from "@opencode-ai/core/session/runner/publish-llm-event" import { SessionSchema } from "@opencode-ai/core/session/schema" import { SessionStore } from "@opencode-ai/core/session/store" -import type { SessionInput } from "@opencode-ai/core/session/input" +import { SessionInput } from "@opencode-ai/core/session/input" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { Prompt } from "@opencode-ai/schema/prompt" +import type { Database } from "@opencode-ai/core/database/database" import { runAtBoundary } from "./boundary" import type { StepDrainInput, StepDrainResult } from "./drain" @@ -39,6 +42,11 @@ export type ModelCallDrainResult = /** 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 + /** The queue this worker polls on its own, when it has one. The tools of this step write the + * tree this worker is standing in, so sending them here keeps them on it. Absent when the + * worker was not given a queue of its own, and never required: the step falls back to the + * shared queue and the tree is rebuilt there. */ + readonly queue?: string } export interface ToolCallDrainInput { @@ -66,9 +74,47 @@ export interface L2DrainDeps { readonly ctx: Context.Context readonly events: EventV2.Interface readonly worktrees: WorktreeMaterializer.Interface + /** The queue this worker polls on its own, reported by the model call so the rest of the step can + * be sent back to it. Absent when the worker has none. */ + readonly stepQueue?: string } -export const makeL2Drains = ({ store, locations, ctx, events, worktrees }: L2DrainDeps) => { +/** + * Admit a prompt to a session that already exists, without waking anything. + * + * This is what a start with no client is made of. A prompt is a durable row before it is work, and + * writing that row needs the store, which a workflow cannot reach; waking the session is the + * workflow's own job (it starts or signals the session's supervisor). Separating the two is what + * lets a schedule fire into a deployment where nothing is running but workers. + * + * Idempotent on the message id, which the workflow derives from the firing, so a re-driven activity + * admits nothing twice. + */ +export const makeScheduleDrains = ({ + db, + events, + ctx, +}: { + readonly db: Database.Interface["db"] + readonly events: EventV2.Interface + readonly ctx: Context.Context +}) => ({ + promptDrain: async (input: { readonly sessionID: string; readonly messageID: string; readonly text: string }) => + SessionInput.admit(db, events, { + id: SessionMessage.ID.make(input.messageID), + sessionID: SessionSchema.ID.make(input.sessionID), + prompt: Prompt.make({ text: input.text }), + delivery: "queue", + }).pipe( + Effect.asVoid, + Effect.provideService(EventV2.EventOwner, `schedule:${input.messageID}`), + Effect.provide(ctx), + Effect.scoped, + Effect.runPromise, + ), +}) + +export const makeL2Drains = ({ store, locations, ctx, events, worktrees, stepQueue }: L2DrainDeps) => { // One session, one owner, a present project tree. `claim` is true only for the model call: it is // the writer that supersedes a previous attempt, and the rest of the step rides its token. const inSession = ( @@ -135,6 +181,7 @@ export const makeL2Drains = ({ store, locations, ctx, events, worktrees }: L2Dra assistantMessageID: result.assistantMessageID, needsContinuation: result.needsContinuation, owner: input.owner, + ...(stepQueue === undefined ? {} : { queue: stepQueue }), }, ), ), diff --git a/packages/temporal/src/l2-step.ts b/packages/temporal/src/l2-step.ts index 60aaefd27240..3e6a2f1edad7 100644 --- a/packages/temporal/src/l2-step.ts +++ b/packages/temporal/src/l2-step.ts @@ -12,7 +12,7 @@ // each // other; what is lost is the overlap between the model and its own tools. -import { ActivityFailure, type ApplicationFailure } from "@temporalio/workflow" +import { ActivityFailure, type ApplicationFailure, TimeoutFailure } from "@temporalio/workflow" import { HALTED_FAILURE_TYPE } from "./protocol" import type { StepDrainInput, StepDrainResult } from "./drain" import type { @@ -36,6 +36,17 @@ export const isHaltFailure = (error: unknown) => error instanceof ActivityFailure && (error.cause as ApplicationFailure | undefined)?.type === HALTED_FAILURE_TYPE +/** + * Nobody took the work. This is the only failure a pinned dispatch is allowed to answer by moving + * the work elsewhere: it means the queue was not polled, so the activity never started and no side + * effect can have happened. Every other failure has to be reported as itself, because a tool that + * ran and then failed must not be run again somewhere else. + */ +export const isUnclaimedFailure = (error: unknown) => + error instanceof ActivityFailure && + error.cause instanceof TimeoutFailure && + error.cause.timeoutType === "SCHEDULE_TO_START" + /** The three activities a stepped turn drives. */ export interface SteppedActivities { readonly runModelCall: (input: ModelCallDrainInput) => Promise @@ -56,6 +67,23 @@ export interface SteppedTurnDeps { * 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, and it is what pinning a + * step's tools to one worker buys back. */ + readonly serial?: boolean + /** The same activities, addressed to one worker's own queue. A step's tools write the tree the + * model call's worker is standing in, so keeping them there is what lets them run at once: they + * see each other's writes through the filesystem rather than through the store. Only offered a + * queue the model call reported, and only used while that worker is still polling. */ + readonly pinnedTo?: (queue: string) => Pick + /** Whether a failure means nobody took the work, which is the one kind a pinned dispatch answers + * by trying the shared queue instead. */ + readonly isUnclaimed?: (error: unknown) => 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 } /** @@ -64,24 +92,118 @@ export interface SteppedTurnDeps { * of a whole-step activity. */ export const makeSteppedTurn = - ({ activities, isCancellation, isHalt, log }: SteppedTurnDeps) => + ({ + activities, + isCancellation, + isHalt, + log, + serial, + nonCancellable, + pinnedTo, + isUnclaimed, + }: 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 + // The worker that made the model call, when it offered its own queue. Everything else in this + // step goes to it first, because it is the host holding the tree the tools are about to write. + const pinned = model.queue && pinnedTo ? pinnedTo(model.queue) : undefined + let unclaimed = false + // Pinned first, shared queue if nobody took it. `isUnclaimed` is the whole safety of that + // fallback: it is true only when the activity never started, so nothing can run twice. Once one + // dispatch has fallen back, the rest of the step goes straight to the shared queue: that worker + // is gone, and every later pin would pay the schedule-to-start wait to learn it again. + // What is left of a step whose worker is gone goes to the shared queue one at a time. There it + // can land on two hosts again, which is the case `serial` exists for, so the rule it applies + // from the start is applied here to the remainder. + let shared: Promise = Promise.resolve() + const onShared = (run: (on: SteppedActivities) => Promise): Promise => { + const next = shared.then( + () => run(activities), + () => run(activities), + ) + shared = next.then( + () => undefined, + () => undefined, + ) + return next + } + const viaPinned = async ( + run: (on: Pick) => Promise, + ): Promise => { + if (!pinned || !isUnclaimed) return run(activities) + if (unclaimed) return onShared(run) + try { + return await run(pinned) + } catch (error) { + if (!isUnclaimed(error)) throw error + unclaimed = true + log?.("the worker that ran the model call is gone; the step moves to the shared queue", { + sessionID: input.sessionID, + step: model.step, + }) + return onShared(run) + } + } + // 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 dispatched = await Promise.allSettled( - model.calls.map((call) => - activities.runToolCall({ sessionID: input.sessionID, call, owner: model.owner }), - ), - ) + const dispatch = (call: (typeof model.calls)[number]) => + viaPinned((on) => on.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 seal = (stopped: boolean) => + viaPinned((on) => + on.sealStep({ + 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 recorded 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, + }), + ) + for (const outcome of dispatched) { if (outcome.status !== "rejected") continue - if (isCancellation(outcome.reason) || isHalt(outcome.reason)) throw outcome.reason + 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. Letting the seal decide is what once carried the agent + // on past a declined permission, because it re-derived "keep going" from the tool parts. + // 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 @@ -100,12 +222,5 @@ export const makeSteppedTurn = if (unsettled.length > 0) log?.("step did not settle every call it dispatched", { step: model.step, calls: unsettled }) - return activities.sealStep({ - sessionID: input.sessionID, - step: model.step, - settlement: model.settlement, - assistantMessageID: model.assistantMessageID, - needsContinuation: model.needsContinuation, - owner: model.owner, - }) + return seal(false) } diff --git a/packages/temporal/src/queue.ts b/packages/temporal/src/queue.ts index 8e12e33f7047..ec5b0a4d3477 100644 Binary files a/packages/temporal/src/queue.ts and b/packages/temporal/src/queue.ts differ diff --git a/packages/temporal/src/supervisor.ts b/packages/temporal/src/supervisor.ts index d279dcfc5164..b2182d9e2a8c 100644 --- a/packages/temporal/src/supervisor.ts +++ b/packages/temporal/src/supervisor.ts @@ -45,6 +45,11 @@ export interface SupervisorRuntime { /** Restart the run with fresh history, carrying whether work is still pending. History-keeping * drivers only (Temporal). */ readonly continueAsNew?: (sessionID: string, startWithWake: boolean) => Promise + /** Whether the driver says this run's history is large enough to roll over. A drain count cannot + * answer this: one drain is a whole turn, and a stepped turn of 200 steps is thousands of events, + * so a handful of drains can cross the server's limit long before the count does. Optional: + * drivers without a history return false. */ + readonly historyWantsRollover?: () => boolean } export interface WorkflowOptions { @@ -81,12 +86,24 @@ export const makeSupervisor = (rt: SupervisorRuntime, options?: WorkflowOptions) .runInDrainScope(async () => { drains++ if (drains >= MAX_DRAINS_PER_RUN) rolloverPending = true + if (rt.historyWantsRollover?.()) rolloverPending = true let step = 1 let promotion: string | null = null let first = true for (;;) { const r: StepDrainResult = await rt.runTurnStep({ sessionID, step, promotion, first, force }) + // Inside the loop as well, because one drain is a whole turn: a long one outgrows the + // history without ever reaching the next drain's check. + if (rt.historyWantsRollover?.()) rolloverPending = true if (!r.continue) break + // A queued prompt continues this same drain as a fresh turn, so a session fed without a + // gap never goes quiet and the rollover it is waiting for never happens. Stop at that + // boundary instead and let the new run pick the queue up: the work is not lost, it is + // one turn later. A steer is not a boundary, so it still rides this drain through. + if (rolloverPending && r.promotion === "queue") { + pendingWake = true + break + } step = r.step promotion = r.promotion first = false diff --git a/packages/temporal/src/workflow.ts b/packages/temporal/src/workflow.ts index a395a39dd674..42e91f9ef0a6 100644 --- a/packages/temporal/src/workflow.ts +++ b/packages/temporal/src/workflow.ts @@ -18,11 +18,16 @@ import { CancellationScope, isCancellation, allHandlersFinished, + workflowInfo, log, + startChild, + getExternalWorkflowHandle, + ParentClosePolicy, } from "@temporalio/workflow" +import { WorkflowExecutionAlreadyStartedError } from "@temporalio/common" import type { StepActivities, SteppedTurnActivities } from "./activities" -import { isHaltFailure, makeSteppedTurn } from "./l2-step" -import { SIGNALS, RESUME_UPDATE } from "./protocol" +import { isHaltFailure, isUnclaimedFailure, makeSteppedTurn } from "./l2-step" +import { SIGNALS, RESUME_UPDATE, WORKFLOW_ID_PREFIX } from "./protocol" import { makeSupervisor, type SupervisorRuntime } from "./supervisor" const activityOptions = { @@ -46,9 +51,39 @@ const { runTurnStep } = proxyActivities(activityOptions) const { runModelCall } = proxyActivities(activityOptions) const { runToolCall } = proxyActivities(activityOptions) // Sealing is a snapshot, a diff and one event. It should not inherit a turn-sized backstop. -const { sealStep } = proxyActivities({ - ...activityOptions, - startToCloseTimeout: "10 minutes", +const sealOptions = { ...activityOptions, startToCloseTimeout: "10 minutes" } as const +const { sealStep } = proxyActivities(sealOptions) + +// How long a pinned activity waits for the worker that ran the model call to take it. It is polling +// its own queue, so this is the time to notice it is gone rather than a queueing delay: nobody else +// can take the work while it stands. Long enough to ride out a restart, short enough that a dead +// worker does not hold the step for a noticeable part of a turn. +const PINNED_SCHEDULE_TO_START = "30 seconds" + +/** The same two activities, addressed to one worker's own queue. Built per queue rather than once, + * because the queue is not known until the model call reports it; that report comes out of history, + * so this is deterministic on replay. */ +const pinnedTo = (taskQueue: string) => ({ + runToolCall: proxyActivities({ + ...activityOptions, + taskQueue, + scheduleToStartTimeout: PINNED_SCHEDULE_TO_START, + }).runToolCall, + sealStep: proxyActivities({ + ...sealOptions, + taskQueue, + scheduleToStartTimeout: PINNED_SCHEDULE_TO_START, + }).sealStep, +}) + +// Admitting a prompt is a row in the store, so it is an activity; it is small and it must not hold +// a firing open if no worker is polling. +const { promptSession } = proxyActivities<{ + promptSession(input: { sessionID: string; messageID: string; text: string }): Promise +}>({ + startToCloseTimeout: "2 minutes", + scheduleToCloseTimeout: "30 minutes", + retry: { maximumAttempts: 10 }, }) export const wake = defineSignal(SIGNALS.wake) @@ -109,20 +144,28 @@ const runtime: SupervisorRuntime = { allHandlersFinished, continueAsNew: (sessionID, startWithWake) => continueAsNew(sessionID, { startWithWake }), + // The server's own read of whether this run has grown enough to roll over. The drain count alone + // misses it: a stepped turn is thousands of events, so a handful of drains can cross the limit. + historyWantsRollover: () => workflowInfo().continueAsNewSuggested, } // Same supervisor, different step body: wake, interrupt, idle timeout and continue-as-new are -// unchanged, and only what "one step" means differs. -const steppedRuntime: SupervisorRuntime = { +// 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, + isUnclaimed: isUnclaimedFailure, + pinnedTo, + 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 @@ -141,6 +184,10 @@ export interface SessionTurnOptions { /** 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 { @@ -148,15 +195,56 @@ export async function sessionTurn(sessionID: string, options?: SessionTurnOption const startWithWake = options?.startWithWake ?? true const idleTimeout = options?.idleTimeout const stepped = options?.stepped === true + const serialTools = options?.serialTools === true if (!idleTimeout && !stepped) return workflows.sessionTurn(sessionID, startWithWake) return makeSupervisor( { - ...(stepped ? steppedRuntime : runtime), + ...(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 }), + continueAsNew(id, { + startWithWake: wake, + idleTimeout, + stepped, + serialTools, + }), }, idleTimeout ? { idleTimeout } : undefined, ).sessionTurn(sessionID, startWithWake) } + +/** + * A turn nobody started. + * + * A schedule fires this, and it runs where no client and no serve process exist: the prompt is + * admitted by an activity, because it is a row in the store, and the session's own supervisor is + * started as an abandoned child (or signalled, when it is already running). Nothing here waits for + * the turn: this workflow's job is to hand the work over and finish, which is what makes a firing + * cheap and a missed one visible in the schedule rather than in a run that never ends. + * + * The message id comes from the firing's own workflow id, so a re-drive admits the same prompt + * rather than a second one. + */ +export async function scheduledPrompt(input: { + readonly sessionID: string + readonly text: string + readonly session?: SessionTurnOptions +}): Promise { + const messageID = `msg_sched_${workflowInfo().workflowId}`.slice(0, 60) + await promptSession({ sessionID: input.sessionID, messageID, text: input.text }) + const options: SessionTurnOptions = { ...input.session, startWithWake: true } + try { + await startChild(sessionTurn, { + workflowId: `${WORKFLOW_ID_PREFIX}${input.sessionID}`, + args: [input.sessionID, options], + parentClosePolicy: ParentClosePolicy.ABANDON, + }) + } catch (error) { + // The session is already being driven, which is the ordinary case for a schedule that fires + // faster than a turn takes. The prompt is admitted either way; what it needs is a wake, because + // a supervisor waiting out its idle period is not watching the store. + if (!(error instanceof WorkflowExecutionAlreadyStartedError)) throw error + await getExternalWorkflowHandle(`${WORKFLOW_ID_PREFIX}${input.sessionID}`).signal(wake) + } +} diff --git a/packages/temporal/test/l2-step.test.ts b/packages/temporal/test/l2-step.test.ts index f92ad5abddd4..7ec8e513faf5 100644 --- a/packages/temporal/test/l2-step.test.ts +++ b/packages/temporal/test/l2-step.test.ts @@ -35,12 +35,7 @@ const INPUT: StepDrainInput = { force: false, } const SEALED: StepDrainResult = { ran: true, continue: true, step: 3, promotion: "steer" } -const call = (id: string, name = "probe_write") => ({ - id, - name, - input: {}, - assistantMessageID: "msg_1", -}) +const call = (id: string, name = "probe_write") => ({ id, name, assistantMessageID: "msg_1" }) const fakes = ( model: ModelCallDrainResult, @@ -150,7 +145,47 @@ describe("stepped turn", () => { expect(result).toEqual(SEALED) }) - it("lets an interrupt end the turn instead of sealing it", async () => { + // 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. + it("runs a step's tools one at a time when told to", 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: true })(INPUT) + expect(tools).toHaveLength(3) + expect(overlapped).toBe(false) + }) + + // And still overlap when nothing is moving, which is the case the split was measured on. + it("runs them together when it is not", async () => { + let inFlight = 0 + let overlapped = false + const { activities } = 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 })(INPUT) + expect(overlapped).toBe(true) + }) + + 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 () => { @@ -160,12 +195,15 @@ describe("stepped turn", () => { const run = makeSteppedTurn({ activities, isCancellation, isHalt })(INPUT) - // A cancellation is not a failed tool. Swallowing it would close a step the user stopped. + // A cancellation is not a failed tool, so it still ends the turn. The step is closed on the way + // out all the same: a stop landing here used to publish no step event at all, and a follower + // waiting on the turn hung. What the seal must not do is decide the turn keeps going. await expect(run).rejects.toBeInstanceOf(FakeCancel) - expect(seals).toHaveLength(0) + expect(seals).toHaveLength(1) + expect(seals[0]?.needsContinuation).toBe(false) }) - it("lets a user halt end the turn instead of sealing it", async () => { + 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 () => { @@ -175,12 +213,142 @@ describe("stepped turn", () => { const run = makeSteppedTurn({ activities, isCancellation, isHalt })(INPUT) - // A decline crosses the activity boundary as an ordinary failure, not a cancel, so without a - // separate test for it the dispatcher would seal the step and the turn would carry on past the - // user's refusal. + // 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 is what once 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) + }) +}) + +// Pinning a step to the worker that made its model call, and what happens when that worker is gone. +// The pin is what lets a step's tools run at once: they write one tree through one filesystem +// instead of shipping it to each other. The fallback is what keeps that from being a worse kind of +// stuck than the shared queue was. +describe("stepped turn, pinned to a worker", () => { + const unclaimed = () => + new ActivityFailure( + "activity failed", + "runToolCall", + "1", + 1 as never, + undefined, + new TimeoutFailure("schedule to start timed out", undefined, "SCHEDULE_TO_START" as never), + ) + const isUnclaimed = (error: unknown) => + error instanceof ActivityFailure && + error.cause instanceof TimeoutFailure && + error.cause.timeoutType === "SCHEDULE_TO_START" + + const called = (model: ModelCallDrainResult) => { + const shared = fakes(model) + const pinnedTools: ToolCallDrainInput[] = [] + const pinnedSeals: SealDrainInput[] = [] + let refuse = false + let refused = 0 + const pinned = { + runToolCall: async (input: ToolCallDrainInput): Promise => { + if (refuse) { + refused++ + throw unclaimed() + } + pinnedTools.push(input) + return { outcome: "settled" } + }, + sealStep: async (input: SealDrainInput): Promise => { + if (refuse) { + refused++ + throw unclaimed() + } + pinnedSeals.push(input) + return SEALED + }, + } + return { + ...shared, + pinnedTools, + pinnedSeals, + refusals: () => refused, + goneAfterModelCall: () => { + refuse = true + }, + run: () => + makeSteppedTurn({ + activities: shared.activities, + isCancellation, + isHalt, + isUnclaimed, + pinnedTo: (queue) => { + expect(queue).toBe("queue-of-the-worker") + return pinned + }, + })(INPUT), + } + } + + const withQueue: ModelCallDrainResult = { + kind: "called", + step: 2, + calls: [call("call_a"), call("call_b")], + owner: "run:1:1", + queue: "queue-of-the-worker", + } + + it("sends the tools and the seal back to the worker that made the model call", async () => { + const { run, pinnedTools, pinnedSeals, tools, seals } = called(withQueue) + + await run() + + expect(pinnedTools.map((t) => t.call.id)).toEqual(["call_a", "call_b"]) + expect(pinnedSeals).toHaveLength(1) + // Nothing reached the shared queue, which is the point: the tree the tools wrote is on that + // worker and nowhere else until the step ships it. + expect(tools).toHaveLength(0) expect(seals).toHaveLength(0) }) + + it("moves the step to the shared queue when nobody takes the pinned work", async () => { + const { run, goneAfterModelCall, pinnedTools, tools, seals } = called(withQueue) + goneAfterModelCall() + + const result = await run() + + // Schedule-to-start is the one failure that says the activity never started, so moving the work + // cannot run a tool twice. Both calls end up on the shared queue, and the step still closes. + expect(pinnedTools).toHaveLength(0) + expect(tools.map((t) => t.call.id).sort()).toEqual(["call_a", "call_b"]) + expect(seals).toHaveLength(1) + expect(result).toEqual(SEALED) + }) + + it("does not offer the pin again once the worker has failed to answer", async () => { + const { run, goneAfterModelCall, tools, seals, refusals } = called({ + ...withQueue, + calls: [call("call_a")], + }) + goneAfterModelCall() + + await run() + + // One refusal, from the tool. The seal that follows goes straight to the shared queue rather + // than spending another schedule-to-start bound on a worker already known to be gone. Counting + // the refusals is the assertion: the work reaches the shared queue either way, so where it + // ended up says nothing about how long the step spent finding out. Calls dispatched together + // do each pay it once, because none of them has learned anything yet when they start. + expect(refusals()).toBe(1) + expect(tools).toHaveLength(1) + expect(seals).toHaveLength(1) + }) + + it("uses the shared queue when the model call reported no queue of its own", async () => { + const { run, tools, pinnedTools } = called({ ...withQueue, queue: undefined }) + + await run() + + expect(tools).toHaveLength(2) + expect(pinnedTools).toHaveLength(0) + }) }) // The bug this predicate exists for was a mismatch between what `boundary.ts` throws and what the