From 13bbae0d4d514880fb3939c1979e135c81653cad Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Mon, 14 Sep 2026 17:40:08 -0700 Subject: [PATCH] Refused a directory a tool call never came back from. A timeout settles the workflow's promise without stopping the process behind it, so a later step landing on the same host can rebuild or capture a directory an abandoned tool is still writing. A marker is written before a tool can touch anything and removed when its body returns; while one stands for another step, the host refuses the directory, or moves it aside and rebuilds when it built the directory itself. The refusal crosses the activity boundary as retryable work with its own delay, so another host takes the step. --- .../core/src/session/execution/worktree.ts | 153 +++++++- packages/core/src/snapshot/tip.ts | 30 ++ packages/core/src/snapshot/writers.ts | 349 ++++++++++++++++++ .../core/test/worktree-materialize.test.ts | 273 ++++++++++++++ packages/temporal/README.md | 13 +- packages/temporal/src/boundary.ts | 11 + packages/temporal/src/drain.ts | 8 +- packages/temporal/src/executor.ts | 2 +- packages/temporal/src/stepped-drain.ts | 44 ++- packages/temporal/src/stepped-turn.ts | 6 +- .../temporal/test/boundary-refusal.test.ts | 42 +++ .../test/stepped-drain-writers.test.ts | 97 ++++- 12 files changed, 994 insertions(+), 34 deletions(-) create mode 100644 packages/core/src/snapshot/writers.ts create mode 100644 packages/temporal/test/boundary-refusal.test.ts diff --git a/packages/core/src/session/execution/worktree.ts b/packages/core/src/session/execution/worktree.ts index 6d31f200a138..dadc6b51b52f 100644 --- a/packages/core/src/session/execution/worktree.ts +++ b/packages/core/src/session/execution/worktree.ts @@ -12,7 +12,8 @@ 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 { readdir, rm, writeFile } from "node:fs/promises" +import { access, mkdir, readdir, rename, rm, stat, writeFile } from "node:fs/promises" +import { constants } from "node:fs" import path from "path" import { Cause, Context, Effect, Layer, Schema } from "effect" import { ChildProcess } from "effect/unstable/process" @@ -27,7 +28,14 @@ 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" +import { + forgetWorktreeTip, + markWorktreeBuilt, + readWorktreeTip, + worktreeWasBuilt, + writeWorktreeTip, +} from "../../snapshot/tip" +import * as Writers from "../../snapshot/writers" export interface Interface { /** @@ -45,7 +53,22 @@ export interface Interface { * 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, options?: { readonly pauseBeforeLock?: number }) => Effect.Effect + readonly ensure: ( + directory: string, + options?: { + readonly pauseBeforeLock?: number + /** + * The step asking for the directory. A call of another step that never came back keeps it, + * because a settled workflow promise does not stop the process behind it. Omitted by callers + * that are not a step, and then any stranded call refuses them. + */ + readonly current?: Writers.Writer + }, + ) => Effect.Effect + + /** Say a call is about to write the directory, and that its body came back. */ + readonly beginWrite: (directory: string, writer: Writers.Writer) => Effect.Effect + readonly endWrite: (directory: string, callID: string) => Effect.Effect } export class Service extends Context.Service()("@opencode/v2/WorktreeMaterializer") {} @@ -54,6 +77,16 @@ export class Service extends Context.Service()("@opencode/v2 // branch over a full untracked tree. const RESTORED = "refs/heads/opencode-restore" +/** + * The directory holds a call from another step that never came back, so this step may not have it. + * Tagged separately from a rebuild failure because it is not a transient one: it stands until that + * call returns or an operator clears it. + */ +export class WorktreeQuarantinedError extends Schema.TaggedErrorClass()( + "WorktreeMaterializer.QuarantinedError", + { message: Schema.String }, +) {} + /** 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", @@ -79,7 +112,7 @@ const layer = Layer.effect( const { db } = yield* Database.Service const locks = KeyedMutex.makeUnsafe() - const materialize = Effect.fnUntraced(function* (tip: typeof SnapshotPackTable.$inferSelect) { + const materialize = Effect.fnUntraced(function* (tip: typeof SnapshotPackTable.$inferSelect, fromEmpty = false) { const worktree = AbsolutePath.make(tip.worktree) const repository = yield* git.repo .create({ worktree, gitDirectory: AbsolutePath.make(path.join(worktree, ".git")) }) @@ -130,6 +163,21 @@ const layer = Layer.effect( ) .pipe(Effect.ignore) yield* writeWorktreeTip(global.data, tip.worktree, tip.tree) + // Which directories this host made rather than adopted, for the one decision that turns on + // it: whether this host may move the directory out of the way when a call that never came + // back makes it unusable. + if (fromEmpty) { + yield* markWorktreeBuilt(global.data, tip.worktree) + // Said while nothing is wrong, because the alternative is saying it when a tool call never + // comes back and this host has to be taken out of service for this directory by hand. + const stuck = yield* cannotMoveAside(tip.worktree) + if (stuck) + yield* Effect.logWarning("this worktree cannot be set aside if a call never comes back", { + worktree: tip.worktree, + because: stuck, + remedy: "mount the volume above the project rather than at it", + }) + } yield* Effect.logInfo("materialized worktree from snapshot packs", { worktree: tip.worktree, packs: rows.length, @@ -153,9 +201,93 @@ const layer = Layer.effect( return isBehind(rows, held) }) + /** + * Why this directory could not be moved out of the way if a call never came back in it, or + * nothing when it could. A mount point cannot be renamed, and neither can a directory whose + * parent this process may not write, and a host with either has only the refusal. + * + * A prediction, for a warning while nothing is wrong. What decides is the rename below, which + * attempts it and falls back: this one moves nothing to find out. + */ + const cannotMoveAside = Effect.fnUntraced(function* (worktree: string) { + return yield* Effect.promise(async () => { + try { + const [here, up] = await Promise.all([stat(worktree), stat(path.dirname(worktree))]) + // A mount point is a different device from the directory it hangs under. + if (here.dev !== up.dev) return "it is a mount point, and a mount point cannot be renamed" + await access(path.dirname(worktree), constants.W_OK) + return undefined + } catch (err) { + return `it could not be read: ${String((err as Error)?.message ?? err)}` + } + }) + }) + + /** + * Move a directory a call never came back from out of the way, so this host can have a fresh + * one. What this is for is the writer nothing can account for: it keeps writing the directory + * it already has open, which is now somewhere else, and nothing it does reaches what is built + * next. That is a closure where the readings that narrow it are not conclusive. + * + * Refused itself in three cases. A directory this host did not build out of an empty one is + * somebody's checkout. A directory holding a call of the step now asking for it has one of this + * host's own tools inside. And a mount point cannot be renamed at all, which is what a project + * directory mounted as a volume is: there the refusal stands. + */ + const moveAside = Effect.fnUntraced(function* (worktree: string, current?: Writers.Writer) { + if (!(yield* worktreeWasBuilt(global.data, worktree))) return undefined + if (current) { + const all = yield* Writers.strandedWriters(global.data, worktree) + if (all.some((w) => w.sessionID === current.sessionID && w.step === current.step)) return undefined + } + const moved = `${worktree}.stranded.${new Date().toISOString().replace(/[:.]/g, "-")}` + const renamed = yield* Effect.promise(() => + rename(worktree, moved).then( + () => true, + () => false, + ), + ) + if (!renamed) return undefined + yield* Effect.promise(() => mkdir(worktree, { recursive: true }).catch(() => {})) + // Both are about the directory that just moved: the note says what it held, and the markers + // say who was inside it. The one here now is empty and nobody is in it. + yield* forgetWorktreeTip(global.data, worktree) + yield* Writers.clearWriters(global.data, worktree) + return moved + }) + + const refuseWhenStranded = Effect.fn("WorktreeMaterializer.refuseWhenStranded")(function* ( + worktree: string, + current?: Writers.Writer, + ) { + const stranded = yield* Writers.strandedWriters(global.data, worktree, current) + if (stranded.length === 0) return + const moved = yield* moveAside(worktree, current) + if (moved !== undefined) { + yield* Effect.logWarning("moved a worktree a call never came back from", { + worktree, + moved, + calls: stranded.length, + }) + return + } + const one = stranded[0] + // Dies, like a rebuild that could not finish: the activity boundary turns it into a failure + // Temporal schedules again, and the next attempt can be taken by a host that is not refused. + return yield* Effect.die( + new WorktreeQuarantinedError({ + message: + `not using ${worktree}: ${stranded.length} tool call(s) from an earlier step never ` + + `returned (${one.callID} of session ${one.sessionID} step ${one.step}, pid ${one.pid}, ` + + `started ${one.started}). Still refused because ${one.because}, and this host may not ` + + `move the directory out of the way. Stop it before this directory is used again.`, + }), + ) + }) + const ensure = Effect.fn("WorktreeMaterializer.ensure")(function* ( directory: string, - options?: { readonly pauseBeforeLock?: number }, + options?: { readonly pauseBeforeLock?: number; readonly current?: Writers.Writer }, ) { // 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. @@ -168,6 +300,9 @@ const layer = Layer.effect( .pipe(Effect.orDie), ) if (!tip) return + // Before anything is rebuilt. A restore is what brings this host to the newest tree, and + // doing that under a call nobody can account for is what makes its later capture look current. + yield* refuseWhenStranded(tip.worktree, options?.current) // 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 @@ -189,7 +324,7 @@ const layer = Layer.effect( // 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( + yield* materialize(tip, !here).pipe( Effect.catchCauseIf( (cause) => !Cause.hasInterrupts(cause), (cause) => @@ -221,7 +356,11 @@ const layer = Layer.effect( ) }) - return Service.of({ ensure }) + return Service.of({ + ensure, + beginWrite: (directory, writer) => Writers.beginWrite(global.data, directory, writer), + endWrite: (directory, callID) => Writers.endWrite(global.data, directory, callID), + }) }), ) diff --git a/packages/core/src/snapshot/tip.ts b/packages/core/src/snapshot/tip.ts index b3495ea74dfd..8cd1177b3567 100644 --- a/packages/core/src/snapshot/tip.ts +++ b/packages/core/src/snapshot/tip.ts @@ -32,3 +32,33 @@ export const writeWorktreeTip = (data: string, worktree: string, tree: string) = await mkdir(path.dirname(file), { recursive: true }).catch(() => {}) await writeFile(file, tree).catch(() => {}) }) + +// Whether this host built the directory out of an empty one, as opposed to capturing one that was +// already somebody's. Kept beside the note rather than in it, so a note written by an older worker +// reads as what it is: a directory nobody here may move. +// +// Only a directory this host built may be moved out of the way when a call that never came back +// makes it unusable. Someone's checkout is not ours to move, and what it holds includes files no +// pack carries. +const builtFile = (data: string, worktree: string) => path.join(data, "worktree-tip", `${Hash.fast(worktree)}.built`) + +export const markWorktreeBuilt = (data: string, worktree: string) => + Effect.promise(async () => { + const file = builtFile(data, worktree) + await mkdir(path.dirname(file), { recursive: true }).catch(() => {}) + await writeFile(file, new Date().toISOString()).catch(() => {}) + }) + +export const forgetWorktreeTip = (data: string, worktree: string) => + Effect.promise(async () => { + await rm(noteFile(data, worktree), { force: true }).catch(() => {}) + await rm(builtFile(data, worktree), { force: true }).catch(() => {}) + }) + +export const worktreeWasBuilt = (data: string, worktree: string) => + Effect.promise(() => + readFile(builtFile(data, worktree), "utf8").then( + () => true, + () => false, + ), + ) diff --git a/packages/core/src/snapshot/writers.ts b/packages/core/src/snapshot/writers.ts new file mode 100644 index 000000000000..d126268085a0 --- /dev/null +++ b/packages/core/src/snapshot/writers.ts @@ -0,0 +1,349 @@ +// Which tool calls are inside their own execution on this host, kept in this host's data directory. +// +// A timeout settles the workflow's promise; it does not stop the process behind it. So after a +// step is abandoned, the host can still be running that step's tool, and nothing the workflow can +// see says whether it is. The step that was abandoned is protected by refusing to move it. What is +// not protected by that is everything after: the turn ends, the next prompt lands wherever there is +// room, and the directory that tool is writing is free again. +// +// A marker is written before a call can have any effect and removed when its body returns. While +// one stands, the directory belongs to that call's step and no other step may rebuild or capture +// it. The refusal outlives the process that made it, and takes itself back where this host can show +// that it is over. Four readings answer that, because a tool can put down anything the worker gave +// it: its process is gone, nothing carrying the name the worker exported is running, nothing is +// left in its control group or its process group, and nothing is standing in the directory. The +// machine not having restarted is what makes the pids in those readings mean anything. What is left +// after all of it is a tool that left the group, was handed an environment of somebody else's +// choosing, shares this process's own control group, and writes the directory without being in it. +// Until then the directory stays refused, which strands a directory and not a session: the work is +// scheduled again and another host can take it. + +import { execFile } from "node:child_process" +import { randomBytes } from "node:crypto" +import { mkdir, readdir, readFile, readlink, realpath, rm, writeFile } from "node:fs/promises" +import os from "node:os" +import path from "path" +import { promisify } from "node:util" +import { Effect } from "effect" +import { Hash } from "../util/hash" + +const run = promisify(execFile) + +// Put in the environment rather than kept in memory, because the point of it is to be inherited: +// a tool's children carry it wherever they end up, including through `setsid`, and a scan finds +// them after the worker that spawned them is gone. Fresh per process, so a worker never answers for +// its predecessor's children. Set at the first marker rather than at import, so loading this module +// changes nothing; a marker is written before its tool can spawn anything, so the children inherit +// it all the same. +const WORKER_ENV = "OPENCODE_WORKTREE_WRITER" +let name: string | undefined +const worker = () => (name ??= process.env[WORKER_ENV] = randomBytes(8).toString("hex")) + +/** What a tool call is, for telling this step's writers from an earlier one's. */ +export interface Writer { + readonly sessionID: string + readonly step: number + readonly callID: string +} + +interface WriterNote extends Writer { + readonly pid: number + // The group the writer's process was in. What a tool starts stays in it unless it asks for a + // group of its own, which is what every daemonizing wrapper does. + readonly pgid?: number + // The worker that ran the call, as a name it put in its own environment. Everything a tool starts + // inherits it, so this is what finds a child the group check lost. + readonly worker?: string + // The control group the writer was in. A tool keeps it through `setsid` and through `sudo`, and + // it can be read whoever owns the process, so it is what answers for a tool running as somebody + // else. Linux only. + readonly cgroup?: string + // When this machine last started, so a pid from before a restart is not read as a live one. + readonly bootAt?: number + readonly host?: string + readonly started: string +} + +// os.uptime has second granularity and drifts between reads, so this is a stamp to compare with a +// tolerance rather than an identifier. A restart moves it by the whole of the last uptime. +const bootAt = () => Math.round(Date.now() - os.uptime() * 1000) +const SAME_BOOT = 60_000 + +const alive = (pid: number) => { + try { + process.kill(pid, 0) + return true + } catch (err) { + // Somebody else's process is still a process. + return (err as NodeJS.ErrnoException).code === "EPERM" + } +} + +/** What is running on this host, in the four shapes a tool this worker started can be recognised + * by. Undefined when the host cannot be asked, which is answered as "everything is still here". + * `ps` is not installed on a slim container image, which is where most of these run, so Linux is + * read from `/proc` and everything else asks `ps`. */ +export interface LiveHere { + readonly groups: Set + readonly workers: Set + // Which control groups still hold a process. Linux only, and the only reading here that answers + // for a process running as somebody else: a control group is inherited across fork and exec, + // `setsid` does not change it, `sudo` does not change it, and `/proc//cgroup` is readable + // whoever owns the process, where that process's environment and working directory are not. + readonly cgroups: Set + // Processes standing in the directory itself: their working directory is inside it, or they hold + // a file under it open. This is the reading that does not depend on the tool having kept anything + // the worker gave it. This host's own processes only, on both platforms. + readonly holding: ReadonlyArray<{ readonly pid: number; readonly how: string }> + // Where this process itself stands. A marker naming either of these is one the reading cannot be + // asked about: a worker restarted from the same shell is in the group its predecessor was in, and + // in a container every process shares one control group. + readonly ourGroup?: number + readonly ourCgroup?: string +} + +const liveHere = async (directory: string): Promise => { + const groups = new Set() + const workers = new Set() + const cgroups = new Set() + const holding: { pid: number; how: string }[] = [] + // Through the links, because a temporary directory on a Mac is reached by one and every reading + // below comes back resolved. Comparing the two forms finds nothing, quietly. + const here = await realpath(directory).catch(() => directory) + const inside = (p?: string) => p !== undefined && (p === here || p.startsWith(`${here}/`)) + const nameOf = (text: string) => { + // The environment is NUL-separated in `/proc` and space-separated in `ps`, so this reads the + // name off either without pretending to parse the whole of it. + const at = text.indexOf(`${WORKER_ENV}=`) + return at < 0 ? undefined : text.slice(at + WORKER_ENV.length + 1).split(/[\0\s]/)[0] + } + // A tool of this worker's own is accounted for by its own marker, and the git subprocesses the + // snapshot store runs stand in the directory by design. Neither is a writer nobody can account + // for. + const ours = (held: string | undefined) => held === worker() + + try { + if (process.platform === "linux") { + for (const name of await readdir("/proc")) { + if (!/^\d+$/.test(name) || name === String(process.pid)) continue + const stat = await readFile(`/proc/${name}/stat`, "utf8").catch(() => undefined) + // A command can hold spaces and brackets, so the fields after it are counted from the last + // close bracket: state, ppid, pgrp. + const pgrp = stat ? Number.parseInt(stat.slice(stat.lastIndexOf(")") + 2).split(" ")[2], 10) : NaN + if (Number.isFinite(pgrp)) groups.add(pgrp) + // Readable whoever owns the process, unlike the two below it. + const cgroup = await readFile(`/proc/${name}/cgroup`, "utf8").catch(() => undefined) + const path = cgroup?.split("\n")[0]?.slice(cgroup.indexOf(":", cgroup.indexOf(":") + 1) + 1) + if (path) cgroups.add(path.trim()) + // Readable for this user's processes, which is what a tool of ours is. Anything else is not + // something this worker started, unless it went through `sudo`, and that one is what the + // control group above answers for. + const held = nameOf(await readFile(`/proc/${name}/environ`, "utf8").catch(() => "")) + if (held !== undefined) workers.add(held) + if (ours(held)) continue + const pid = Number.parseInt(name, 10) + const cwd = await readlink(`/proc/${name}/cwd`).catch(() => undefined) + if (inside(cwd)) { + holding.push({ pid, how: "its working directory is in it" }) + continue + } + for (const fd of await readdir(`/proc/${name}/fd`).catch(() => [] as string[])) { + const open = await readlink(`/proc/${name}/fd/${fd}`).catch(() => undefined) + if (!inside(open)) continue + holding.push({ pid, how: `it holds ${open} open` }) + break + } + } + return { groups, workers, cgroups, holding, ourGroup: await group(), ourCgroup: await cgroup() } + } + + // `-E` prints each process's environment after its command, for the processes this user owns. + // Without the name in its own environment, and standing somewhere else: `ps` and `lsof` are + // children of this process, so they inherit what we hold and where we are, and would otherwise + // report themselves as work this worker left behind. + const { [WORKER_ENV]: _ours, ...env } = process.env + const asked = { maxBuffer: 32 * 1024 * 1024, cwd: "/", env } + const { stdout } = await run("ps", ["-A", "-E", "-o", "pid=,pgid=,command="], asked) + const named = new Map() + for (const line of stdout.split("\n")) { + const [pid, pgid] = line + .trim() + .split(/\s+/, 2) + .map((n) => Number.parseInt(n, 10)) + if (!Number.isFinite(pid) || pid === process.pid) continue + if (Number.isFinite(pgid)) groups.add(pgid) + const held = nameOf(line) + if (held !== undefined) workers.add(held) + named.set(pid, held) + } + // One reading for every process, rather than walking the directory: `lsof +D` stats the whole + // tree, and the answer wanted here is about the processes, not about the files. + const cwds = await run("lsof", ["-a", "-d", "cwd", "-Fpn"], asked).catch(() => undefined) + let at: number | undefined + for (const line of cwds?.stdout.split("\n") ?? []) { + if (line.startsWith("p")) at = Number.parseInt(line.slice(1), 10) + if (!line.startsWith("n") || at === undefined || at === process.pid) continue + if (inside(line.slice(1)) && !ours(named.get(at))) { + holding.push({ pid: at, how: "its working directory is in it" }) + } + } + return { groups, workers, cgroups, holding, ourGroup: await group(), ourCgroup: await cgroup() } + } catch { + return undefined + } +} + +const readGroup = async (pid: number): Promise => { + try { + if (process.platform === "linux") { + const stat = await readFile(`/proc/${pid}/stat`, "utf8") + const pgrp = Number.parseInt(stat.slice(stat.lastIndexOf(")") + 2).split(" ")[2], 10) + return Number.isFinite(pgrp) ? pgrp : undefined + } + const { stdout } = await run("ps", ["-o", "pgid=", "-p", String(pid)]) + const pgid = Number.parseInt(stdout.trim(), 10) + return Number.isFinite(pgid) ? pgid : undefined + } catch { + return undefined + } +} + +// Once per process: a process cannot change the group it is in. +let ourGroup: Promise | undefined +const group = () => (ourGroup ??= readGroup(process.pid)) + +// Nor, without asking, the control group it is in. Absent off Linux, where there are none. +const readCgroup = async (): Promise => { + if (process.platform !== "linux") return undefined + const text = await readFile("/proc/self/cgroup", "utf8").catch(() => undefined) + const line = text?.split("\n")[0] + // `::`, and the path is the part that names the unit or container. + return line ? line.slice(line.indexOf(":", line.indexOf(":") + 1) + 1).trim() || undefined : undefined +} + +let ourCgroup: Promise | undefined +const cgroup = () => (ourCgroup ??= readCgroup()) + +/** + * Why this marker can still be a tool inside its own execution, or nothing when it cannot. That is + * the only thing the refusal is worth its cost for, and the reason is carried out of here because + * an operator is who acts on it. + * + * Everything here is a reason to stop refusing, never a reason to start: what cannot be answered is + * answered as still running. Exported for a test, because one of the readings it decides on is + * Linux's alone and a Mac cannot produce it. + */ +export const insideBecause = (note: WriterNote, live: LiveHere | undefined): string | undefined => { + // A pid from another machine says nothing here, and two hosts sharing one data directory is the + // only way to get one. Neither of them can see the other's processes. + if (note.host !== undefined && note.host !== os.hostname()) + return `it was written on ${note.host}, and this is ${os.hostname()}` + // Written by a worker that did not date its marker, so there is nothing to tell a live pid from + // a reused one. + if (note.bootAt === undefined) return "it does not say which boot its pid belongs to" + // The machine restarted. Nothing it was running came back with it. + if (Math.abs(note.bootAt - bootAt()) > SAME_BOOT) return undefined + if (alive(note.pid)) return `pid ${note.pid} is still running` + // The writer is gone, and what a tool starts can outlive it. Nothing below is asked of the pids + // themselves, which come round again; it is asked of what those processes carry and where they + // stand. + if (live === undefined) return "this host could not be asked what is running on it" + // Anything the worker started, wherever it ended up. A tool that daemonizes leaves the group and + // keeps the environment, which is why this is the check that decides most cases. + if (note.worker !== undefined && live.workers.has(note.worker)) return "something it started is still running" + // The control group, for a tool running as somebody else: `sudo` empties the environment it + // passes on and this survives it. Only when the writer was in a group of its own, which a service + // manager or a container per worker gives it: in the container this process is in, every process + // shares one, and finding ourselves there is not evidence about anything. + if (note.cgroup !== undefined && note.cgroup !== live.ourCgroup && live.cgroups.has(note.cgroup)) + return `something is still in ${note.cgroup}` + // The process group, for a tool that was given an environment of somebody else's choosing. Only + // when this process is somewhere else, for the reason above. + if (note.pgid !== undefined && note.pgid !== live.ourGroup && live.groups.has(note.pgid)) + return `something is still in process group ${note.pgid}` + // And last, the directory itself. A tool that left the group and kept nothing the worker gave it + // is still standing where it writes, which is the one thing it cannot put down and go on writing. + const standing = live.holding[0] + return standing ? `pid ${standing.pid} is in the directory: ${standing.how}` : undefined +} + +const writersDir = (data: string, directory: string) => path.join(data, "worktree-writers", Hash.fast(directory)) + +const writerFile = (data: string, directory: string, callID: string) => + path.join(writersDir(data, directory), `${Hash.fast(callID)}.json`) + +/** Say a call is about to write this directory. `endWrite` says its body came back. */ +export const beginWrite = (data: string, directory: string, writer: Writer) => + Effect.promise(async () => { + const file = writerFile(data, directory, writer.callID) + await mkdir(path.dirname(file), { recursive: true }).catch(() => {}) + const note: WriterNote = { + ...writer, + pid: process.pid, + ...((await group()) === undefined ? {} : { pgid: await group() }), + ...((await cgroup()) === undefined ? {} : { cgroup: await cgroup() }), + worker: worker(), + bootAt: bootAt(), + host: os.hostname(), + started: new Date().toISOString(), + } + await writeFile(file, JSON.stringify(note)).catch(() => {}) + }) + +/** Whatever it did to the directory, it is not still doing it. */ +export const endWrite = (data: string, directory: string, callID: string) => + Effect.promise(() => rm(writerFile(data, directory, callID), { force: true }).catch(() => {})) + +/** + * The calls of some other step that never came back. A step's own tools run at once on one host by + * design, so their markers are not a reason to refuse; a marker from another step is the case the + * workflow cannot see. + */ +export const strandedWriters = (data: string, directory: string, current?: Writer) => + Effect.promise(async () => { + const dir = writersDir(data, directory) + const names = await readdir(dir).catch(() => [] as string[]) + if (names.length === 0) return [] + const live = await liveHere(directory) + const found: (WriterNote & { because: string })[] = [] + for (const name of names) { + const file = path.join(dir, name) + const text = await readFile(file, "utf8").catch(() => undefined) + if (text === undefined) continue + let note: WriterNote + try { + note = JSON.parse(text) as WriterNote + } catch { + // A note nothing can parse is still a note somebody wrote before running a tool. + found.push({ + sessionID: "unknown", + step: -1, + callID: name, + pid: -1, + started: "unknown", + because: "nothing can read what it says", + }) + continue + } + // A marker that cannot be a live tool any more is dropped rather than reported: the refusal + // exists because nothing could prove the tool stopped, so where something can, it stops + // standing. Its own step's siblings are not a refusal either way. + const because = insideBecause(note, live) + if (because === undefined) { + await rm(file, { force: true }).catch(() => {}) + continue + } + if (!current || note.sessionID !== current.sessionID || note.step !== current.step) + found.push({ ...note, because }) + } + return found + }) + +/** Forget them, for an operator who has stopped whatever was left running. */ +export const clearWriters = (data: string, directory: string) => + Effect.promise(async () => { + const dir = writersDir(data, directory) + const names = await readdir(dir).catch(() => [] as string[]) + await rm(dir, { recursive: true, force: true }).catch(() => {}) + return names.length + }) diff --git a/packages/core/test/worktree-materialize.test.ts b/packages/core/test/worktree-materialize.test.ts index 7befc9a2c946..5fc2f6394ec0 100644 --- a/packages/core/test/worktree-materialize.test.ts +++ b/packages/core/test/worktree-materialize.test.ts @@ -4,8 +4,12 @@ // simulate a fresh host, "host B" materializes it back from the store alone. import { describe, expect } from "bun:test" import { $ } from "bun" +import { execFile, spawn } from "node:child_process" +import { randomBytes } from "node:crypto" +import { promisify } from "node:util" import { realpathSync } from "node:fs" import { mkdir, readdir, readFile, rm, writeFile } from "node:fs/promises" +import os from "node:os" import path from "path" import { asc } from "drizzle-orm" import { Effect, Fiber, Layer } from "effect" @@ -22,6 +26,7 @@ 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 * as Writers from "@opencode-ai/core/snapshot/writers" import { testEffect } from "./lib/effect" import { tmpdir } from "./fixture/tmpdir" @@ -450,6 +455,204 @@ describe("WorktreeMaterializer", () => { ) }) +// A process in a group of its own, which is what a worker somebody's supervisor started has, and +// what a tool that daemonizes gives itself. It carries a worker name of this test's choosing, so +// what each case turns on is the one thing that case is about. Without one it gets this process's +// environment, which is the case that says the name is exported rather than only written down. +const run = promisify(execFile) + +// Per run, because a name is what the host looks for and an assertion that fails before its child +// is killed leaves that child running. A fixed name would then answer for every later run. +const runToken = randomBytes(4).toString("hex") +const named = (worker: string) => `${worker}-${runToken}` + +const spawned = (worker?: string) => { + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { + detached: true, + stdio: "ignore", + env: worker === undefined ? process.env : { ...process.env, OPENCODE_WORKTREE_WRITER: named(worker) }, + }) + return { child, pid: child.pid! } +} + +const ended = async (started: ReturnType) => { + const exited = new Promise((resolve) => started.child.once("exit", resolve)) + started.child.kill("SIGKILL") + await exited + return { pid: started.pid, pgid: started.pid } +} + +/** Say the one marker on this host was written by another process, or before a restart. */ +const editMarker = async (data: string, worktree: string, patch: Record) => { + const root = path.join(data, "worktree-writers") + const [dir] = await readdir(root) + const [name] = await readdir(path.join(root, dir)) + const file = path.join(root, dir, name) + const note = JSON.parse(await readFile(file, "utf8")) + await writeFile(file, JSON.stringify({ ...note, ...patch })) +} + +describe("WorktreeMaterializer quarantine", () => { + // Refusing to move a step off a host protects that step and nothing after it: the turn ends, the + // next prompt lands wherever there is room, and the tool from before can still be writing. A + // marker says which calls are inside their own execution, and the directory belongs to that + // call's step until it returns. + it.live("refuses a directory to another step while an earlier call has not returned", () => + 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-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, 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)) + + const B = yield* Layer.build(materializeStack(file, data)) + const worktrees = yield* WorktreeMaterializer.Service.pipe(Effect.provide(B)) + + // A call of an earlier step that never came back. + const stranded = { sessionID: "ses_one", step: 1, callID: "call_stranded" } + yield* worktrees.beginWrite(worktree, stranded) + + // Another step wants the directory. It is not this call's step, so it is refused, and the + // refusal is a defect the activity boundary turns into a failure Temporal schedules again. + const later = { sessionID: "ses_one", step: 2, callID: "call_later" } + const refused = yield* Effect.exit(worktrees.ensure(worktree, { current: later })) + expect(refused._tag).toBe("Failure") + + // A sibling of the same step is not stranded: two tools of one step share this directory by + // design, and refusing them would be refusing the feature. + const sibling = { sessionID: "ses_one", step: 1, callID: "call_sibling" } + const allowed = yield* Effect.exit(worktrees.ensure(worktree, { current: sibling })) + expect(allowed._tag).toBe("Success") + + // When the call comes back, whatever it did to the directory, it is not still doing it. + yield* worktrees.endWrite(worktree, stranded.callID) + const afterReturn = yield* Effect.exit(worktrees.ensure(worktree, { current: later })) + expect(afterReturn._tag).toBe("Success") + + // A worker that died mid-tool leaves its marker behind. Most of that is answerable without a + // person: the writer's process is gone, nothing it started is left in its group, and the + // machine has not restarted underneath the pids that say so. + const usable = () => + Effect.exit(worktrees.ensure(worktree, { current: later })).pipe(Effect.map((exit) => exit._tag === "Success")) + + // Written by this process, which is running. Nothing to conclude, so the refusal stands. + yield* worktrees.beginWrite(worktree, { sessionID: "ses_one", step: 3, callID: "call_dead" }) + expect(yield* usable()).toBe(false) + + // The worker died and left nothing behind: no process of its own, nothing carrying its name, + // and an empty group. That is the whole of the proof that its tools are over. + const gone = yield* Effect.promise(() => ended(spawned("gone"))) + const stale = { pid: gone.pid, pgid: gone.pgid, worker: named("gone") } + yield* Effect.promise(() => editMarker(data, worktree, stale)) + expect(yield* usable()).toBe(true) + + // The worker died and something it started did not. That is what the refusal is for. + const orphan = spawned("orphaned") + yield* worktrees.beginWrite(worktree, { sessionID: "ses_one", step: 3, callID: "call_dead" }) + yield* Effect.promise(() => editMarker(data, worktree, { ...stale, pgid: orphan.pid, worker: named("orphaned") })) + expect(yield* usable()).toBe(false) + yield* Effect.promise(() => ended(orphan)) + expect(yield* usable()).toBe(true) + + // A tool that asks for a group of its own is out of the group check's reach. What it cannot + // put down is the name its worker left in the environment it inherited. + const escaped = spawned("escaped") + yield* worktrees.beginWrite(worktree, { sessionID: "ses_one", step: 3, callID: "call_dead" }) + yield* Effect.promise(() => editMarker(data, worktree, { ...stale, worker: named("escaped") })) + expect(yield* usable()).toBe(false) + yield* Effect.promise(() => ended(escaped)) + expect(yield* usable()).toBe(true) + + // A tool that leaves the group and is handed an environment of somebody else's choosing has + // put down everything the worker gave it. What it cannot put down and go on writing is the + // directory it writes, so that is the last reading, and the one that depends on nothing the + // tool kept. + const standing = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { + detached: true, + stdio: "ignore", + cwd: worktree, + env: { PATH: process.env.PATH ?? "" }, + }) + yield* worktrees.beginWrite(worktree, { sessionID: "ses_one", step: 3, callID: "call_dead" }) + yield* Effect.promise(() => editMarker(data, worktree, { ...stale, worker: named("nothing-carries-this") })) + expect(yield* usable()).toBe(false) + yield* Effect.promise(() => ended({ child: standing, pid: standing.pid! })) + expect(yield* usable()).toBe(true) + + // The control group is Linux's, and a tool keeps it through `setsid` and through `sudo`, + // which empties the environment it passes on. A Mac has no such thing to read, so the rule is + // asked on its own, with the reading a Linux host would have made handed to it. + const readings = { + groups: new Set(), + workers: new Set(), + holding: [], + ourCgroup: "/the-one-this-process-is-in", + } + const dead = { + sessionID: "ses_one", + step: 3, + callID: "call_dead", + host: os.hostname(), + pid: gone.pid, + // The stamp the module compares against, not the wall clock: a marker dated now would read + // as one from before a restart and never reach the rule this is about. + bootAt: Math.round(Date.now() - os.uptime() * 1000), + started: new Date().toISOString(), + } + const inCgroup = (cgroup: string, live: string[]) => + Writers.insideBecause({ ...dead, cgroup }, { ...readings, cgroups: new Set(live) }) + expect(inCgroup("/system.slice/opencode-1.service", ["/system.slice/opencode-1.service"])).toContain( + "still in /system.slice/opencode-1.service", + ) + expect(inCgroup("/system.slice/opencode-1.service", ["/system.slice/opencode-2.service"])).toBeUndefined() + // And the one this process is in says nothing either way, which is every process in a + // container. + expect(inCgroup(readings.ourCgroup, [readings.ourCgroup])).toBeUndefined() + + // And the name has to reach the tool, not only the marker. This child is given no environment + // of its own, so the only way it carries the name is that the worker exported it. + const inheriting = spawned() + yield* worktrees.beginWrite(worktree, { sessionID: "ses_one", step: 3, callID: "call_dead" }) + yield* Effect.promise(() => + editMarker(data, worktree, { ...stale, worker: process.env.OPENCODE_WORKTREE_WRITER }), + ) + expect(yield* usable()).toBe(false) + yield* Effect.promise(() => ended(inheriting)) + expect(yield* usable()).toBe(true) + + // A worker restarted from the same shell is in the group its predecessor was in, so the group + // answers for this process rather than for the marker. What the dead worker started is what + // decides, and it started nothing. + const ourGroup = yield* Effect.promise(async () => { + const { stdout } = await run("ps", ["-o", "pgid=", "-p", String(process.pid)]) + return Number.parseInt(stdout.trim(), 10) + }) + yield* worktrees.beginWrite(worktree, { sessionID: "ses_one", step: 3, callID: "call_dead" }) + yield* Effect.promise(() => editMarker(data, worktree, { ...stale, pgid: ourGroup })) + expect(yield* usable()).toBe(true) + + // A pid means nothing across a restart, so a marker from before one is not read as live. + yield* worktrees.beginWrite(worktree, { sessionID: "ses_one", step: 3, callID: "call_dead" }) + yield* Effect.promise(() => editMarker(data, worktree, { bootAt: 0 })) + expect(yield* usable()).toBe(true) + }), + ) +}) + // A dispatch the session stopped waiting for keeps running, and the files it ships afterwards would // be the newest state the store holds: the tip check cannot refuse them, because the host is still // standing exactly where it was told to stand. The event log already fences a superseded attempt @@ -504,3 +707,73 @@ describe("SnapshotSync owner fence", () => { }), ) }) + +// Four readings narrow what a tool can put down to be found by, and none of them closes it. Moving +// the directory does: the writer nobody can account for goes on writing the one it has open, under +// its new name, and the host builds a fresh one here. +describe("WorktreeMaterializer set-aside", () => { + it.live("moves a directory it built rather than refusing it, and never one it adopted", () => + 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-a-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, 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)) + + // Host B, which has never seen this directory: it builds the tree here out of nothing, and + // that is what makes the directory its own to move. + const dataB = path.join(root, "host-b-data") + const onB = path.join(root, "project") + const B = yield* Layer.build(materializeStack(file, dataB)) + const worktrees = yield* WorktreeMaterializer.Service.pipe(Effect.provide(B)) + yield* Effect.promise(() => rm(onB, { recursive: true, force: true })) + yield* worktrees.ensure(onB) + expect(yield* Effect.promise(() => readdir(onB))).toContain("tracked.txt") + + // A call that never came back, and a file it is still writing. + yield* worktrees.beginWrite(onB, { sessionID: "ses_one", step: 1, callID: "call_stranded" }) + yield* Effect.promise(() => writeFile(path.join(onB, "still-writing.txt"), "from the tool\n")) + + const later = { sessionID: "ses_one", step: 2, callID: "call_later" } + const moved = yield* Effect.exit(worktrees.ensure(onB, { current: later })) + expect(moved._tag).toBe("Success") + const asideNames = (yield* Effect.promise(() => readdir(root))).filter((n) => n.startsWith("project.stranded.")) + expect(asideNames).toHaveLength(1) + // What the tool wrote went with the directory it was writing, and what the session shipped is + // in the fresh one. + expect(yield* Effect.promise(() => readdir(path.join(root, asideNames[0]!)))).toContain("still-writing.txt") + const now = yield* Effect.promise(() => readdir(onB)) + expect(now).toContain("tracked.txt") + expect(now).not.toContain("still-writing.txt") + + // And a directory this host did not build is never moved: what it holds is somebody's, and + // includes files no pack carries. + const adopted = path.join(root, "adopted") + yield* Effect.promise(async () => { + await mkdir(adopted, { recursive: true }) + await writeFile(path.join(adopted, "theirs.txt"), "not ours\n") + }) + yield* worktrees.beginWrite(adopted, { sessionID: "ses_one", step: 1, callID: "call_theirs" }) + const refused = yield* Effect.exit(worktrees.ensure(adopted, { current: later })) + // Nothing is stored for that directory, so `ensure` returns before the refusal can apply; what + // is asserted is that it is still there, whole. + expect(refused._tag).toBe("Success") + expect(yield* Effect.promise(() => readdir(root))).toContain("adopted") + expect(yield* Effect.promise(() => readdir(adopted))).toContain("theirs.txt") + }), + ) +}) diff --git a/packages/temporal/README.md b/packages/temporal/README.md index 5db6e53a6323..f6756948ceb7 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -260,10 +260,14 @@ tool-output files under the data directory. The runner never reads these to rebu only affect the diff, restore and full-output features. Point the data directory at shared storage to make them portable. -One rule bounds what the rebuild 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 alone. Packs are ordered -by their chain, not by `time_created`, because hosts do not agree on the time. +Two rules bound what the rebuild 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 alone. And a tree is moved +only when this host built it from packs, so a developer's own checkout is logged and never +rewritten. Packs are ordered by their chain, not by `time_created`, because hosts do not agree on +the time. While a tool call runs, the host records it beside the directory, and a later step is +refused that directory until the call is over; a directory whose writer cannot be accounted for is +moved aside and rebuilt fresh. ## A session that outlives its client @@ -339,6 +343,7 @@ TEMPORAL_ADDRESS=temporal.internal:7233 \ | Restart during an approval or question | The pending row is kept; execution has to be restarted | A retried ask adopts the row and can be answered from another process | | A behind host publishes files | Not reachable | Refused: the pack store checks the chain and the owner token | | A fresh worker receives a project | Needs the directory | Packs rebuild the tracked files at the recorded path; ignored files do not travel | +| A later turn reuses a directory an abandoned tool may still write | Not reachable | The directory is refused until the call can be shown to be over, then moved aside and rebuilt | | A turn never stops stepping | The coordinator's own loop | The supervisor stops after 200 steps and says so | An unknown tool outcome is a loss of evidence, not proof that the tool stopped or failed. The model diff --git a/packages/temporal/src/boundary.ts b/packages/temporal/src/boundary.ts index e413da286394..47ac00194fc8 100644 --- a/packages/temporal/src/boundary.ts +++ b/packages/temporal/src/boundary.ts @@ -41,6 +41,13 @@ const halted = (sessionID: string, declined?: SessionRunDeclinedError) => { // non-retryable, because re-running a step whose input the model already answered is worse than // failing it. A libsql blip during a seal would otherwise fail the step for good rather than move // it to another worker. +const QUARANTINED = "WorktreeMaterializer.QuarantinedError" +// A refusal must not climb the backoff a failing activity earns: the interval doubles per attempt, +// and the host that answers first and refuses fastest is exactly the one that would push the next +// attempt minutes out while a free host sits idle. Long enough not to spin, short enough that the +// work reaches another host in about the time one dispatch takes. +const REFUSAL_RETRY = "2 seconds" + const TRANSIENT = new Set([ "ToolOutputStore.StorageError", "SqlError", @@ -48,6 +55,9 @@ const TRANSIENT = new Set([ // 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", + // And a directory this host is refused. It is this host saying no, not the work failing: the + // same dispatch runs fine on a host that is not holding somebody's abandoned tool. + QUARANTINED, ]) export const runAtBoundary = async ( @@ -78,5 +88,6 @@ export const runAtBoundary = async ( type: squashed?._tag ?? "SessionRunError", nonRetryable: !(squashed?._tag !== undefined && TRANSIENT.has(squashed._tag)), details: encoded === undefined ? undefined : [encoded], + ...(squashed?._tag === QUARANTINED ? { nextRetryDelay: REFUSAL_RETRY } : {}), }) } diff --git a/packages/temporal/src/drain.ts b/packages/temporal/src/drain.ts index 60cb438e55b2..6597b6b106fd 100644 --- a/packages/temporal/src/drain.ts +++ b/packages/temporal/src/drain.ts @@ -57,6 +57,8 @@ export interface InSessionOptions { /** Take the event log for this attempt. True for the writer that supersedes the attempt before it; * the other units of a stepped turn ride that writer's token. */ readonly claim?: boolean + /** The call about to run, so the tree check can tell this step's writers from an earlier step's. */ + readonly current?: { readonly sessionID: string; readonly step: number; readonly callID: string } /** Leave the project tree alone. For a seal closing a step away from its host: rebuilding here * would put this host on the newest state while the one that ran the tools may still be writing, * and nothing the seal does needs the files. */ @@ -81,8 +83,10 @@ export const makeInSession = if (!session) return undefined // Take the event log before running so a superseded attempt's later appends are fenced. if (options.claim && owner) yield* events.claim(session.id, owner) - // A worker taking this step on a host without the project tree rebuilds it from snapshot packs. - if (!options.withoutTheTree) yield* worktrees.ensure(session.location.directory) + // A worker taking this step on a host without the project tree rebuilds it from snapshot + // packs, unless a call of an earlier step never came back on this host. + if (!options.withoutTheTree) + yield* worktrees.ensure(session.location.directory, options.current ? { current: options.current } : undefined) return yield* SessionRunner.Service.use((runner) => use(runner, session)).pipe( Effect.provide(locations.get(session.location)), ) diff --git a/packages/temporal/src/executor.ts b/packages/temporal/src/executor.ts index acc7998de2e8..d96f9cf20e1c 100644 --- a/packages/temporal/src/executor.ts +++ b/packages/temporal/src/executor.ts @@ -115,7 +115,7 @@ const layer = Layer.effect( const drains = makeDrains({ store, locations, ctx, events, worktrees }) // The stepped mode's three drains. Registered unconditionally: which mode a session runs is a // property of its workflow input, so a worker has to be able to serve either. - const stepped = makeSteppedDrains({ inSession: drains.inSession, stepQueue: STEP_QUEUE }) + const stepped = makeSteppedDrains({ inSession: drains.inSession, 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 }) diff --git a/packages/temporal/src/stepped-drain.ts b/packages/temporal/src/stepped-drain.ts index 6faacc08e53d..eac9084d7f2c 100644 --- a/packages/temporal/src/stepped-drain.ts +++ b/packages/temporal/src/stepped-drain.ts @@ -11,6 +11,7 @@ import { Effect } from "effect" import type { DeferredToolCall, ToolCallOutcome } from "@opencode-ai/core/session/runner" import type { StepSettlement } from "@opencode-ai/core/session/runner/publish-llm-event" import type { SessionInput } from "@opencode-ai/core/session/input" +import { WorktreeMaterializer } from "@opencode-ai/core/session/execution/worktree" import { runAtBoundary } from "./boundary" import { type InSession, type StepDrainInput, type StepDrainResult, toStepResult } from "./drain" @@ -41,6 +42,8 @@ export interface ToolCallDrainInput { readonly sessionID: string readonly call: DeferredToolCall readonly owner: string + /** Which step this call belongs to, so a host can tell it from a call an earlier step left. */ + readonly step: number } export interface ToolCallDrainResult { @@ -62,12 +65,21 @@ export interface SealDrainInput { export interface SteppedDrainDeps { readonly inSession: InSession + /** Records which call is inside its own execution on this host, and takes the record back. */ + readonly worktrees: Pick /** 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 makeSteppedDrains = ({ inSession, stepQueue }: SteppedDrainDeps) => { +// A call, as the host records it: enough to tell this step's writers from an earlier step's. +const writer = (input: ToolCallDrainInput) => ({ + sessionID: input.sessionID, + step: input.step, + callID: input.call.id, +}) + +export const makeSteppedDrains = ({ inSession, worktrees, stepQueue }: SteppedDrainDeps) => { // Only the model call claims the log: it is the writer that supersedes a previous attempt, and // the rest of the step rides its token. const modelCallDrain = async ( @@ -106,16 +118,26 @@ export const makeSteppedDrains = ({ inSession, stepQueue }: SteppedDrainDeps) => runAtBoundary( input.sessionID, signal, - inSession(input.sessionID, input.owner, {}, (runner, session) => - runner.runToolCall({ sessionID: session.id, call: input.call }).pipe( - // A stop landing mid-tool leaves the call recorded as running, where a whole step closes - // the tools it opened before it returns. Nothing else closes it until the next turn's - // entry check, so a transcript would show the call still going long after the stop. - Effect.onInterrupt(() => - turnEnded() - ? runner.failToolCall({ sessionID: session.id, call: input.call }).pipe(Effect.ignore) - : Effect.void, - ), + inSession(input.sessionID, input.owner, { current: writer(input) }, (runner, session) => + Effect.acquireUseRelease( + // Said before the tool can touch anything, and taken back when its body returns. It is + // the only record on this host of a call still inside its own execution, and what a + // later step reads before it uses this directory: a timeout settles the workflow's + // promise without stopping the process behind it. + worktrees.beginWrite(session.location.directory, writer(input)), + () => + runner.runToolCall({ sessionID: session.id, call: input.call }).pipe( + // A stop landing mid-tool leaves the call recorded as running, where a whole step + // closes the tools it opened before it returns. Nothing else closes it until the + // next turn's entry check, so a transcript would show the call still going long + // after the stop. + Effect.onInterrupt(() => + turnEnded() + ? runner.failToolCall({ sessionID: session.id, call: input.call }).pipe(Effect.ignore) + : Effect.void, + ), + ), + () => worktrees.endWrite(session.location.directory, input.call.id), ), ).pipe(Effect.map((result) => result ?? { outcome: "already-settled" as const })), ) diff --git a/packages/temporal/src/stepped-turn.ts b/packages/temporal/src/stepped-turn.ts index 1ff26740928b..8f57ae0dfa6b 100644 --- a/packages/temporal/src/stepped-turn.ts +++ b/packages/temporal/src/stepped-turn.ts @@ -154,7 +154,11 @@ export const makeSteppedTurn = // the seal closes its call as an error and the model gets to react, which is better than losing // the step. A cancel and a user halt are different, and both have to propagate. const dispatch = (call: (typeof model.calls)[number]) => - viaPinned((on) => on.runToolCall({ sessionID: input.sessionID, call, owner: model.owner })) + viaPinned((on) => + // The step travels with the call, because the host tells this step's writers from an + // earlier step's by it. + on.runToolCall({ sessionID: input.sessionID, call, owner: model.owner, step: model.step }), + ) const dispatched: PromiseSettledResult[] = [] if (serial) { // One at a time, and still settled rather than thrown, so a tool that fails does not take the diff --git a/packages/temporal/test/boundary-refusal.test.ts b/packages/temporal/test/boundary-refusal.test.ts new file mode 100644 index 000000000000..99c524a26c17 --- /dev/null +++ b/packages/temporal/test/boundary-refusal.test.ts @@ -0,0 +1,42 @@ +// What a refused directory costs the session is decided at the activity boundary, not in the +// materializer that raises it. +// +// Two things have to be true of it. It has to be retryable, or a host holding somebody's abandoned +// tool ends the turn for every host: the refusal is this host saying no, and the same dispatch runs +// fine on a host that is not holding one. And it has to carry its own retry delay, because the +// backoff a failing activity earns doubles per attempt, and the host that answers first and refuses +// fastest is exactly the one that would push the next attempt minutes out while a free host idles. + +import { expect, it } from "bun:test" +import { Effect } from "effect" +import { ApplicationFailure } from "@temporalio/common" +import { WorktreeMaterializer } from "@opencode-ai/core/session/execution/worktree" +import { runAtBoundary } from "../src/boundary" + +const crossing = (body: Effect.Effect) => + runAtBoundary("ses_refused", new AbortController().signal, body).then( + () => undefined, + (err: unknown) => err as ApplicationFailure, + ) + +it("hands a refused directory back as retryable work with its own delay", async () => { + const refused = await crossing( + Effect.die( + new WorktreeMaterializer.WorktreeQuarantinedError({ + message: "not using /project: a call of an earlier step never returned", + }), + ), + ) + expect(refused).toBeInstanceOf(ApplicationFailure) + expect(refused?.type).toBe("WorktreeMaterializer.QuarantinedError") + expect(refused?.nonRetryable).toBeFalsy() + expect(refused?.nextRetryDelay).toBeDefined() +}) + +it("leaves an ordinary run error non-retryable and on the backoff", async () => { + // The contrast that makes the above mean something: re-running a step whose input the model has + // already answered is worse than failing it, so everything that is about the work stays as it was. + const failed = await crossing(Effect.die(new Error("the tool blew up"))) + expect(failed?.nonRetryable).toBe(true) + expect(failed?.nextRetryDelay).toBeUndefined() +}) diff --git a/packages/temporal/test/stepped-drain-writers.test.ts b/packages/temporal/test/stepped-drain-writers.test.ts index 6359d7bdddfb..e78978401ae6 100644 --- a/packages/temporal/test/stepped-drain-writers.test.ts +++ b/packages/temporal/test/stepped-drain-writers.test.ts @@ -1,8 +1,10 @@ -// The seal is the one unit of a step that can run away from the host that ran the step, and when it -// does it has to leave the tree alone: rebuilding here would put this host on the newest state while -// the one that ran the tools may still be writing, and what it captured would be the state before -// them. This drives the real drain, because the option is threaded through it and nothing else -// checks that it reaches the materializer. +// The marker a tool call writes is the only record on a host of a call still inside its own +// execution, and every refusal the worktree materializer makes is built on it being there. The store +// side has its own tests; this one drives the real drain, because the three lines that bracket the +// tool body are what put a marker on the host at all. +// +// Take the bracket out of `toolCallDrain` and the first assertion fails: the tool runs with nothing +// on the host saying it is inside, and a later step is free to take the directory. import { expect } from "bun:test" import { mkdtemp } from "node:fs/promises" @@ -14,6 +16,7 @@ import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { Global } from "@opencode-ai/core/global" import { WorktreeMaterializer } from "@opencode-ai/core/session/execution/worktree" import { SessionRunner } from "@opencode-ai/core/session/runner" +import * as Writers from "@opencode-ai/core/snapshot/writers" import { testEffect } from "../../core/test/lib/effect" import { type DrainDeps, makeInSession } from "../src/drain" import { makeSteppedDrains } from "../src/stepped-drain" @@ -26,7 +29,7 @@ const materializerStack = (data: string) => [Global.node, Layer.succeed(Global.Service, Global.make({ data }))], ]) -/** A drain over a runner that reports what it was asked to do. */ +/** A drain over a runner that reports what the host said about it while it was running. */ const drainsOver = ( worktrees: WorktreeMaterializer.Interface, directory: string, @@ -37,8 +40,8 @@ const drainsOver = ( store: { get: () => Effect.succeed({ id: "ses_writers", location: { directory } }), } as unknown as DrainDeps["store"], - // The location resolves to a runner and nothing else: what is under test is what surrounds - // the runner call, not what the runner does inside it. + // The location resolves to a runner and nothing else: what is under test is the bracket + // around the call, not what the tool does inside it. locations: { get: () => Layer.succeed(SessionRunner.Service, runner as SessionRunner.Interface), } as unknown as DrainDeps["locations"], @@ -46,8 +49,86 @@ const drainsOver = ( events: { claim: () => Effect.void } as unknown as DrainDeps["events"], worktrees, }), + worktrees, }) +it.live("marks the directory while a tool call runs and takes the mark back after it", () => + Effect.gen(function* () { + const root = yield* Effect.promise(() => mkdtemp(path.join(tmpdir(), "opencode-l2-writers-"))) + const data = path.join(root, "host-data") + const directory = path.join(root, "project") + const worktrees = yield* WorktreeMaterializer.Service.pipe( + Effect.provide(yield* Layer.build(materializerStack(data))), + ) + + const seen: Writers.Writer[][] = [] + const { toolCallDrain } = drainsOver(worktrees, directory, { + runToolCall: ((input: { call: { id: string } }) => + Effect.gen(function* () { + // Inside the body, which is the window the refusal exists for. Asked with no step of its + // own, so what comes back is every marker the host is holding. + seen.push(yield* Writers.strandedWriters(data, directory)) + return { outcome: "settled" as const, call: input.call.id } + })) as unknown as SessionRunner.Interface["runToolCall"], + }) + + yield* Effect.promise(() => + toolCallDrain( + { + sessionID: "ses_writers", + call: { id: "call_one", name: "write", assistantMessageID: "msg_one" }, + owner: "run:1:1", + step: 3, + }, + new AbortController().signal, + ), + ) + + expect(seen).toHaveLength(1) + expect(seen[0].map((w) => ({ session: w.sessionID, step: w.step, call: w.callID }))).toEqual([ + { session: "ses_writers", step: 3, call: "call_one" }, + ]) + // And the mark is gone once the body returned, or every later step would be refused a directory + // nothing is writing. + expect(yield* Writers.strandedWriters(data, directory)).toEqual([]) + }), +) + +it.live("takes the mark back when the tool fails rather than returns", () => + Effect.gen(function* () { + const root = yield* Effect.promise(() => mkdtemp(path.join(tmpdir(), "opencode-l2-writers-"))) + const data = path.join(root, "host-data") + const directory = path.join(root, "project") + const worktrees = yield* WorktreeMaterializer.Service.pipe( + Effect.provide(yield* Layer.build(materializerStack(data))), + ) + + const { toolCallDrain } = drainsOver(worktrees, directory, { + runToolCall: (() => + Effect.die(new Error("the tool blew up"))) as unknown as SessionRunner.Interface["runToolCall"], + }) + + const failed = yield* Effect.promise(() => + toolCallDrain( + { + sessionID: "ses_writers", + call: { id: "call_two", name: "write", assistantMessageID: "msg_two" }, + owner: "run:1:1", + step: 4, + }, + new AbortController().signal, + ).then( + () => undefined, + (err: unknown) => err, + ), + ) + expect(String(failed)).toContain("the tool blew up") + // A call that failed is not a call still inside its own execution. Leaving the mark would refuse + // the directory to everything after it for a tool that is over. + expect(yield* Writers.strandedWriters(data, directory)).toEqual([]) + }), +) + it.live("seals a step away from its host without touching the tree", () => Effect.gen(function* () { const root = yield* Effect.promise(() => mkdtemp(path.join(tmpdir(), "opencode-l2-writers-")))