From 47643506a30acf88eece0569674ab127f0d80fdc Mon Sep 17 00:00:00 2001 From: Jalil Date: Fri, 13 Nov 2026 17:31:29 -0800 Subject: [PATCH] perf(headless-computer): skip checks on a computer just seen running Every computer call asked the provider whether the computer was running, and every desktop call also asked the desktop for its health, even when the previous call had confirmed both a second earlier. Opening the computer view looked the computer up twice (three times on a process that had not used it yet), and the view renews its ticket every 45 seconds. What this process saw of a computer is now kept for 15 seconds (seen-running.ts): - vm() returns without a lookup when the provider showed the computer running within that time; a command that comes back counts too. - startDesktop returns without a health check when the desktop was found current on that computer within that time. - desktopViewer asks nothing when the computer was seen running, and otherwise looks it up once. A computer found running by that lookup is kept, so vm() does not look it up again. Any provider call that throws, and any pause, stop or removal made by this process, forgets what was seen, so the next call asks again. A desktop operation that exits non-zero forgets that the desktop was current. When a lookup was skipped and the computer had been paused or removed since, the first call on it is refused. That call is sent once more after vm() has resumed or recreated the computer, but only when it failed within five seconds, did not time out, a lookup shows the computer is not running, and this process did not pause or remove it itself. The idle check and pauseIdle keep reading the provider directly and never receive a handle that skipped a lookup. A view never wakes a computer: when its skipped lookup turns out wrong it looks the computer up and answers paused. A paused computer whose desktop start fails is no longer replaced as unresponsive, which would have deleted it. Signed-off-by: Jalil --- ee/packages/headless-computer/src/computer.ts | 98 ++++++-- .../headless-computer/src/desktop-timing.ts | 5 +- .../src/seen-running.test.ts | 215 ++++++++++++++++++ .../headless-computer/src/seen-running.ts | 161 +++++++++++++ 4 files changed, 463 insertions(+), 16 deletions(-) create mode 100644 ee/packages/headless-computer/src/seen-running.test.ts create mode 100644 ee/packages/headless-computer/src/seen-running.ts diff --git a/ee/packages/headless-computer/src/computer.ts b/ee/packages/headless-computer/src/computer.ts index 06cf971ec22..f833dc6c801 100644 --- a/ee/packages/headless-computer/src/computer.ts +++ b/ee/packages/headless-computer/src/computer.ts @@ -7,6 +7,7 @@ import { demonstrationReplySchema, type DemonstrationCommand } from "@openwork/t import { Desktop, DesktopError, type CarriedState, type DesktopState, type DesktopView } from "./desktop.js" import { DESKTOP_TOOL } from "./desktop-tool.js" import { DesktopTiming, timedProvider } from "./desktop-timing.js" +import { SeenRunning, watchedProvider } from "./seen-running.js" /** One provider-neutral Linux computer per conversation. Jobs own idle policy and files; adapters own transport. */ export type ComputerOptions = { @@ -103,6 +104,8 @@ export class Computers { private readonly recovering = new Map>() /** When each session's computer was last replaced for a desktop that wouldn't start. */ private readonly recovered = new Map() + /** Computers seen running moments ago, which `vm()` and `startDesktop` don't ask about again. */ + private readonly seen = new SeenRunning() private readonly snapshot: string private readonly files?: ComputerFiles @@ -120,7 +123,7 @@ export class Computers { * `readFile` turns a file into what the model can see (images, PDFs, text), the same way tool results are read. */ constructor(private readonly options: ComputerOptions, deps: { files?: ComputerFiles; readFile: FileReader }) { - this.provider = timedProvider(options.provider, key => this.timings.get(key)) + this.provider = watchedProvider(timedProvider(options.provider, key => this.timings.get(key)), this.seen, (sessionId, box, error) => this.afterWrongGuess(sessionId, box, error)) this.blocks = computerBlocks(this.provider) this.snapshot = this.provider.currentImage()?.id ?? "configured image" this.files = deps.files @@ -178,21 +181,55 @@ export class Computers { const timing = this.track(sessionId, "viewer") let outcome = "failed" try { - const pending = this.vms.get(sessionId) - const existing = pending ? await this.provider.inspect(await pending) : await this.provider.find({ idempotencyKey: this.slug(sessionId), labels: this.spec(sessionId).labels }) + // Seen running moments ago: it isn't paused, so nothing is asked. A view never wakes a computer, so when that + // turns out wrong the computer is looked up after all. + let failed: { error: unknown } | undefined + if (await this.seenRunning(sessionId)) { + try { + const viewer = await this.view(this.desktop, sessionId, input) + outcome = "ok" + return viewer + } catch (error) { failed = { error } } + } + const existing = await this.lookup(sessionId) if (existing?.state === "stopped" || existing?.state === "stopping") { outcome = "paused" return { state: "paused" as const } } - const box = await this.vm(sessionId) - try { - const viewer = { state: "connected" as const, ...await this.desktop.viewer(await this.startDesktop(sessionId, box), input) } - outcome = "ok" - return viewer - } finally { this.release(sessionId) } + // Still running: it failed for another reason. + if (failed && existing?.state === "running") throw failed.error + const viewer = await this.view(this.desktop, sessionId, input) + outcome = "ok" + return viewer } finally { this.finish(timing, outcome) } } + /** A viewer ticket from the session's computer, which the caller knows or takes to be running. */ + private async view(desktop: Desktop, sessionId: string, input: { view: DesktopView; control: boolean; origin: string }) { + const box = await this.vm(sessionId) + // Not woken on a wrong guess: `desktopViewer` looks it up instead. + this.seen.take(box) + try { return { state: "connected" as const, ...await desktop.viewer(await this.startDesktop(sessionId, box), input) } } + finally { this.release(sessionId) } + } + + /** Whether the session's computer, as this process holds it, was seen running moments ago. */ + private async seenRunning(sessionId: string) { + const pending = this.vms.get(sessionId) + return pending !== undefined && this.seen.running(await pending) + } + + /** + * The session's computer as its provider has it now, without starting it. One found running is kept, so the `vm()` + * that follows doesn't look it up again. + */ + private async lookup(sessionId: string) { + const pending = this.vms.get(sessionId) + const existing = pending ? await this.provider.inspect(await pending) : await this.provider.find({ idempotencyKey: this.slug(sessionId), labels: this.spec(sessionId).labels }) + if (existing?.state === "running" && !this.vms.has(sessionId)) this.vms.set(sessionId, Promise.resolve(existing)) + return existing + } + browserDemonstration(sessionId: string, command: DemonstrationCommand) { return this.holding(sessionId, () => this.demonstrate(sessionId, command)) } @@ -200,8 +237,7 @@ export class Computers { private async demonstrate(sessionId: string, command: DemonstrationCommand) { if (!this.desktop) throw new Error("Desktop access is not configured.") if (command.action === "read") { - const pending = this.vms.get(sessionId) - const existing = pending ? await this.provider.inspect(await pending) : await this.provider.find({ idempotencyKey: this.slug(sessionId), labels: this.spec(sessionId).labels }) + const existing = await this.lookup(sessionId) if (!existing || existing.state !== "running") throw new Error("Resume the computer before checking its demonstration.") } const box = await this.vm(sessionId) @@ -293,9 +329,15 @@ export class Computers { // replacement, is still waited for. if (!this.checkpointing.delete(sessionId)) await this.pausing.get(sessionId) const prior = this.vms.get(sessionId) + let guessed = false // Reserve the transition before awaiting provider state. Concurrent viewer // and tool requests must not both issue a start against the same paused VM. const pending = prior ? prior.then(async box => { + // Seen running moments ago: not asked again. `afterWrongGuess` is what happens when it no longer is. + if (this.seen.running(box)) { + guessed = true + return box + } const current = await this.provider.inspect(box) if (current.state === "running") return current if (current.state === "missing") this.copied.delete(sessionId) @@ -303,7 +345,21 @@ export class Computers { }) : this.resolve(sessionId) this.vms.set(sessionId, pending) pending.catch(() => { if (this.vms.get(sessionId) === pending) this.vms.delete(sessionId) }) - return pending + const box = await pending + return guessed ? this.seen.guess(box, sessionId) : box + } + + /** + * A call failed on a computer `vm()` took to be running without asking. Returns the computer to run it on once more + * when that one turns out to have been paused or removed, so the call never started; null leaves the failure as it + * is. A call that timed out may have started, and a computer still running failed for another reason: neither is + * run twice. + */ + private async afterWrongGuess(sessionId: string, box: SandboxHandle, error: unknown): Promise { + if (isSandboxError(error) && error.code === "timeout") return null + const current = await this.provider.inspect(box).catch(() => null) + if (!current || current.state === "running") return null + return this.vm(sessionId) } private spec(sessionId: string): SandboxSpec { @@ -457,16 +513,21 @@ export class Computers { private async startDesktop(sessionId: string, box: SandboxHandle, self = 0): Promise { const desktop = this.desktop if (!desktop) throw new Error("Desktop access is not configured.") + // Found current on this computer moments ago: it isn't asked for its health again. + if (this.seen.desktopCurrent(box)) return box this.awaitingDesktop.set(sessionId, (this.awaitingDesktop.get(sessionId) ?? 0) + self) try { - this.desktops.set(sessionId, await desktop.start(box)) + await this.begin(desktop, sessionId, box) return box } catch (error) { if (!(error instanceof DesktopError)) throw error if (error.desktopCode === "desktop_unresponsive") { + // A paused computer isn't unresponsive: it was paused since it was last seen, and replacing it would delete it. + const state = (await this.provider.inspect(box).catch(() => null))?.state + if (state === "stopped" || state === "stopping") throw error const fresh = await this.recover(sessionId, box) if (!fresh) throw error - this.desktops.set(sessionId, await desktop.start(fresh)) + await this.begin(desktop, sessionId, fresh) return fresh } if (error.desktopCode !== "desktop_outdated") throw error @@ -474,7 +535,7 @@ export class Computers { if (await desktop.replacedInVain(box, NEEDED_REPLACEMENT_RETRY_MS)) throw error const fresh = await this.replace(sessionId, box, { required: true }) if (!fresh) throw error - this.desktops.set(sessionId, await desktop.start(fresh)) + await this.begin(desktop, sessionId, fresh) return fresh } finally { const left = (this.awaitingDesktop.get(sessionId) ?? self) - self @@ -483,6 +544,13 @@ export class Computers { } } + /** Starts the desktop on `box` and keeps what it found, for the idle check and for the next `startDesktop`. */ + private async begin(desktop: Desktop, sessionId: string, box: SandboxHandle) { + const state = await desktop.start(box) + this.desktops.set(sessionId, state) + this.seen.desktop(box, state === "current") + } + /** * The replacement for a computer whose desktop won't start, or null when the computer stays: something other than * that start is running on it, or it is itself a replacement made moments ago. diff --git a/ee/packages/headless-computer/src/desktop-timing.ts b/ee/packages/headless-computer/src/desktop-timing.ts index 449fd74caef..5b250df4e08 100644 --- a/ee/packages/headless-computer/src/desktop-timing.ts +++ b/ee/packages/headless-computer/src/desktop-timing.ts @@ -18,7 +18,10 @@ function bridgeStep(command: string) { export class DesktopTiming { private readonly started = performance.now() private readonly steps: Record = {} - /** The VM's state when the request first looked it up: `stopped` is paused, `missing` means a new computer. */ + /** + * The VM's state when the request first looked it up: `stopped` is paused, `missing` means a new computer. Unset when + * it was seen running moments ago and never looked up. + */ before?: string constructor(readonly sessionId: string, readonly operation: string) {} diff --git a/ee/packages/headless-computer/src/seen-running.test.ts b/ee/packages/headless-computer/src/seen-running.test.ts new file mode 100644 index 00000000000..1d9d4bab9cd --- /dev/null +++ b/ee/packages/headless-computer/src/seen-running.test.ts @@ -0,0 +1,215 @@ +import assert from "node:assert/strict" +import { createHash } from "node:crypto" +import { test, type TestContext } from "node:test" +import { SandboxError, type RunResult, type RunSpec, type SandboxProvider } from "@openwork/sandbox" +import { createFakeProvider } from "@openwork/sandbox/testing" +import identity from "../dist/desktop-identity.mjs" +import { Computers } from "./computer.js" +import { SEEN_MS } from "./seen-running.js" + +/** What a desktop running this runner's image and guest answers to a health check. */ +const VERSION = createHash("sha256").update(identity.image).update(identity.guest).digest("hex") +const VIEW = { view: "computer" as const, control: false, origin: "https://chat.example.test" } +const OPEN = { view: "browser", operation: "open", args: {} } +const SESSION = "chat" + +/** + * A desktop computer on the fake provider, counting what the provider and the computer are asked. `wakesOnCommand`: a + * paused computer starts again on any command, as on a provider that resumes on use. + */ +function fixture(t: TestContext, input: { onRun?: (spec: RunSpec) => Promise | RunResult | undefined; wakesOnCommand?: boolean } = {}) { + const { onRun } = input + const commands: string[] = [] + const provider = createFakeProvider({ + capabilities: { platform: { os: "linux", isolation: "microvm" } }, + async onRun(spec) { + commands.push(spec.command) + const answer = await onRun?.(spec) + if (answer) return answer + const ok = (stdout: string) => ({ exitCode: 0, stdout, stderr: "" }) + if (spec.command.includes("call '/health'")) return ok(JSON.stringify({ version: VERSION })) + if (spec.command.includes("call '/viewer'")) return ok(JSON.stringify({ token: "t".repeat(43), paused: false })) + if (spec.command.includes("call '/")) return ok(JSON.stringify({ ok: true })) + return ok("0") + }, + }) + const run = provider.blocks?.run + const used: SandboxProvider = !input.wakesOnCommand || !run ? provider : { + ...provider, + blocks: { + ...provider.blocks, + run: async (box, spec) => { + const id = box.ref.ref.sandboxId + if (id && provider.fake.sandboxes().some(record => record.id === id && record.state === "stopped")) provider.fake.setState(id, "running") + return run(box, spec) + }, + }, + } + const computers = new Computers({ provider: used, idlePauseMs: 60_000, keepDays: 1, desktop: true }, { readFile: async () => ({ text: "" }) }) + t.after(() => computers.delete(SESSION)) + const lines: string[] = [] + t.mock.method(console, "log", (line: unknown) => { lines.push(String(line)) }) + return { + provider, + computers, + /** How many commands containing `part` reached the computer. */ + sent: (part: string) => commands.filter(command => command.includes(part)).length, + lookups: () => provider.fake.count("inspect") + provider.fake.count("find"), + /** Changes the computer's state at the provider, as something other than this runner would. */ + behindItsBack(state: "stopped" | "missing") { + const record = provider.fake.sandbox(computers.slug(SESSION)) + assert.ok(record) + provider.fake.setState(record.id, state) + }, + state: () => provider.fake.sandboxes().at(-1)?.state, + timings: () => lines.filter(line => line.startsWith("[headless-computer] desktop timing ")).map(line => JSON.parse(line.slice(35)) as { operation: string; outcome: string; before: string | null; steps: Record }), + } +} + +test("calls close together ask once whether the computer is running and its desktop is current", async t => { + const f = fixture(t) + await f.computers.desktopRun(SESSION, OPEN) + // Creating it: found missing, then seen running. Its desktop is asked for its health once. + const lookups = f.lookups() + assert.equal(f.sent("call '/health'"), 1) + + await f.computers.desktopRun(SESSION, OPEN) + assert.equal((await f.computers.bash(SESSION, "echo one", 10)).isError, false) + const viewer = await f.computers.desktopViewer(SESSION, VIEW) + assert.equal(viewer.state, "connected") + assert.equal(f.lookups(), lookups, "no lookup after the first call") + assert.equal(f.sent("call '/health'"), 1, "no health check after the first call") + assert.equal(f.sent("call '/browser/task'"), 2) + + // The view's log line shows only what it did: a ticket and an address. + const timing = f.timings().find(line => line.operation === "viewer") + assert.deepEqual([timing?.outcome, timing?.before, Object.keys(timing?.steps ?? {}).sort()], ["ok", null, ["endpoint", "viewer"]]) +}) + +test("a view on a process that hasn't used the computer looks it up once", async t => { + const f = fixture(t) + await f.computers.start(SESSION) + const later = new Computers({ provider: f.provider, idlePauseMs: 60_000, keepDays: 1, desktop: true }, { readFile: async () => ({ text: "" }) }) + const lookups = f.lookups() + assert.equal((await later.desktopViewer(SESSION, VIEW)).state, "connected") + assert.equal(f.lookups(), lookups + 1) + const [timing] = f.timings() + assert.deepEqual([timing?.before, timing?.steps.vm_lookup?.count, timing?.steps.health?.count], ["running", 1, 1]) +}) + +test("calls further apart ask again", async t => { + t.mock.timers.enable({ apis: ["Date"] }) + const f = fixture(t) + await f.computers.desktopRun(SESSION, OPEN) + const lookups = f.lookups() + t.mock.timers.tick(SEEN_MS - 1) + await f.computers.desktopRun(SESSION, OPEN) + assert.deepEqual([f.lookups(), f.sent("call '/health'")], [lookups, 1]) + // The second call's command came back, so the computer was running then; its desktop was last asked at the start. + t.mock.timers.tick(1) + await f.computers.desktopRun(SESSION, OPEN) + assert.deepEqual([f.lookups(), f.sent("call '/health'")], [lookups, 2]) + t.mock.timers.tick(SEEN_MS) + await f.computers.desktopRun(SESSION, OPEN) + assert.deepEqual([f.lookups(), f.sent("call '/health'")], [lookups + 1, 3]) +}) + +test("a computer paused or removed behind the runner's back is brought back by the next command, which runs once", async t => { + const f = fixture(t) + await f.computers.bash(SESSION, "echo one", 10) + f.behindItsBack("stopped") + assert.equal((await f.computers.bash(SESSION, "echo two", 10)).isError, false) + assert.equal(f.provider.fake.count("start"), 1) + assert.equal(f.sent("echo two"), 1) + + f.behindItsBack("missing") + assert.equal((await f.computers.bash(SESSION, "echo three", 10)).isError, false) + assert.equal(f.provider.fake.count("create"), 2) + assert.equal(f.sent("echo three"), 1) + assert.equal(f.state(), "running") +}) + +test("a view of a computer paused behind the runner's back says so and leaves it paused", async t => { + const f = fixture(t) + await f.computers.desktopRun(SESSION, OPEN) + f.behindItsBack("stopped") + assert.deepEqual(await f.computers.desktopViewer(SESSION, VIEW), { state: "paused" }) + assert.equal(f.provider.fake.count("start"), 0) + assert.equal(f.state(), "stopped") + // A step that needs the desktop wakes it, as before. + assert.equal((await f.computers.desktopRun(SESSION, OPEN)).isError, false) + assert.equal(f.provider.fake.count("start"), 1) +}) + +test("a paused computer whose desktop was never started here is not taken for one that won't start", async t => { + const f = fixture(t) + await f.computers.start(SESSION) + f.behindItsBack("stopped") + assert.deepEqual(await f.computers.desktopViewer(SESSION, VIEW), { state: "paused" }) + assert.equal(f.provider.fake.count("destroy"), 0, "replacing it would delete the person's computer") + assert.equal(f.provider.fake.count("start"), 0) +}) + +test("a view right after this runner paused the computer sends it nothing", async t => { + const f = fixture(t, { wakesOnCommand: true }) + await f.computers.desktopRun(SESSION, OPEN) + assert.deepEqual(await f.computers.desktopAction(SESSION, "pause"), { ok: true }) + const sent = f.sent("") + assert.deepEqual(await f.computers.desktopViewer(SESSION, VIEW), { state: "paused" }) + assert.equal(f.sent(""), sent, "a command would wake the paused computer") + assert.equal(f.state(), "stopped") +}) + +test("a command cut short by this runner pausing the computer is not run again", async t => { + let cut = () => {} + const paused = new Promise(resolve => { cut = resolve }) + const f = fixture(t, { async onRun(spec) { + if (!spec.command.includes("slow")) return undefined + await paused + throw new SandboxError({ providerId: "fake", code: "transient", message: "connection closed" }) + } }) + await f.computers.bash(SESSION, "echo one", 10) + const slow = f.computers.bash(SESSION, "slow", 10) + while (f.sent("slow") === 0) await new Promise(resolve => setTimeout(resolve, 1)) + assert.deepEqual(await f.computers.desktopAction(SESSION, "pause"), { ok: true }) + cut() + await assert.rejects(slow, /connection closed/) + assert.equal(f.sent("slow"), 1) + assert.equal(f.provider.fake.count("start"), 0) + assert.equal(f.state(), "stopped") +}) + +test("a command whose outcome is unknown is not run again, and the next one looks the computer up", async t => { + const f = fixture(t, { onRun(spec) { + if (spec.command.includes("unsure")) throw new SandboxError({ providerId: "fake", code: "timeout", retryable: false, message: "unknown outcome" }) + return undefined + } }) + await f.computers.bash(SESSION, "echo one", 10) + const lookups = f.lookups() + assert.match((await f.computers.bash(SESSION, "unsure", 10)).output, /may still be running/) + assert.equal(f.sent("unsure"), 1) + assert.equal(f.lookups(), lookups) + await f.computers.bash(SESSION, "echo two", 10) + assert.equal(f.lookups(), lookups + 1) +}) + +test("a desktop that failed an operation is asked for its health before the next one", async t => { + let failing = true + const f = fixture(t, { onRun: spec => spec.command.includes("call '/browser/task'") && failing ? { exitCode: 1, stdout: "", stderr: "" } : undefined }) + await assert.rejects(f.computers.desktopRun(SESSION, OPEN), /didn't finish/) + failing = false + await f.computers.desktopRun(SESSION, OPEN) + assert.equal(f.sent("call '/health'"), 2) +}) + +test("an idle check asks the provider itself, and runs nothing on a computer paused since it was seen", async t => { + const f = fixture(t) + await f.computers.bash(SESSION, "echo one", 10) + f.behindItsBack("stopped") + const lookups = f.lookups() + const sent = f.sent("") + await f.computers.pauseIdle(SESSION) + assert.equal(f.lookups(), lookups + 1) + assert.equal(f.sent(""), sent, "a command would wake the paused computer") + assert.equal(f.provider.fake.count("start"), 0) +}) diff --git a/ee/packages/headless-computer/src/seen-running.ts b/ee/packages/headless-computer/src/seen-running.ts new file mode 100644 index 00000000000..550fe21885a --- /dev/null +++ b/ee/packages/headless-computer/src/seen-running.ts @@ -0,0 +1,161 @@ +import type { RunSpec, SandboxHandle, SandboxProvider } from "@openwork/sandbox" + +/** + * How long a computer the provider showed running is taken to still be running, and a desktop found current to still + * be current. Calls a few seconds apart then ask neither again; anything further apart asks as before. + */ +export const SEEN_MS = 15_000 + +/** + * A paused or removed computer refuses a call at once. One that fails later than this had started, so its outcome is + * unknown and it is not run again. + */ +const REFUSED_WITHIN_MS = 5_000 + +/** A desktop operation sent through the guest's helper (desktop.ts `call`), as opposed to a person's command. */ +const DESKTOP_CALL = /openwork-desktop-bridge call / + +type Guess = { sessionId: string; ended: number } + +const keyOf = (box: SandboxHandle) => JSON.stringify(box.ref) + +/** + * What this process saw of each computer moments ago: that it was running, and that its desktop was current. Anything + * that fails on a computer, pauses it, stops it or removes it forgets both, so the next call asks again. + */ +export class SeenRunning { + private readonly seen = new Map() + /** How many times this process paused, stopped or removed each computer. */ + private readonly ended = new Map() + /** Handles given out on what was seen instead of asking the provider, until their first call. */ + private readonly guesses = new WeakMap() + + /** Whether `box` was seen running within SEEN_MS. */ + running(box: SandboxHandle) { + const entry = this.seen.get(keyOf(box)) + return entry !== undefined && Date.now() - entry.at < SEEN_MS + } + + /** Whether the desktop on `box` was found current within SEEN_MS, on a computer still taken to be running. */ + desktopCurrent(box: SandboxHandle) { + const at = this.seen.get(keyOf(box))?.desktopAt + return at !== undefined && Date.now() - at < SEEN_MS && this.running(box) + } + + /** Notes what starting the desktop on `box` found. */ + desktop(box: SandboxHandle, current: boolean) { + const entry = this.seen.get(keyOf(box)) + if (!entry) return + if (current) entry.desktopAt = Date.now() + else delete entry.desktopAt + } + + /** Records what the provider just said of a computer. */ + observe(box: T): T { + if (box?.state === "running") this.seen.set(keyOf(box), { ...this.seen.get(keyOf(box)), at: box.observedAt }) + else if (box) this.forget(box) + return box + } + + /** A command came back from `box`, so it is still running. Never brings back what a failure or a pause forgot. */ + ran(box: SandboxHandle) { + const entry = this.seen.get(keyOf(box)) + if (entry) entry.at = Date.now() + } + + forget(box: SandboxHandle) { + this.seen.delete(keyOf(box)) + } + + /** This process is pausing, stopping or removing `box`: a call that cuts short is not one to run again. */ + end(box: SandboxHandle) { + this.forget(box) + this.ended.set(keyOf(box), (this.ended.get(keyOf(box)) ?? 0) + 1) + } + + /** `box` for a caller who gets it on what was seen: a copy, so that its first call can be told from later ones. */ + guess(box: SandboxHandle, sessionId: string): SandboxHandle { + const copy = { ...box } + this.guesses.set(copy, { sessionId, ended: this.ended.get(keyOf(box)) ?? 0 }) + return copy + } + + /** Takes the guess off a handle. Whoever takes it deals with it having been wrong. */ + take(box: SandboxHandle) { + const guess = this.guesses.get(box) + this.guesses.delete(box) + return guess + } + + /** Whether this process has left `box` alone since `guess` was made. */ + stands(box: SandboxHandle, guess: Guess) { + return guess.ended === (this.ended.get(keyOf(box)) ?? 0) + } +} + +/** + * The provider with what each call shows of a computer kept in `seen`. The first call on a computer that was only + * taken to be running (`SeenRunning.guess`) is run once more, on the computer `afterWrongGuess` returns, when it was + * refused and this process did not pause or remove the computer itself. + */ +export function watchedProvider( + provider: SandboxProvider, + seen: SeenRunning, + afterWrongGuess: (sessionId: string, box: SandboxHandle, error: unknown) => Promise, +): SandboxProvider { + async function on(box: SandboxHandle, call: (box: SandboxHandle) => Promise): Promise { + const guess = seen.take(box) + const started = Date.now() + try { return await call(box) } + catch (error) { + seen.forget(box) + const refused = guess && seen.stands(box, guess) && Date.now() - started < REFUSED_WITHIN_MS + const again = refused ? await afterWrongGuess(guess.sessionId, box, error) : null + if (!again) throw error + return call(again) + } + } + async function ending(box: SandboxHandle, call: () => Promise) { + seen.end(box) + // Forgotten again once it ends: a lookup during a slow pause may still have shown it running. + try { await call() } + finally { seen.forget(box) } + } + const blocks = provider.blocks + const run = blocks?.run + const files = blocks?.files + const pause = blocks?.pause + return { + ...provider, + describe: () => provider.describe(), + currentImage: () => provider.currentImage(), + create: async (spec, opts) => seen.observe(await provider.create(spec, opts)), + find: async query => seen.observe(await provider.find(query)), + inspect: async box => seen.observe(await provider.inspect(box)), + stop: (box, opts) => ending(box, () => provider.stop(box, opts)), + destroy: (box, opts) => ending(box, () => provider.destroy(box, opts)), + endpoint: (box, port, opts) => on(box, box => provider.endpoint(box, port, opts)), + ...(blocks ? { + blocks: { + ...blocks, + ...(run ? { + run: async (box: SandboxHandle, spec: RunSpec) => { + const result = await on(box, box => run(box, spec)) + seen.ran(box) + // A desktop that failed an operation is asked how it is before the next one. + if (result.exitCode !== 0 && DESKTOP_CALL.test(spec.command)) seen.desktop(box, false) + return result + }, + } : {}), + ...(files ? { + files: { + read: (box, path, opts) => on(box, box => files.read(box, path, opts)), + write: (box, path, bytes, opts) => on(box, box => files.write(box, path, bytes, opts)), + stat: (box, path, opts) => on(box, box => files.stat(box, path, opts)), + }, + } : {}), + ...(pause ? { pause: (box: SandboxHandle, opts: { timeoutMs: number }) => ending(box, () => pause(box, opts)) } : {}), + }, + } : {}), + } +}