From 8ce22602c3619dfc1226ca73b3ce4a7ffdb558a8 Mon Sep 17 00:00:00 2001 From: Jalil Date: Fri, 13 Nov 2026 17:40:11 -0800 Subject: [PATCH] fix(workbot): keep a computer command's result when its wait is cut A command ran inside the request that waited for it. When that request was cut (a runner deploy cancels a busy cell, a node is lost, or the provider's request limit is reached) nobody knew how the command had ended: bash answered "No confirmed result", and a step restarted after the cut recorded the call as interrupted, so the model had to guess whether its side effects had happened. bash now starts each command once on the computer as a detached job named after the call. The job writes its output and exit code to files under /workspace/.jobs and stops itself at the command's time limit; the request only waits for it. A wait that is cut is taken up again for what is left of the command's time, and making the job's output file is what claims a command, so it can never start twice. A step that restarts with a kept answer asks the computer for each command without a recorded result before giving up on it. A command that ended returns its real exit code and output, with its outputs added to the person's Files; one still running is waited for; one the computer never started stays an interrupted call. Calls are keyed by turn, run, step and call id, because a model may reuse its call ids. The idle pause still sees such a command as a running job. A job's .running file now names its process, so a file left behind by a command that was killed no longer keeps a computer from pausing. A command that reaches its time limit is stopped and reports what it printed so far. It runs through the existing command path from the runner, so no computer image or guest changes. Preview rendering now logs when it starts and when it ends (ids, outcome and duration), to show how often a render is lost with its runner. Signed-off-by: Jalil --- ee/apps/headless-runner/scripts/steps.test.ts | 57 +++- ee/apps/headless-runner/src/computer-route.ts | 4 +- ee/apps/headless-runner/src/runner.ts | 29 +- ee/apps/headless-runner/src/steps.ts | 3 +- ee/apps/headless-runner/src/types.ts | 8 +- .../src/computer-commands.test.ts | 177 +++++++++++ ee/packages/headless-computer/src/computer.ts | 280 ++++++++++++++++-- 7 files changed, 524 insertions(+), 34 deletions(-) create mode 100644 ee/packages/headless-computer/src/computer-commands.test.ts diff --git a/ee/apps/headless-runner/scripts/steps.test.ts b/ee/apps/headless-runner/scripts/steps.test.ts index a9b707b8df..8c2ffd509a 100644 --- a/ee/apps/headless-runner/scripts/steps.test.ts +++ b/ee/apps/headless-runner/scripts/steps.test.ts @@ -19,7 +19,7 @@ import { APPS_UNREACHABLE, CUT_OFF_INSTRUCTION, Runner, type RunnerOptions } fro import { SelfResume } from "../src/self-resume.js" import { FRESH_REPEAT, type StepState } from "../src/steps.js" import { Store } from "../src/store.js" -import type { Message } from "../src/types.js" +import type { Message, SessionComputer, ToolResult } from "../src/types.js" const usage = { inputTokens: 3, cachedInputTokens: 1, outputTokens: 2 } const unlimited: RunnerOptions["limits"] = { maxConcurrentTurns: 4, maxSteps: 20, turnTimeoutMs: Number.POSITIVE_INFINITY, credentialRefreshMs: Number.POSITIVE_INFINITY, contextCharBudget: 100_000 } @@ -63,6 +63,7 @@ type WorldInput = { errors?: ErrorReporter limits?: Partial driver?: (db: SqlDriver) => SqlDriver + computer?: SessionComputer } /** @@ -89,6 +90,7 @@ function world(directory: string, input: WorldInput) { errors: input.errors, now: input.clock, stepAlarm: input.alarm, + computer: input.computer, }) const steps = runner.recoverSteps() return { store, runner, recovered, steps } @@ -323,6 +325,59 @@ test("a tool call whose result wasn't recorded is never run again: the model get assert.deepEqual(tokens, ["caller-token", "fresh-token-1"]) }) +test("a computer command cut off with its step is picked up from the computer, never run again; one it never started stays interrupted", async () => { + using temp = tempDir() + const requests: ModelRequest[] = [] + const model: ModelClient = { + async complete(request) { + requests.push(request) + if (resultsOf(request).length === 0) { + return { text: "Building it.", toolCalls: [{ id: "call_build", name: "bash", input: { command: "make report" } }, { id: "call_send", name: "bash", input: { command: "send report" } }], usage } + } + return { text: "The report is built; I couldn't confirm it was sent.", toolCalls: [], usage } + }, + } + // The conversation's computer, which outlives the runner: it keeps what it started, by the call's key. + const started: Array<{ command: unknown; key: string | undefined }> = [] + const asked: string[] = [] + const computer = (results: Record): SessionComputer => ({ + prompt: "", + tools: [{ name: "bash", description: "Runs a command", inputSchema: { type: "object" } }], + toolNames: new Set(["bash"]), + image: "test", + async run(_sessionId, _name, input, callKey) { + started.push({ command: input.command, key: callKey }) + return never() + }, + async pickUp(_sessionId, _name, input, callKey) { + asked.push(callKey) + return results[String(input.command)] ?? null + }, + known: () => false, + prewarm: () => undefined, + release: () => undefined, + delete: async () => undefined, + previewSavedFile: async () => undefined, + }) + const first = world(temp.directory, { model, computer: computer({}) }) + const session = first.store.createSession({ computer: true }) + first.runner.send({ sessionId: session.id, messageId: "m1", prompt: "Build and send the report", credentials: { modelApiKey: "caller-key" }, execution: "steps" }) + await until(() => started.length === 1, "the build starting") + // The runner dies while the build runs. On the computer it finishes; the send never started. + const second = world(temp.directory, { model, computer: computer({ "make report": { output: "[exit 0]\nreport.pdf written", isError: false } }) }) + await second.runner.idle() + assert.equal(second.store.getTurn(session.id, "m1")?.status, "completed") + assert.deepEqual(started.map((call) => call.command), ["make report"], "no command ran a second time") + assert.equal(asked.length, 2) + assert.equal(asked[0], started[0]?.key, "the computer is asked by the key the command was started under") + assert.notEqual(asked[1], asked[0]) + const results = transcript(second.store, session.id, "m1").filter((message) => message.role === "tool") + assert.deepEqual(results.map((message) => [message.callId, message.isError]), [["call_build", false], ["call_send", true]]) + assert.equal(results[0]?.output, "[exit 0]\nreport.pdf written") + assert.ok(results[1]?.output.includes("not retried")) + assert.equal(requests.length, 2, "the step finished from the kept answer, without asking the model again") +}) + test("without a resume grant, a turn that lost its credentials is interrupted as on any restart, and resumes when re-sent", async () => { using temp = tempDir() let calls = 0 diff --git a/ee/apps/headless-runner/src/computer-route.ts b/ee/apps/headless-runner/src/computer-route.ts index d54da4b81c..2c84d908fb 100644 --- a/ee/apps/headless-runner/src/computer-route.ts +++ b/ee/apps/headless-runner/src/computer-route.ts @@ -152,7 +152,9 @@ export function routeComputers(input: { desktopViewer: async (sessionId, view) => (await desktop(sessionId, false)).desktopViewer(sessionId, view), browserDemonstration: async (sessionId, command) => (await desktop(sessionId, true)).browserDemonstration(sessionId, command), desktopAction: async (sessionId, action) => (await desktop(sessionId, false)).desktopAction(sessionId, action), - run: async (sessionId, name, args) => (await routed(sessionId, true)).run(sessionId, name, args), + run: async (sessionId, name, args, callKey) => (await routed(sessionId, true)).run(sessionId, name, args, callKey), + // Asked where the conversation's computer is, never placing one: a call no computer started has no result. + pickUp: async (sessionId, name, args, callKey) => (await (await route(sessionId, false))?.pickUp?.(sessionId, name, args, callKey)) ?? null, known: (sessionId) => current(sessionId)?.known(sessionId) ?? false, prewarm: (sessionId) => void route(sessionId, true).then((computer) => computer?.prewarm(sessionId)).catch(() => undefined), // Both sides: a no-op where the conversation has no computer, and one switched away from still pauses. diff --git a/ee/apps/headless-runner/src/runner.ts b/ee/apps/headless-runner/src/runner.ts index c3933a17c8..eb51da901c 100644 --- a/ee/apps/headless-runner/src/runner.ts +++ b/ee/apps/headless-runner/src/runner.ts @@ -540,6 +540,12 @@ function toolMessage(call: ToolCall, outcome: ToolResult): Message { } } +/** + * Names one call of one step for good: the same when the step starts again, never shared with another step or run (a + * model may reuse its call ids). The computer keeps a command's result under it. + */ +const stepCallKey = (messageId: string, state: StepState, call: ToolCall) => `${messageId}/${state.stretchStartedAt}/${state.steps}/${call.id}` + /** A step's kept answer as messages: each call with its recorded result, or `missing` as an error for the rest. */ function pendingOutcomes(pending: PendingStep, missing: string): ToolResult[] { return pending.toolCalls.map((call) => pending.results.find((entry) => entry.callId === call.id)?.result ?? { output: missing, isError: true }) @@ -1109,7 +1115,7 @@ export class Runner { private async runCall( setup: TurnSetup, call: ToolCall, - context: { sessionId: string; kind: JobKind; tools: ToolSession | null; signal: AbortSignal; held: HeldCredentials }, + context: { sessionId: string; kind: JobKind; tools: ToolSession | null; signal: AbortSignal; held: HeldCredentials; callKey?: string }, ): Promise { const { store } = this.options const { readOnly, files, reactions, taskTools, turn, computer, memory } = setup @@ -1139,7 +1145,7 @@ export class Runner { ? desktopViewRefused(store.getSession(sessionId)?.desktopViews ?? null, call.input) ?? await computer.desktopRun(sessionId, call.input, signal, { browserRecovery: store.getSession(sessionId)?.browserRecovery === true }).catch((error: unknown) => ({ output: error instanceof Error ? error.message : "Desktop unavailable", isError: true })) : computer?.toolNames.has(call.name) - ? await computer.run(sessionId, call.name, call.input).catch((error: unknown) => ({ + ? await computer.run(sessionId, call.name, call.input, context.callKey).catch((error: unknown) => ({ output: `The computer didn't respond: ${error instanceof Error ? error.message : "unknown error"}. Its outcome is unknown; check before repeating it.`, isError: true, })) @@ -1845,8 +1851,23 @@ export class Runner { // The step stopped while its tool calls ran: it finishes from the kept answer. A call without a recorded // result is not run again (it may have had side effects); the model gets an error and checks. answer = state.pending + const readOnly = (lane.held?.credentials.readOnly ?? state.readOnly) === true + // Except a command on the conversation's computer, which went on without the runner and kept its result + // there: the computer is asked for it. A command it never started stays an interrupted call. + const computer = session?.computer && !readOnly && kind !== "report" && this.options.computer?.available?.(sessionId) !== false ? this.options.computer : undefined + for (const call of answer.toolCalls) { + const pickUp = computer?.pickUp + if (!pickUp || call.inputError || !computer.toolNames.has(call.name) || answer.results.some((entry) => entry.callId === call.id)) continue + signal.throwIfAborted() + const key = stepCallKey(messageId, state, call) + const found = await lane.timing.tool(call.name, () => pickUp.call(computer, sessionId, call.name, call.input, key).catch(() => null)) + if (!found) continue + answer.results.push({ callId: call.id, result: found }) + state = { ...state, pending: answer } + store.setStepState(sessionId, messageId, state) + } outcomes = pendingOutcomes(answer, INTERRUPTED_CALL) - reactions = session?.reactions === true && kind === "message" && (lane.held?.credentials.readOnly ?? state.readOnly) !== true + reactions = session?.reactions === true && kind === "message" && !readOnly } else { const turnMessages = () => store.turnMessages(sessionId, messageId).map((entry) => entry.message) // History from a run in one go (or an import) may hold calls without results: never re-run. @@ -1998,7 +2019,7 @@ export class Runner { store.setStepState(sessionId, messageId, state) } const runCall = (call: ToolCall) => - lane.timing.tool(call.name, () => this.runCall(setup, call, { sessionId, kind, tools, signal, held: { credentials: lane.held?.credentials ?? {}, at: lane.issuedAt } })) + lane.timing.tool(call.name, () => this.runCall(setup, call, { sessionId, kind, tools, signal, held: { credentials: lane.held?.credentials ?? {}, at: lane.issuedAt }, callKey: stepCallKey(messageId, state, call) })) if (runsTogether(setup, answer.toolCalls)) { // Calls that can't change anything (lookups in the person's apps) run at the same time. signal.throwIfAborted() diff --git a/ee/apps/headless-runner/src/steps.ts b/ee/apps/headless-runner/src/steps.ts index 09daa2440e..b0cd2cc7d2 100644 --- a/ee/apps/headless-runner/src/steps.ts +++ b/ee/apps/headless-runner/src/steps.ts @@ -14,7 +14,8 @@ import { toolAppSchema, toolCallSchema, type RepeatLimits, type ToolCall, type T * minutes, then its turn fails with `step_crash_loop`. * - **No repeated writes.** Once the model has answered with tool calls, the answer is kept (`pending`) before any * call runs, and each result as it returns. A step that dies there finishes from it: a call without a recorded - * result is never run again, it gets an error result so the model can check its effect. + * result is never run again, it gets an error result so the model can check its effect. A command on the + * conversation's computer went on without the runner and kept its result there, so the computer is asked for it. * - **Stream checkpoints.** A long model call keeps its text so far (at most every 2 s). A step that restarts after a * cancel continues that text where the provider accepts a partial answer (Anthropic Messages without extended * thinking) and otherwise starts over. Nothing but text is kept, so a half-written tool call is never run. diff --git a/ee/apps/headless-runner/src/types.ts b/ee/apps/headless-runner/src/types.ts index 97a11513e4..ba2006735d 100644 --- a/ee/apps/headless-runner/src/types.ts +++ b/ee/apps/headless-runner/src/types.ts @@ -152,7 +152,13 @@ export type SessionComputer = { desktopViewer?(sessionId: string, input: { view: "browser" | "computer"; control: boolean; origin: string }): Promise<{ state: "paused" } | { state: "connected"; url: string; expiresAt: number; paused: boolean }> browserDemonstration?(sessionId: string, command: import("@openwork/types/browser-demonstration").DemonstrationCommand): Promise desktopAction?(sessionId: string, action: "continue" | "pause" | "resume" | "rebuild"): Promise<{ ok: boolean }> - run(sessionId: string, name: string, input: Record): Promise + /** `callKey` names this one call for good, so a command started under it can be picked up later (`pickUp`). */ + run(sessionId: string, name: string, input: Record, callKey?: string): Promise + /** + * The result of a command started under `callKey` whose result never reached the runner: its real result once it + * ends, or that it is still running. Null when the computer never started it. Never starts a command or a computer. + */ + pickUp?(sessionId: string, name: string, input: Record, callKey: string): Promise /** Whether this conversation has a computer here at all (see computer-route.ts); unset means every conversation does. */ available?(sessionId: string): boolean known(sessionId: string): boolean diff --git a/ee/packages/headless-computer/src/computer-commands.test.ts b/ee/packages/headless-computer/src/computer-commands.test.ts new file mode 100644 index 0000000000..d080fd8073 --- /dev/null +++ b/ee/packages/headless-computer/src/computer-commands.test.ts @@ -0,0 +1,177 @@ +/** + * Commands that outlive the request waiting for them (`awaitCommand` in computer.ts). + * Run with `tsx --test src/computer-commands.test.ts`. + */ +import assert from "node:assert/strict" +import { test } from "node:test" +import { SandboxError } from "@openwork/sandbox" +import { createFakeProvider } from "@openwork/sandbox/testing" +import { Computers } from "./computer.js" +import type { ComputerFile, ComputerFiles } from "./types.js" + +type Job = { status: "running" | "done" | "lost"; exitCode: number; stopped?: boolean; stdout: string; stderr: string; starts: number } +const MARKER = "---computer-command---" + +/** + * A computer that keeps commands the way the real one does: a request names a command's files, starts the command + * only when it carries the start and the computer has none under that name, and answers how the command stands. + */ +function fixture() { + const jobs = new Map() + const requests: Array<{ job: string; starts: boolean; waitSeconds: number }> = [] + const added: string[] = [] + const state = { + /** What the next command started does. */ + next: { status: "done", exitCode: 0, stdout: "", stderr: "" } as Omit, + /** How many of the next requests for a command lose their answer, after the computer acted on them. */ + cut: 0, + /** Files in /workspace/out, as the outputs listing prints them. */ + outputs: "", + /** The page images a preview render lists. */ + pages: "", + } + const provider = createFakeProvider({ onRun(spec) { + const script = spec.env?.WORKBOT_COMMAND + const job = script ? /^job='\/workspace\/\.jobs\/([^']+)'/.exec(script)?.[1] : undefined + if (!script || !job) { + if (spec.command.includes("pdftoppm")) return { exitCode: 0, stdout: state.pages, stderr: "" } + if (spec.command.includes('echo "$SRC"')) return { exitCode: 0, stdout: spec.env?.SRC ?? "", stderr: "" } + return { exitCode: 0, stdout: spec.command.includes("/workspace/out") ? `${state.outputs}---computer-exported---\n{}` : "0", stderr: "" } + } + const starts = script.includes("setsid nohup") + requests.push({ job, starts, waitSeconds: Number(/end=\$\(\(SECONDS \+ (\d+)\)\)/.exec(script)?.[1]) }) + const known = jobs.get(job) + if (starts && known) known.starts += 1 + if (starts && !known) jobs.set(job, { ...state.next, starts: 1 }) + if (state.cut > 0) { + state.cut -= 1 + throw new SandboxError({ providerId: "fake", code: "timeout", retryable: false, message: "Operation exceeded its deadline; outcome may be unknown" }) + } + const found = jobs.get(job) + if (!found) return { exitCode: 0, stdout: `${MARKER} unknown 0 0 0 0\n`, stderr: "" } + return { exitCode: 0, stdout: `${MARKER} ${found.status} ${found.exitCode} ${found.stopped ? 1 : 0} ${found.stdout.length} ${found.stderr.length}\n${found.stdout}`, stderr: found.stderr } + } }) + const files: ComputerFiles = { + maxFileBytes: 1_000_000, + list: () => [], + read: async (_sessionId, id) => ({ file: { id, name: "deck.pdf", size: 3, source: "agent", updatedAt: 1 }, bytes: new TextEncoder().encode("pdf") }), + add: async (_sessionId, input) => { + added.push(input.name) + const file: ComputerFile = { id: `file-${added.length}`, name: input.name, size: input.bytes.byteLength, source: "agent", updatedAt: 1 } + return file + }, + replace: async () => null, + putPreview: async () => true, + } + /** One runner process; several over the same provider stand in for one that stopped and the one that took over. */ + const process = () => new Computers({ provider, idlePauseMs: 60_000, keepDays: 1 }, { files, readFile: async () => ({ text: "" }) }) + return { provider, jobs, requests, added, state, process, started: () => [...jobs.values()].reduce((total, job) => total + job.starts, 0) } +} + +test("a command's result reads as it always has, and its outputs go to the person's Files", async () => { + const f = fixture() + const computers = f.process() + f.state.next = { status: "done", exitCode: 2, stdout: "built 3 pages\n", stderr: "warning: font missing\n" } + const failed = await computers.run("session", "bash", { command: "make report" }, "call-1") + assert.deepEqual(failed, { output: "[exit 2]\nbuilt 3 pages\n[stderr]\nwarning: font missing", isError: true }) + + const box = await computers.vm("session") + await f.provider.blocks?.files?.write(box, "/workspace/out/report.csv", new TextEncoder().encode("a,b"), { timeoutMs: 1000 }) + f.state.outputs = "report.csv\t3\t1.0\n" + f.state.next = { status: "done", exitCode: 0, stdout: "x".repeat(40_000), stderr: "" } + const long = await computers.run("session", "bash", { command: "make report" }, "call-2") + assert.equal(long.output, `[exit 0]\n…[10000 earlier characters cut]\n${"x".repeat(30_000)}\n[Added to their Files: report.csv]`) + assert.equal(long.isError, false) + // Each command waits a little longer than its own limit, in one request. + assert.deepEqual(f.requests.map((request) => [request.starts, request.waitSeconds]), [[true, 130], [true, 130]]) +}) + +test("a command whose wait is cut is picked up with its real exit code and output, and is never started again", async () => { + const f = fixture() + f.state.next = { status: "done", exitCode: 3, stdout: "half the rows failed\n", stderr: "" } + // The request that started it never answers; the same call waits again, without the start. + f.state.cut = 1 + const result = await f.process().run("session", "bash", { command: "import rows", timeout_seconds: 30 }, "call-1") + assert.deepEqual(result, { output: "[exit 3]\nhalf the rows failed", isError: true }) + // The second wait is for what is left of the command's time: all of it here, as the first was cut at once. + assert.deepEqual(f.requests.map((request) => [request.starts, request.waitSeconds]), [[true, 40], [false, 40]]) + + // The runner that was waiting is gone. The one that takes over asks for the call, and gets its outputs handed over. + f.state.next = { status: "running", exitCode: 0, stdout: "", stderr: "" } + f.state.cut = 2 + const lost = await f.process().run("session", "bash", { command: "render video", timeout_seconds: 30 }, "call-2") + assert.match(lost.output, /^\[No confirmed result after 30s\. The command may still be running/) + const job = f.jobs.get(f.requests.at(-1)?.job ?? "") + assert.ok(job) + Object.assign(job, { status: "done", exitCode: 0, stdout: "rendered\n" }) + const next = f.process() + const box = await next.vm("session") + await f.provider.blocks?.files?.write(box, "/workspace/out/video.mp4", new TextEncoder().encode("mp4"), { timeoutMs: 1000 }) + f.state.outputs = "video.mp4\t3\t1.0\n" + const picked = await next.pickUp("session", "bash", { command: "render video", timeout_seconds: 30 }, "call-2") + assert.deepEqual(picked, { output: "[exit 0]\nrendered\n[Added to their Files: video.mp4]", isError: false }) + assert.deepEqual(f.added, ["video.mp4"]) + + assert.equal(f.started(), 2, "each command was started once") + assert.equal(f.requests.filter((request) => request.starts).length, 2, "picking a command up never asks to start it") +}) + +test("a command still running when it is picked up says so, and how to check on it", async () => { + const f = fixture() + f.state.next = { status: "running", exitCode: 0, stdout: "frame 12 of 900\n", stderr: "" } + f.state.cut = 2 + await f.process().run("session", "bash", { command: "render", timeout_seconds: 300 }, "call-1") + f.requests.length = 0 + const result = await f.process().pickUp("session", "bash", { command: "render", timeout_seconds: 300 }, "call-1") + const job = [...f.jobs.keys()][0] + assert.equal(result?.isError, true) + assert.equal(result?.output, `[Still running: it did not stop at its 300s limit. Its output so far is in /workspace/.jobs/${job}.out and ${job}.err, and /workspace/.jobs/${job}.exit will hold its exit code once it ends. Check on it there; do not repeat side effects.]\nframe 12 of 900`) + // It waited as long as the command could take to stop, in requests short enough for the provider. + assert.deepEqual(f.requests.map((request) => [request.starts, request.waitSeconds]), [[false, 290], [false, 20]]) + assert.equal(f.started(), 1) +}) + +test("a command stopped at its time limit, or one whose process is gone, says what is known", async () => { + const f = fixture() + const computers = f.process() + f.state.next = { status: "done", exitCode: 124, stopped: true, stdout: "step 1 done\n", stderr: "" } + assert.deepEqual(await computers.run("session", "bash", { command: "slow", timeout_seconds: 5 }, "call-1"), { + output: "[Stopped after 5s, its time limit, before it finished. Use background for long jobs.]\nstep 1 done", + isError: true, + }) + f.state.next = { status: "lost", exitCode: 0, stdout: "step 1 done\n", stderr: "" } + const lost = await computers.run("session", "bash", { command: "slow" }, "call-2") + assert.match(lost.output, /^\[The command's process ended without an exit code.*\]\nstep 1 done$/) + assert.equal(lost.isError, true) +}) + +test("a call the computer never started has no result, and asking never starts a command or a computer", async () => { + const f = fixture() + assert.equal(await f.process().pickUp("session", "bash", { command: "send invoices" }, "call-1"), null) + assert.equal(f.provider.fake.count("create"), 0, "no computer is made to ask") + + await f.process().run("session", "bash", { command: "echo ready" }, "call-0") + f.requests.length = 0 + assert.equal(await f.process().pickUp("session", "bash", { command: "send invoices" }, "call-1"), null) + assert.equal(await f.process().pickUp("session", "look", { paths: ["chart.png"] }, "call-0"), null) + assert.deepEqual(f.requests.map((request) => request.starts), [false]) + assert.equal(f.started(), 1) +}) + +test("a preview render logs when it starts and when it ends, with ids and times only", async (t) => { + const f = fixture() + const computers = f.process() + const lines: string[] = [] + t.mock.method(console, "log", (line: unknown) => void lines.push(String(line))) + const box = await computers.vm("session") + const page = new Uint8Array(24) + new DataView(page.buffer).setUint32(16, 1600) + new DataView(page.buffer).setUint32(20, 900) + await f.provider.blocks?.files?.write(box, "/tmp/computer-preview/file-9/page-1.png", page, { timeoutMs: 1000 }) + f.state.pages = "/tmp/computer-preview/file-9/page-1.png\n" + await computers.previewSavedFile("session", "file-9") + const previews = lines.filter((line) => line.startsWith("[headless-computer] preview ")) + assert.equal(previews[0], '[headless-computer] preview started {"sessionId":"session","fileId":"file-9","floating":false}') + assert.match(previews[1] ?? "", /^\[headless-computer\] preview finished \{"sessionId":"session","fileId":"file-9","floating":false,"outcome":"kept","pages":1,"ms":\d+\}$/) + assert.equal(previews.length, 2) +}) diff --git a/ee/packages/headless-computer/src/computer.ts b/ee/packages/headless-computer/src/computer.ts index 06cf971ec2..26ac9d724b 100644 --- a/ee/packages/headless-computer/src/computer.ts +++ b/ee/packages/headless-computer/src/computer.ts @@ -245,9 +245,28 @@ export class Computers { return { ok: true } } - /** Runs one of the computer's tools (`bash` or `look`) for a conversation. */ - run(sessionId: string, name: string, input: Record): Promise { - return this.holding(sessionId, () => runComputerTool(this, sessionId, name, input)) + /** + * Runs one of the computer's tools (`bash` or `look`) for a conversation. `callKey` names this one call for good: a + * command started under it can be picked up later (`pickUp`) if the wait for it is lost. + */ + run(sessionId: string, name: string, input: Record, callKey?: string): Promise { + return this.holding(sessionId, () => runComputerTool(this, sessionId, name, input, callKey)) + } + + /** + * The result of a command started under `callKey` whose result never reached the runner: its real result once it + * ends, or that it is still running. Null when this computer never started it. Never starts a command or a computer. + */ + pickUp(sessionId: string, name: string, input: Record, callKey: string): Promise { + return this.holding(sessionId, async () => { + const parsed = name === "bash" ? bashInput.safeParse(input) : null + if (!parsed?.success || !(await this.has(sessionId))) return null + const job = commandJob(callKey) + const limit = parsed.data.timeout_seconds ?? DEFAULT_COMMAND_SECONDS + const outcome = await this.awaitCommand(sessionId, job, limit) + if (outcome.status === "unknown" || outcome.status === "unconfirmed" || outcome.status === "direct") return null + return this.commandResult(sessionId, job, limit, outcome, []) + }) } /** The snapshot new computers boot from, for logs. */ @@ -330,9 +349,10 @@ export class Computers { return this.blocks.run(box, spec) } - /** Recover before launching work; never retry an unknown command outcome. */ - private async exec(sessionId: string, command: string, timeoutSeconds: number) { + /** Recover before launching work; never retry an unknown command outcome. `reached`: the computer is up, and is being asked. */ + private async exec(sessionId: string, command: string, timeoutSeconds: number, reached?: () => void) { const box = await this.vm(sessionId) + reached?.() if (this.desktop) return this.execute(box, { command: `printf %s ${shellQuote(JSON.stringify({ command }))} | sudo -n /usr/local/sbin/openwork-desktop-bridge command`, timeoutMs: timeoutSeconds * 1000, @@ -443,8 +463,12 @@ export class Computers { } } + /** + * How many jobs are running: `background` jobs and commands (see `awaitCommand`), which both keep a `.running` file. + * A command's file names its process, so one left behind by a command that was killed isn't counted for ever. + */ private async runningJobs(box: SandboxHandle, timeoutMs = 10_000) { - const jobs = await this.execute(box, { command: `ls ${WORKSPACE}/.jobs/*.running 2>/dev/null | wc -l`, activity: "maintenance", timeoutMs }) + const jobs = await this.execute(box, { command: RUNNING_JOBS, activity: "maintenance", timeoutMs }) return Number(jobs.stdout?.trim() ?? "0") } @@ -715,11 +739,25 @@ export class Computers { const path = `/tmp/computer-preview/source/${fileId}${extension}` await this.execute(vm, { command: "mkdir -p /tmp/computer-preview/source", timeoutMs: 10_000 }) await this.blocks.files.write(vm, path, found.bytes, { timeoutMs: 30_000 }) - await this.renderPreview(sessionId, fileId, path) + await this.renderPreview(sessionId, fileId, path, false) } - /** Renders an Office file with LibreOffice on the computer and keeps its pages as images, the file's preview. */ - private async renderPreview(sessionId: string, fileId: string, path: string) { + /** + * Renders an Office file with LibreOffice on the computer and keeps its pages as images, the file's preview. + * `floating`: nothing waits for it (a command's output), so nothing keeps the runner alive until it is done. + */ + private async renderPreview(sessionId: string, fileId: string, path: string, floating = true) { + if (!this.files) return + // A line when it starts and one when it ends, ids and times only: a start without an end is a preview that was + // lost with its runner, which is how often that happens. + const started = Date.now() + console.log(`[headless-computer] preview started ${JSON.stringify({ sessionId, fileId, floating })}`) + const done = { outcome: "failed", pages: 0 } + try { await this.renderPages(sessionId, fileId, path, done) } + finally { console.log(`[headless-computer] preview finished ${JSON.stringify({ sessionId, fileId, floating, ...done, ms: Date.now() - started })}`) } + } + + private async renderPages(sessionId: string, fileId: string, path: string, done: { outcome: string; pages: number }) { if (!this.files) return // The version being rendered: if the file is revised meanwhile, these pages are stale and are not kept. const version = this.files.list(sessionId).find((file) => file.id === fileId)?.updatedAt @@ -743,7 +781,10 @@ export class Computers { timeoutMs: 150_000, }) const pdfPath = rendered.stdout?.trim() - if (!pdfPath) return + if (!pdfPath) { + done.outcome = "not_rendered" + return + } // One image per page (at most 60), 1600px on the long side: the panel lays them out like a deck. const paged = await this.execute(vm, { command: `pdftoppm -png -l ${MAX_PREVIEW_PAGES} -scale-to 1600 "$PDF" "$OUT/page" && ls "$OUT"/page-*.png | sort -V`, @@ -755,28 +796,82 @@ export class Computers { for (const pagePath of pagePaths) pages.push(new Uint8Array(await this.blocks.files.read(vm, pagePath, { timeoutMs: 30_000 }))) const size = pngSize(pages[0]) const current = this.files.list(sessionId).find((file) => file.id === fileId)?.updatedAt - if (size && current === version) await this.files.putPreview(sessionId, fileId, pages, size) + done.pages = pages.length + done.outcome = !size ? "no_pages" : current !== version ? "revised" : "failed" + if (size && current === version) done.outcome = (await this.files.putPreview(sessionId, fileId, pages, size)) ? "kept" : "not_kept" await this.execute(vm, { command: `rm -rf "$OUT"`, env: { OUT: outDir }, timeoutMs: 10_000 }).catch(() => undefined) } - async bash(sessionId: string, command: string, timeoutSeconds: number): Promise { + async bash(sessionId: string, command: string, timeoutSeconds: number, callKey?: string): Promise { const copied = await this.copyUploads(sessionId) - let result - try { result = await this.exec(sessionId, command, timeoutSeconds) } - catch (error) { - if (!isSandboxError(error) || error.code !== "timeout") throw error - return { output: `[No confirmed result after ${timeoutSeconds}s. The command may still be running; do not repeat side effects. Use background for long jobs.]`, isError: true } + const job = commandJob(callKey) + return this.commandResult(sessionId, job, timeoutSeconds, await this.awaitCommand(sessionId, job, timeoutSeconds, command), copied) + } + + /** + * Waits for a command to end, starting it first when `command` is given. The command runs on the computer apart + * from the request that waits for it, keeps its output and exit code in files, and stops itself at its time limit. + * So a wait that is cut (the provider's limit, a runner that stops) loses nothing: the next wait for the same `job` + * finds it, and a command is never started twice. `unknown`: this computer has no such command. + */ + private async awaitCommand(sessionId: string, job: string, limitSeconds: number, command?: string): Promise { + // The command is over by then: its limit, the time it gets to stop, and a moment to record how it ended. + const longest = limitSeconds + STOP_GRACE_SECONDS + 5 + let left = longest + let start = command === undefined ? undefined : { command, limitSeconds } + let askedAt: number | undefined + let cut = false + for (;;) { + const waitSeconds = Math.min(left, MAX_WAIT_SECONDS) + left -= waitSeconds + const result = await this.exec(sessionId, commandScript(job, waitSeconds, start), waitSeconds + WAIT_SLACK_SECONDS, () => { askedAt ??= Date.now() }).catch((error: unknown) => { + if (!isSandboxError(error) || error.code !== "timeout") throw error + return null + }) + if (!result) { + // The wait was cut, not the command: one more wait, for what is left of its time, which starts nothing. Not + // when the computer never came up, nor a second time. + if (cut || askedAt === undefined) return { status: "unconfirmed" } + cut = true + left = Math.max(1, longest - Math.floor((Date.now() - askedAt) / 1000)) + start = undefined + continue + } + const outcome = readCommand(result) + if (outcome.status !== "running" || left <= 0) return outcome + start = undefined + } + } + + /** What the model is told about a command, and its outputs added to the person's Files once it has ended. */ + private async commandResult(sessionId: string, job: string, limitSeconds: number, outcome: CommandOutcome, copied: string[]): Promise { + const copiedNote = copied.length ? `[Copied the person's files to: ${copied.join(", ")}]` : "" + const unfinished = (note: string, output: { stdout: string; stderr: string }) => ({ + output: [copiedNote, note, output.stdout, output.stderr ? `[stderr]\n${output.stderr}` : ""].filter(Boolean).join("\n"), + isError: true, + }) + if (outcome.status === "unconfirmed" || outcome.status === "unknown") { + return unfinished(`[No confirmed result after ${limitSeconds}s. The command may still be running; do not repeat side effects. Use background for long jobs.]`, { stdout: "", stderr: "" }) + } + if (outcome.status === "running") { + return unfinished(`[Still running: it did not stop at its ${limitSeconds}s limit. Its output so far is in ${JOBS_DIR}/${job}.out and ${job}.err, and ${JOBS_DIR}/${job}.exit will hold its exit code once it ends. Check on it there; do not repeat side effects.]`, outcome) + } + if (outcome.status === "lost") { + return unfinished("[The command's process ended without an exit code: it was killed, or the computer restarted under it. Check its effect before repeating it.]", outcome) + } + if (outcome.status === "done" && outcome.stopped) { + return unfinished(`[Stopped after ${limitSeconds}s, its time limit, before it finished. Use background for long jobs.]`, outcome) } const { saved, skipped } = await this.exportOutputs(sessionId).catch(() => ({ saved: [], skipped: [] })) const parts = [ - copied.length ? `[Copied the person's files to: ${copied.join(", ")}]` : "", - `[exit ${result.exitCode}]`, - tail(result.stdout ?? ""), - result.stderr?.trim() ? `[stderr]\n${tail(result.stderr)}` : "", + copiedNote, + `[exit ${outcome.exitCode}]`, + outcome.stdout, + outcome.stderr ? `[stderr]\n${outcome.stderr}` : "", saved.length ? `[Added to their Files: ${saved.join(", ")}]` : "", skipped.length ? `[Not added to their Files: ${skipped.join("; ")}. Make it smaller, or tell them it is too big to hand over.]` : "", ] - return { output: parts.filter(Boolean).join("\n"), isError: result.exitCode !== 0 } + return { output: parts.filter(Boolean).join("\n"), isError: outcome.exitCode !== 0 } } async look(sessionId: string, paths: string[]): Promise { @@ -835,9 +930,142 @@ function formatBytes(bytes: number) { return `${bytes} bytes` } -function tail(text: string) { +/** `earlier`: how much of the output the computer had already dropped before this end of it. */ +function tail(text: string, earlier = 0) { const trimmed = text.trimEnd() - return trimmed.length > OUTPUT_TAIL_CHARS ? `…[${trimmed.length - OUTPUT_TAIL_CHARS} earlier characters cut]\n${trimmed.slice(-OUTPUT_TAIL_CHARS)}` : trimmed + const cut = Math.max(0, trimmed.length - OUTPUT_TAIL_CHARS) + earlier + return cut ? `…[${cut} earlier characters cut]\n${trimmed.slice(-OUTPUT_TAIL_CHARS)}` : trimmed +} + +/** Where commands and `background` jobs keep their files: `.running` while one runs. */ +const JOBS_DIR = `${WORKSPACE}/.jobs` +/** How long a command gets to stop once its time is up, before it is killed. */ +const STOP_GRACE_SECONDS = 5 +/** How much longer than its wait a request for a command may take, and the longest wait that leaves within MAX_COMMAND_SECONDS. */ +const WAIT_SLACK_SECONDS = 10 +const MAX_WAIT_SECONDS = MAX_COMMAND_SECONDS - WAIT_SLACK_SECONDS +/** The end of each stream is kept and read back in bytes: all that its last OUTPUT_TAIL_CHARS characters can take. */ +const OUTPUT_TAIL_BYTES = OUTPUT_TAIL_CHARS * 4 +/** A command's files stay this long after it started, for a wait that comes back for its result. */ +const COMMAND_KEPT_MINUTES = 24 * 60 +/** Starts the first line of what `commandScript` prints. */ +const COMMAND_MARKER = "---computer-command---" + +type CommandOutput = { stdout: string; stderr: string } +/** + * How a command stands. `done`: it ended (`stopped`: at its time limit). `direct`: it ran inside the request, with no + * files to come back to. `lost`: it was started, and is neither running nor done. `unknown`: the computer has no such + * command. `unconfirmed`: the computer couldn't be asked. + */ +type CommandOutcome = + | ({ status: "done"; exitCode: number; stopped: boolean } & CommandOutput) + | ({ status: "direct"; exitCode: number } & CommandOutput) + | ({ status: "running" } & CommandOutput) + | ({ status: "lost" } & CommandOutput) + | { status: "unknown" } + | { status: "unconfirmed" } + +/** Reads the state of process `$pid` into `$state`, false when there is none; `Z` is one that ended and awaits its parent. */ +const PROCESS_STATE = `read -r _pid _name state _rest 2>/dev/null < "/proc/$pid/stat"` + +/** The name of a command's files: the same for the same call, so a later wait finds them; one of a kind without one. */ +function commandJob(callKey?: string) { + return `bash-call-${createHash("sha256").update(callKey ?? crypto.randomUUID()).digest("hex").slice(0, 24)}` +} + +/** + * Counts running jobs. `background` leaves its `.running` file empty until it removes it; a command's names its + * process and the boot it belongs to, so a file left by a command that was killed, or by a restart, doesn't count. + */ +const RUNNING_JOBS = [ + `boot=$(cat /proc/sys/kernel/random/boot_id 2>/dev/null); count=0`, + `for file in ${JOBS_DIR}/*.running; do`, + ` [ -e "$file" ] || continue`, + ` pid=; at=; state=; read -r pid at 2>/dev/null < "$file"`, + ` case "$pid" in`, + ` ''|*[!0-9]*) count=$((count + 1)) ;;`, + ` *) [ "$at" = "$boot" ] && ${PROCESS_STATE} && [ "$state" != Z ] && count=$((count + 1)) ;;`, + ` esac`, + `done`, + `echo "$count"`, +].join("\n") + +/** + * Runs a command in its own session, so it outlives the request that started it: $1 its files, $2 its time limit in + * seconds, $3 how many bytes of each stream to keep, $4 the command. Its output goes to `$1.out` and `$1.err`; when it + * ends, `$1.exit` gets ` <1 when its limit stopped it> `. + */ +const COMMAND_RUNNER = [ + `echo "$$ $(cat /proc/sys/kernel/random/boot_id 2>/dev/null)" > "$1.running.new" && mv "$1.running.new" "$1.running"`, + `timeout -k ${STOP_GRACE_SECONDS} "$2" bash -c "$4" > "$1.out" 2> "$1.err" < /dev/null`, + `code=$?; stopped=0; sizes=`, + `case "$code" in 124|137) [ "$SECONDS" -ge "$2" ] && stopped=1 ;; esac`, + `for stream in out err; do`, + ` size=$(wc -c < "$1.$stream"); sizes="$sizes $size"`, + ` [ "$size" -gt "$3" ] && tail -c "$3" "$1.$stream" > "$1.$stream.new" && mv "$1.$stream.new" "$1.$stream"`, + `done`, + `echo "$code $stopped$sizes" > "$1.exit.new" && mv "$1.exit.new" "$1.exit"`, + `rm -f "$1.running"`, +].join("\n") + +/** + * What one request for a command runs: with `start`, it starts the command unless the computer already has it, then + * it waits up to `waitSeconds` for the command to end. It prints ` `, then the end of the command's stdout, with the end of its stderr on stderr. + */ +function commandScript(job: string, waitSeconds: number, start?: { command: string; limitSeconds: number }) { + return [ + `job=${shellQuote(`${JOBS_DIR}/${job}`)}; boot=$(cat /proc/sys/kernel/random/boot_id 2>/dev/null)`, + ...(start ? [ + `run=${shellQuote(start.command)}`, + `mkdir -p ${JOBS_DIR} 2>/dev/null`, + // Making its output file is what claims a command, and a file can be made once: it never starts twice. + `if ( set -C; : > "$job.out" ) 2>/dev/null; then`, + ` find ${JOBS_DIR} -maxdepth 1 -name 'bash-call-*' -mmin +${COMMAND_KEPT_MINUTES} -delete 2>/dev/null`, + ` echo "$$ $boot" > "$job.running"`, + ` setsid nohup bash -c ${shellQuote(COMMAND_RUNNER)} _ "$job" ${start.limitSeconds} ${OUTPUT_TAIL_BYTES} "$run" > /dev/null 2>&1 < /dev/null &`, + // Nowhere to keep its files (a full disk): it runs inside this request, as commands did before. + `elif [ ! -e "$job.out" ]; then exec bash -c "$run"`, + `fi`, + ] : []), + `look() {`, + ` status=done; [ -e "$job.exit" ] && return`, + ` status=running; pid=; at=; state=; read -r pid at 2>/dev/null < "$job.running"`, + ` [ -n "$pid" ] && [ "$at" = "$boot" ] && ${PROCESS_STATE} && [ "$state" != Z ] && return`, + // It may have ended between the two looks. + ` status=done; [ -e "$job.exit" ] && return`, + ` status=lost; [ -e "$job.out" ] && return`, + ` status=unknown`, + `}`, + `look; end=$((SECONDS + ${waitSeconds}))`, + `while [ "$status" = running ] && [ "$SECONDS" -lt "$end" ]; do`, + // Looks often at first, so a quick command isn't held up, then less and less. + ` if [ "$SECONDS" -lt 1 ]; then sleep 0.01; elif [ "$SECONDS" -lt 10 ]; then sleep 0.05; else sleep 0.25; fi`, + ` look`, + `done`, + `code=0; stopped=0; out=0; err=0`, + `if [ "$status" = done ]; then read -r code stopped out err < "$job.exit"`, + `elif [ "$status" != unknown ]; then out=$(wc -c 2>/dev/null < "$job.out"); err=$(wc -c 2>/dev/null < "$job.err"); fi`, + 'echo "' + COMMAND_MARKER + ' $status ${code:-0} ${stopped:-0} ${out:-0} ${err:-0}"', + `if [ "$status" != unknown ]; then tail -c ${OUTPUT_TAIL_BYTES} "$job.out" 2>/dev/null; [ -e "$job.err" ] && tail -c ${OUTPUT_TAIL_BYTES} "$job.err" >&2; fi`, + `exit 0`, + ].join("\n") +} + +/** What a request for a command answered. Without `commandScript`'s first line, the command ran inside the request. */ +function readCommand(result: { exitCode: number; stdout?: string; stderr?: string }): CommandOutcome { + const stdout = result.stdout ?? "" + const stderr = result.stderr ?? "" + if (!stdout.startsWith(`${COMMAND_MARKER} `)) return { status: "direct", exitCode: result.exitCode, stdout: tail(stdout), stderr: stderr.trim() ? tail(stderr) : "" } + const lineEnd = stdout.indexOf("\n") + const [, status, code, stopped, outBytes, errBytes] = stdout.slice(0, lineEnd === -1 ? stdout.length : lineEnd).split(" ") + if (status !== "done" && status !== "running" && status !== "lost") return { status: "unknown" } + // What the computer dropped before the end it kept, so the note on what was cut still counts all of it. + const dropped = (bytes: string | undefined) => Math.max(0, (Number(bytes) || 0) - OUTPUT_TAIL_BYTES) + const output = { stdout: tail(lineEnd === -1 ? "" : stdout.slice(lineEnd + 1), dropped(outBytes)), stderr: stderr.trim() ? tail(stderr, dropped(errBytes)) : "" } + if (status !== "done") return { status, ...output } + const exitCode = Number(code) + return Number.isInteger(exitCode) && code !== "" ? { status, exitCode, stopped: stopped === "1", ...output } : { status: "lost", ...output } } function safeJson(text: string): unknown { @@ -898,11 +1126,11 @@ const bashInput = z.object({ }) const lookInput = z.object({ paths: z.array(z.string().min(1)).min(1).max(4) }) -async function runComputerTool(computers: Computers, sessionId: string, name: string, input: Record): Promise { +async function runComputerTool(computers: Computers, sessionId: string, name: string, input: Record, callKey?: string): Promise { if (name === "bash") { const parsed = bashInput.safeParse(input) if (!parsed.success) return { output: "bash needs a command (and optionally timeout_seconds up to 300).", isError: true } - return computers.bash(sessionId, parsed.data.command, parsed.data.timeout_seconds ?? DEFAULT_COMMAND_SECONDS) + return computers.bash(sessionId, parsed.data.command, parsed.data.timeout_seconds ?? DEFAULT_COMMAND_SECONDS, callKey) } if (name === "look") { const parsed = lookInput.safeParse(input)