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-")))