diff --git a/ee/packages/headless-computer/src/computer.ts b/ee/packages/headless-computer/src/computer.ts index 06cf971ec2..f833dc6c80 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 449fd74cae..5b250df4e0 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 0000000000..1d9d4bab9c --- /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 0000000000..550fe21885 --- /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)) } : {}), + }, + } : {}), + } +}