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)