Files
reverse-engineering-anything/src/desktop/AgentEnvComputer.ts
T
df2d018b30 feat(desktop): read gzip documents as the document inside (#9111)
* fix(desktop): release a capture when its start is cancelled after launch

Cancelling start_build_capture during the opened app's screenshot left
the capture running, so the next start was refused. The capture is now
kept for the owner's cleanup, and the next start stops it first.

* fix(desktop): keep a published witness error and tag an app data refusal

A second failure to close a published witness returns the original error,
so its record and cleanup lease stay bound. A reset the app itself
refuses is an AppDataRefusal; only a reply that is not that refusal stays
unsettled. A missing child of a cut view is undecided with the same
description in both evaluators, and the shared cases cover it.

* fix(desktop): align a cut view's missing text with the Python check

A missing child of a truncated view is undecided for the same sentence in both evaluators, including the pointer the view does not contain.

* fix(desktop): keep one published witness cleanup across retries

A published record and its bound cleanup error stay together. A later close failure returns that error, and the unused path that built a fresh adapter error is gone.

* fix(desktop): keep a bundle task read inside the witness result

A cancelled, refused, or failed read of the bundle's task files returns the witness's Result, including a cleanup lease when a handle would not close.

* test(desktop): pass the settle timing when reading a witness bundle

The witness takes the settle timing before the bundle path, so the bundle-read tests keep the path in its own argument.

* fix(desktop): bound a guest file answer and split a long controller argument

A data/get whose base64 answer would exceed a quarter of the V8 heap is refused before it is read. Controller JSON that exceeds Linux's single-argument limit travels as several arguments the controller joins.

* fix(desktop): keep an underreported guest file from raising its read

A data/get answer is admitted only when its base64 body fits in the heap quarter after the copies held during parse. Decoded bytes larger than the computer reported are dropped.

* fix(desktop): grade shipped seeds and keep selected settings and sqlite logs

Seed checks follow the files the bundle ships, so an omitted lock is absent. An incomplete checkpoint is not reported as no change. A selected link keeps its spelling, and a database's wal and journal stay supporting state. Closing the fake computer stops a detached capture.

* fix(desktop): keep added-record checks off ids, dates, and joined timestamps

A position-named record's fields keep their paths, so an identifier, a stamped date, and text past the limit are not suggested checks. A whole-text check does not climb into an ancestor whose text joins a timestamp.

* fix(desktop): bound home fetches and keep unsettled edits readable

Split home-folder reads by the guest answer budget, keep a capacity error
from the capture output, and drop the superseded task-argument refusal.
Grade kept text per line, leave unretained dconf unread, and reserve
by-position notes for XML children. Map a linked database's sidecars
through the selected spelling and stop a detached capture only when its
start identity still owns the pid.

* fix(desktop): describe unretained dconf as unread

The key comparison treats a missing copy as unknown. The comment still
said that copy counted as an empty key list.

* test(desktop): keep a capture capacity error through endedCapture

An output past the guest answer ceiling stays the resource constraint,
with its heap recovery and non-retryable status, instead of a generic
capture-exit message.

* fix(desktop): keep nested save times and formatting runs out of checks

A field check drops save-time leaves before it joins the field, so a nested
timestamp is not part of the suggested text. A text check beside a save time
uses the paragraph when the diff names a formatting run.

* test(desktop): read paragraph pointers from checks that have them

* test(desktop): expect a large task to export intact

The 96 KiB refusal is gone. A prompt past that size reaches the computer
and comes back unchanged, instead of failing before export.

* fix(desktop): keep real edits that save-side heuristics were dropping

Repeated unnamed fields stay content checks. A formula that still has
text is not renumbered with its ExprID. One gained style attribute is
the edit; a repeated fill-in stays upkeep. Phone numbers and plain
clock times stay text, and a decimal is the element's whole text. An
older verifier matches on the task it states, not its evaluator source.

* fix(desktop): cover repeated records and keep timestamps out of checks

An added record whose words are already in the seed still gets the list's
text when that text fits. Readiness is a start reason, so a final check
names seed locks without matching the message. Field text split by a
timestamp stays separate, and a timestamp line is not offered as a guard.

* fix(desktop): compare a cell's joined decimal within tolerance

The whole-element check keeps numeric tolerance, so 437.80 and a long
float still match 437.8, including when a run splits the digits. A
different number or other text does not. Both evaluators read an
element's text for that comparison.

* fix(desktop): grade a number from a complete XML element only

A tolerance check joins an element's text only in an XML file or an XML
archive entry, and leaves that check undecided when the view is cut, so a
dropped descendant cannot turn the kept runs into a passing number. A JSON
object with the same shape stays the value itself. Equivalent decimals and a
split run on a whole cell still match.

* fix(desktop): accept an older grader only when its index helper agrees

The witness still matches a bundle whose evaluator is this REA's, or the older
index expression alone when no path segment is a leading-zero number. A
truncated verifier, a changed grader, or that older helper beside a segment
such as 01 no longer counts as the same task.

* fix(desktop): require the added record and leave save times out of list checks

A record a selector names is checked itself, so its words have to be in that
record. When an unnamed record only repeats words the seed already has, a
list that holds a save time is checked by the text around that time. A later
save of the same records still passes, and a missing or rewritten record fails.

* fix(desktop): judge the older index helper by pointer segments only

A leading zero in a file name or a URL does not change how the older index
helper reads a pointer. The witness still rejects that helper when a pointer
segment is a number such as 01.

* fix(desktop): do not suggest list text the seed already passes

A repeated record is shown by text the seed does not hold. The seed's own
text is not a check. A pointer inside a check value is not a pointer segment
for the older index helper.

* fix(desktop): keep a linked settings file in the folder the link points at

A selected ~/.config alias no longer moves MuseScore settings to that alias.
The seed keeps settings/MuseScore or Documents, which is the path the export
opens the note from.

* fix(desktop): split home fetches by answer size and keep a size refusal

A batch of tiny files stops before each path's envelope exceeds the host
answer budget. A capture output past that ceiling stays a resource error
instead of becoming a generic exit.

* fix(desktop): seed a selected link under the path the app uses

The selected spelling wins, so ~/.config/MuseScore is not rewritten to the
folder it points at and the note stays the one launched document. A database
wal and journal follow that database. A launch argument is respelled the same
way. Teardown signals a detached PID only when its start identity still matches.

* fix(desktop): keep a witness's actions and an incomplete checkpoint

A failed witness save retries the action log it first recorded. A deletion
left unconfirmed between the seed and the end is not reported as no change.
Windows packs npm through a shell, which can start npm.cmd.

* fix(desktop): read verifier pointers from the parsed checks

A brace inside a check value is not an object boundary. A leading-zero
pointer after that value still rejects the older index helper.

* fix(desktop): prefer the most specific home spelling and retarget launch files

A selected parent directory no longer hides a settings folder walked through
a link. A launch filename under that folder is replaced with the seeded
file's absolute path.

* fix(desktop): pack the build runtime through npm's JavaScript entry

npm pack and npm ci no longer spawn npm.cmd through a shell, so a temporary
directory whose path contains spaces stays one argument.

* fix(desktop): spell launch settings and keep a database log exact

A database log matches only that file, so a neighboring folder is not moved.
A launch option and the working directory use the same home spelling as the seed.

* fix(desktop): move a state database's logs with the database

The write-ahead log and journal follow the spelling the database received,
including a walked state link. A neighboring folder stays put. A separate
--config argument uses that spelling too.

* fix(desktop): keep a selected file ahead of a database-log guess

A file selected as itself keeps that spelling even when its name ends in
-journal. Relative launch arguments are resolved in the captured working
directory before that directory moves.

* fix(desktop): keep outside-home database logs and option values

A selected database outside the home keeps its write-ahead log. Launch
options are rewritten only when they name a captured file or settings path.

* fix(desktop): follow a database mapping and own limitation text

A selected database keeps its log when a parent folder moves or the file
sits outside the home. Trash and seed-lock tests compare the owning notice
with the paths, and launch options that are not captured paths stay as written.

* fix(desktop): retarget a settings folder that holds captured files

A nested profile directory is a captured path when a seeded file is inside
it. An option value that only resolves under the home stays as written.

* fix(desktop): keep a program word and a directory's neighbor

The program after env stays the program even when a captured file sits in a
folder of that name. A selected directory is not a database, so a neighboring
file keeps the parent folder's spelling.

* fix(desktop): record whether a selected path is a directory

Capture stores the walked path's kind. A neighboring file of a selected
directory keeps the parent folder's spelling even when that directory never
changes and its children are not content roots.

* fix(desktop): mark containing paths without a pairwise scan

A recorded path marks its known ancestors as directories in one walk up
the path, instead of comparing every recorded path with every other.

* fix(desktop): keep a selected folder that is named like a database log

A folder the caller selected stays a selected root when its name ends in
-journal or -wal. A generated log beside a database file still does not.

* fix(desktop): refuse a build command that exceeds the argument vector

Splitting one argument still leaves the whole command under Linux's E2BIG
limit. A request past that budget is refused before the process starts.

* fix(desktop): remap a captured path in an env assignment

The program word stays the program. A NAME=value assignment keeps its name
and follows a captured path value when that path's spelling moves.

* fix(desktop): keep oversized text out of draft guards

A value or line longer than the draft's text bound is not copied into a
guard, so one large frame cannot fill the task by itself.

* fix(desktop): keep an empty env value empty

An empty assignment is not a path. Resolving it named the working directory
and rewrote the value when that directory's spelling moved.

* fix(desktop): check build runtime export capability

* fix(desktop): provide complete runtime repack command

* fix(desktop): retain runtime probe failures

* test(desktop): identify capture independently of schema preflight

* fix(desktop): preserve document URLs and value formats

* fix(desktop): bound added checks and omit noisy text guards

* fix(desktop): preserve launch URLs and JSON draft values

* fix(desktop): recognize captured file URL documents

* fix(desktop): render captured file URLs for runtime home

* fix(desktop): distinguish phone numbers from save metadata

* fix(desktop): preserve launch URLs and user text

* fix(desktop): preserve local file URL authority

* fix(desktop): preserve local file URL authority

* fix(desktop): render captured file URLs at launch

* fix(desktop): classify numeric timestamps using field context

* fix(desktop): classify numeric timestamps using field context

* fix(desktop): retain context for generated draft values

* fix(desktop): retain context for generated draft values

* test(desktop): cover generated IDs in text guards

* test(desktop): cover generated IDs in text guards

* fix(desktop): isolate launcher URL encoding

* fix(desktop): isolate launcher URL encoding

* fix(desktop): isolate launcher URL encoding

* fix(desktop): retain clean text in noisy record fields

* style(desktop): format the noisy-record draft files with the pinned oxfmt

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test(desktop): move launch execution to boundary suite

* fix(desktop): redact build schema failure output

* refactor(desktop): normalize schema failure once

* fix(desktop): omit unsafe schema parser snippets

* fix(desktop): preserve draft change and selected path evidence

* test(desktop): group retained dconf note regressions

* fix(desktop): respect archive entry format in draft checks

* test(desktop): separate unowned-file draft note cases

* docs(desktop): clarify task checks before scratch export

* fix(desktop): respell selected paths in file URLs

* test(desktop): group captured document spelling cases

* fix(desktop): bound draft preview serialization

* fix(desktop): restore selected state aliases

* fix(desktop): respect entry format in draft noise

* test(desktop): match exported alias evidence at its JSON boundary

* style: fix desktop launch test formatting

* style: format native command path schema

* fix(desktop): map selected external launch paths

* style(desktop): format seed document regressions

* test(desktop): group launch URL regressions

* fix(desktop): omit redundant nested state aliases

* fix(desktop): compare added free text as a whole

* fix(desktop): check recompressed seeds as shipped

* fix(desktop): map captured launch paths once

* refactor: remove unused whole-text guard state

* refactor: stop passing unused text children

* test: align free-text checks with whole-note policy

* fix: validate recompressed desktop archive seeds

* style(desktop): format archive seed regression

* fix(desktop): redact runtime schema errors

* fix(desktop): normalize captured file URLs

* fix(desktop): report seeded files by capture path

* fix(desktop): retain startup readiness failure stage

* fix(desktop): retain JSON date checks

* fix(desktop): preserve archive formatting checks

* fix(desktop): preserve explicitly selected journal files

* fix(desktop): report build command transport limits

* style(desktop): format capture review fixes

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: morluto <76467478+morluto@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-09 20:48:08 +08:00

726 lines
26 KiB
TypeScript

import { createHash, randomUUID } from "node:crypto";
import { setTimeout as delay } from "node:timers/promises";
import { getHeapStatistics } from "node:v8";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/client";
import { z } from "zod";
import { AnalysisResourceConstraintError } from "../domain/analysisErrorCore.js";
import { jsonObjectSchema, type JsonValue } from "../domain/jsonValue.js";
import { StructuredContentMcpClient } from "../providers/StructuredContentMcpClient.js";
import { validateMcpHttpResponse } from "../providers/Utf8McpResponse.js";
import type {
ComputerHelper,
ComputerRelease,
} from "./AgentEnvComputerHelper.js";
import { drawnScreenshot } from "./DrawnScreenshot.js";
import { awaitHttpReply, type HttpReply } from "./HttpReply.js";
const READY_TIMEOUT_MS = 180_000;
const REQUEST_TIMEOUT_MS = 120_000;
const PROGRESS_MS = 5_000;
const RECORDINGS = ".agentenv-recordings";
/** A control route's answer: readiness, or a recording's manifest, near 1.5 MB for a day of 15 s segments. */
const CONTROL_RESPONSE_BYTES = 4 * 1024 * 1024;
/** What a data/get answer holds around one file's base64 bytes. */
const ENVELOPE_BYTES = 64 * 1024;
/** An exec answer: two streams of at most 1 MiB each, which JSON escaping can grow sixfold. */
const EXEC_RESPONSE_BYTES = 16 * 1024 * 1024;
/** The base64 answer for `bytes` of files, with each file's envelope. */
export const guestFileAnswerBytes = (bytes: number, files: number): number =>
Math.ceil(bytes / 3) * 4 + ENVELOPE_BYTES * Math.max(files, 1);
/** The most REA reads into memory for one answer of unknown size: a quarter of the V8 heap, as for recordings. */
const heapQuarter = () => Math.floor(getHeapStatistics().heap_size_limit / 4);
/** Where the data/get answer ceiling comes from. The guest's reported size does not change it. */
export const GUEST_FILE_ANSWER_SOURCE = "v8.heap_size_limit/4";
/**
* Copies of one answer held together while it is parsed: the raw chunks, the
* concatenated buffer, and the text. The decoded files replace the text
* afterward and are smaller than the base64 body.
*/
export const GUEST_FILE_ANSWER_COPIES = 3;
/** The host ceiling for one data/get answer, including base64 expansion and each file's envelope. */
export const guestFileAnswerCeiling = (): number => heapQuarter();
/** Wire bytes that fit in the ceiling once the parse's copies are counted. */
export const guestFileAnswerWorkingBytes = (answerCeiling: number): number =>
Math.floor(answerCeiling / GUEST_FILE_ANSWER_COPIES);
const guestFileAnswerError = (
fileBytes: number,
files: number,
answerCeiling: number,
remediation: string,
limits: Readonly<Record<string, JsonValue>>,
): AnalysisResourceConstraintError =>
new AnalysisResourceConstraintError(
"data/get",
"memory",
remediation,
{
guest_file_bytes: fileBytes,
file_count: files,
answer_ceiling_bytes: answerCeiling,
answer_working_bytes: guestFileAnswerWorkingBytes(answerCeiling),
answer_copies: GUEST_FILE_ANSWER_COPIES,
ceiling_source: GUEST_FILE_ANSWER_SOURCE,
...limits,
},
{ remediationAction: remediation, retryable: false },
);
/**
* Refuse a data/get whose base64 answer, counted once for each copy the parse
* holds, would exceed the host ceiling. A larger report is refused before the
* answer is read. The reported size does not raise the ceiling.
*/
export const guestFileAnswerRefusal = (
fileBytes: number,
files: number,
answerCeiling: number,
): AnalysisResourceConstraintError | undefined => {
const answerBytes = guestFileAnswerBytes(fileBytes, files);
const workingBytes = guestFileAnswerWorkingBytes(answerCeiling);
if (answerBytes <= workingBytes) return undefined;
const remediation = `This data/get answer is limited to ${String(workingBytes)} wire bytes, one quarter of this process's V8 heap divided by ${String(GUEST_FILE_ANSWER_COPIES)} copies held while the answer is parsed (source ${GUEST_FILE_ANSWER_SOURCE}; the heap quarter is ${String(answerCeiling)} bytes). The computer reported ${String(fileBytes)} file bytes across ${String(files)} paths, which expand to ${String(answerBytes)} bytes once each file is base64 and its envelope is counted. The reported size does not raise the ceiling. Restart REA with a larger Node --max-old-space-size so this heap quarter covers the answer.`;
return guestFileAnswerError(fileBytes, files, answerCeiling, remediation, {
answer_bytes: answerBytes,
});
};
/** One decoded part of a data/get answer. */
export interface GuestFilePart {
readonly path: string | undefined;
readonly bytes: Buffer;
}
/**
* The decoded files, or a refusal when they contain more bytes than the
* computer reported. An underreported size must not admit the larger body.
*/
export const retainGuestFiles = (
parts: readonly GuestFilePart[],
reportedBytes: number,
files: number,
answerCeiling: number,
):
| { readonly ok: true; readonly files: ReadonlyMap<string, Buffer> }
| { readonly ok: false; readonly error: AnalysisResourceConstraintError } => {
let decodedBytes = 0;
const retained = new Map<string, Buffer>();
for (const part of parts) {
decodedBytes += part.bytes.length;
if (part.path !== undefined) retained.set(part.path, part.bytes);
}
if (decodedBytes <= reportedBytes) return { ok: true, files: retained };
const remediation = `The computer reported ${String(reportedBytes)} file bytes, but the data/get answer decoded to ${String(decodedBytes)} bytes. The extra bytes were not kept. A reported size is only a ceiling when the answer is no larger.`;
return {
ok: false,
error: guestFileAnswerError(
reportedBytes,
files,
answerCeiling,
remediation,
{ decoded_file_bytes: decodedBytes },
),
};
};
const readinessSchema = z.object({
ready: z.boolean(),
checks: z
.array(
z.object({
name: z.string(),
ok: z.boolean(),
detail: z.string().nullish(),
final: z.boolean().optional(),
}),
)
.default([]),
});
const recordingId = z.string().regex(/^rec-[0-9a-f]+$/u);
const recordingStartSchema = z.looseObject({
recording_id: recordingId,
started_at: z.string(),
});
const recordingManifestSchema = z.looseObject({
recording_id: recordingId,
segments: z.array(
z.looseObject({
index: z.int().nonnegative(),
path: z.string(),
bytes: z.int().nonnegative(),
sha256: z.string().regex(/^[0-9a-f]{64}$/u),
}),
),
});
/** An ended recording's manifest, as `computer-recording/v1` returns it. */
export type RecordingManifest = z.infer<typeof recordingManifestSchema>;
const dataGetSchema = z.union([
z.object({
result: z.object({
parts: z.array(
z.looseObject({
file: z.looseObject({ bytes: z.string().optional() }).optional(),
metadata: z
.record(z.string(), z.looseObject({ path: z.string().optional() }))
.optional(),
}),
),
}),
}),
z.object({ error: z.looseObject({ message: z.string() }) }),
]);
const rpcSchema = z.union([
z.object({ result: z.looseObject({}) }),
z.object({ error: z.looseObject({ message: z.string() }) }),
]);
const execSchema = z.object({
exit_code: z.int().nullable(),
stdout: z.string(),
stderr: z.string(),
truncated: z.boolean(),
timed_out: z.boolean(),
});
/** A command's outcome through `computer-exec/v1`, its output without the control token. */
export type ComputerCommandResult = z.infer<typeof execSchema>;
const cardSchema = z.looseObject({
children_environments: z
.array(z.looseObject({ name: z.string(), url: z.string().optional() }))
.optional(),
});
/** How long a route may take and how much of its answer REA reads. */
interface RequestBounds {
readonly limit?: { readonly bytes: number; readonly what: string };
/** The deadline of a route that answers only once its work ends, which REA waits out past fetch's 300 s for headers. */
readonly timeoutMs?: number;
}
/** A driver tool's answer as REA forwards it: its content blocks unchanged and its structured result. */
export interface DriverResult {
readonly isError: boolean;
readonly content: readonly Readonly<Record<string, JsonValue>>[];
readonly structured: Readonly<Record<string, JsonValue>> | null;
}
/** A response body read only while it stays within `maxBytes`. */
const readBounded = async (
response: HttpReply,
maxBytes: number,
what: string,
): Promise<string> => {
const chunks: Uint8Array[] = [];
let bytes = 0;
if (response.body !== null)
for await (const chunk of response.body) {
bytes += chunk.byteLength;
if (bytes > maxBytes)
throw new Error(`${what} is more than ${String(maxBytes)} bytes.`);
chunks.push(chunk);
}
return Buffer.concat(chunks, bytes).toString("utf8");
};
/** Why a computer did not come up. Readiness is a fact, not a phrase in the message. */
export type ComputerStartReason = "handshake" | "not_ready" | "unusable";
/** Why a computer did not come up, with what the caller can do about it. */
export class ComputerStartFailure extends Error {
readonly reason: ComputerStartReason;
constructor(
message: string,
readonly remediation: string | undefined,
options?: ErrorOptions & { readonly reason?: ComputerStartReason },
) {
super(message, options);
this.name = "ComputerStartFailure";
this.reason = options?.reason ?? "unusable";
}
}
/**
* Whether `cause` is a start that failed because the computer never became
* ready. The helper's own sentence is classified once, where it is read.
*/
export const computerFailedOnReadiness = (
cause: unknown,
): cause is ComputerStartFailure =>
cause instanceof ComputerStartFailure && cause.reason === "not_ready";
/** agentenv's probe says the computer never became ready. That sentence is outside REA. */
const helperWasNotReady = (error: string | undefined) =>
error !== undefined && /\bnot ready\b|wasn't ready|didn't boot/u.test(error);
/** Every few seconds while a computer comes up, report the phase and agentenv's latest line. */
const progressHeartbeat = (
helper: ComputerHelper,
phase: { readonly name: string },
progress: ((message: string) => void) | undefined,
) => {
if (progress === undefined) return undefined;
const startedAt = Date.now();
return setInterval(() => {
const line = helper
.output()
.stderr.trimEnd()
.split("\n")
.at(-1)
?.slice(-200);
progress(
`${phase.name} (${String(Math.round((Date.now() - startedAt) / 1000))} s)${line === undefined || line === "" ? "" : `: ${line}`}`,
);
}, PROGRESS_MS);
};
const connectDriver = async (
mcpUrl: string,
signal: AbortSignal,
): Promise<StructuredContentMcpClient> => {
const client = new StructuredContentMcpClient({
name: "rea-build-computer",
version: "1",
});
const transport: StreamableHTTPClientTransport =
new StreamableHTTPClientTransport(new URL(mcpUrl), {
requestInit: { redirect: "error" },
fetch: async (input, init) =>
validateMcpHttpResponse(await fetch(input, init), () => {
void transport.close().catch(() => undefined);
}),
});
await client.connect(transport, { timeout: REQUEST_TIMEOUT_MS, signal });
return client;
};
/**
* One agentenv computer kept up for REA: the helper that holds it, one MCP
* session to its driver, and its control token, which never leaves this
* object.
*/
/**
* The app refused a data method. The refusal is the app's own error, not a
* lost reply. Callers must use this tag rather than the message text.
*/
export class AppDataRefusal extends Error {
readonly _tag = "AppDataRefusal" as const;
constructor(
readonly method: "data/get" | "data/reset",
readonly appMessage: string,
) {
super(`${method} failed: ${appMessage}`);
this.name = "AppDataRefusal";
}
}
export class AgentEnvComputer {
readonly sandboxId: string;
readonly #baseUrl: string;
readonly #token: string | null;
readonly #helper: ComputerHelper;
readonly #client: StructuredContentMcpClient;
private constructor(parts: {
readonly sandboxId: string;
readonly baseUrl: string;
readonly token: string | null;
readonly helper: ComputerHelper;
readonly client: StructuredContentMcpClient;
}) {
this.sandboxId = parts.sandboxId;
this.#baseUrl = parts.baseUrl;
this.#token = parts.token;
this.#helper = parts.helper;
this.#client = parts.client;
}
/**
* Connect to the computer a helper keeps up: wait for its handshake and
* readiness, open an MCP session to its driver and take its first frame
* once the desktop has drawn, reporting progress throughout. A failure
* throws a `ComputerStartFailure`, or the signal's reason when cancelled;
* the helper's owner releases it either way.
*/
static async connect(
helper: ComputerHelper,
signal: AbortSignal,
progress?: (message: string) => void,
): Promise<{
readonly computer: AgentEnvComputer;
readonly frame: DriverResult;
}> {
const phase = { name: "agentenv is deploying the computer" };
const heartbeat = progressHeartbeat(helper, phase, progress);
try {
return await AgentEnvComputer.#connect(helper, signal, phase);
} finally {
clearInterval(heartbeat);
}
}
static async #connect(
helper: ComputerHelper,
signal: AbortSignal,
phase: { name: string },
): ReturnType<typeof AgentEnvComputer.connect> {
const handshake = await helper.handshake(signal);
if (handshake === undefined || "error" in handshake)
throw new ComputerStartFailure(
`agentenv did not bring the computer up: ${handshake?.error ?? "its helper ended without a handshake"}.`,
"Read the helper's output; `agent-env run <bundle> --task smoke --no-emulate` reproduces the deploy.",
{
reason: helperWasNotReady(handshake?.error)
? "not_ready"
: "handshake",
},
);
phase.name = "waiting for the computer's readiness checks";
let client: StructuredContentMcpClient | undefined;
let stage: "readiness" | undefined;
try {
client = await connectDriver(handshake.mcp_url, signal);
const computer = new AgentEnvComputer({
sandboxId: handshake.sandbox_id,
baseUrl: handshake.base_url.replace(/\/+$/u, ""),
token: handshake.control_token,
helper,
client,
});
stage = "readiness";
await computer.#waitReady(signal);
stage = undefined;
phase.name = "waiting for the desktop to draw";
const frame = await drawnScreenshot(
(retry) => computer.callTool("computer_screenshot", {}, retry),
signal,
);
return { computer, frame };
} catch (cause: unknown) {
await client?.close().catch(() => undefined);
if (signal.aborted) throw signal.reason;
if (cause instanceof ComputerStartFailure) throw cause;
throw new ComputerStartFailure(
`The computer came up but REA could not use it: ${cause instanceof Error ? cause.message : String(cause)}`,
undefined,
{ cause, reason: stage === "readiness" ? "not_ready" : "unusable" },
);
}
}
/** Call one of the driver's `computer_*` tools on REA's MCP session. */
async callTool(
name: string,
args: Readonly<Record<string, JsonValue>>,
signal: AbortSignal,
): Promise<DriverResult> {
const result = await this.#client.callTool(
{ name, arguments: args },
{ timeout: REQUEST_TIMEOUT_MS, signal },
);
const structured = jsonObjectSchema.safeParse(result.structuredContent);
return {
isError: result.isError === true,
content: result.content.map((block) => jsonObjectSchema.parse(block)),
structured: structured.success ? structured.data : null,
};
}
/** Start a recording under the caller's request id; repeating the id answers the same recording. */
async startRecording(requestId: string, signal?: AbortSignal) {
return recordingStartSchema.parse(
await this.#request(
"/agentenv/ext/computer-recording/start",
{ request_id: requestId },
signal,
),
);
}
/** End a recording; its manifest lists the segments the computer keeps until teardown. */
async endRecording(
id: string,
signal?: AbortSignal,
): Promise<RecordingManifest> {
return recordingManifestSchema.parse(
await this.#request(
"/agentenv/ext/computer-recording/end",
{ recording_id: id },
signal,
),
);
}
/**
* Fetch an ended recording's segments through the driver's data plane,
* one request per segment, each read within its manifest size and checked
* against it. A recording larger than `maxBytes`, by default a quarter of
* the V8 heap as for retained Evidence, is refused before any read.
*/
async recordingSegments(
manifest: RecordingManifest,
signal?: AbortSignal,
maxBytes = Math.floor(getHeapStatistics().heap_size_limit / 4),
): Promise<readonly { readonly path: string; readonly bytes: Buffer }[]> {
const total = manifest.segments.reduce(
(sum, segment) => sum + segment.bytes,
0,
);
if (total > maxBytes)
throw new Error(
`Recording ${manifest.recording_id} holds ${String(total)} bytes, more than the ${String(maxBytes)} REA reads into memory for one recording; end recordings sooner so each stays smaller.`,
);
const segments: { readonly path: string; readonly bytes: Buffer }[] = [];
for (const segment of manifest.segments)
segments.push({
path: segment.path,
bytes: await this.#segment(segment, signal),
});
return segments;
}
async #segment(
segment: RecordingManifest["segments"][number],
signal: AbortSignal | undefined,
): Promise<Buffer> {
const path = `${RECORDINGS}/${segment.path}`;
const answer = dataGetSchema.parse(
await this.#request(
"/agentenv",
{
jsonrpc: "2.0",
id: segment.index + 1,
method: "data/get",
params: { paths: [path] },
},
signal,
{
limit: {
// Base64 plus the JSON-RPC envelope and the part's metadata.
bytes: Math.ceil(segment.bytes / 3) * 4 + ENVELOPE_BYTES,
what: `The computer's answer for recording segment ${segment.path}`,
},
},
),
);
if ("error" in answer)
throw new Error(
`data/get of recording segment ${segment.path} failed: ${answer.error.message}`,
);
const encoded = answer.result.parts.find(
(part) => Object.values(part.metadata ?? {})[0]?.path === path,
)?.file?.bytes;
const bytes =
encoded === undefined ? undefined : Buffer.from(encoded, "base64");
if (
bytes?.length !== segment.bytes ||
createHash("sha256").update(bytes).digest("hex") !== segment.sha256
)
throw new Error(
`Recording segment ${segment.path} is missing or differs from its manifest.`,
);
return bytes;
}
/**
* The files under paths of the computer user's home, by home path, through
* the driver's `data/get`; its answer is read within `bytes` of files.
*/
async homeFiles(
paths: readonly string[],
bytes: number,
signal?: AbortSignal,
): Promise<ReadonlyMap<string, Buffer>> {
const ceiling = guestFileAnswerCeiling();
const refusal = guestFileAnswerRefusal(bytes, paths.length, ceiling);
if (refusal !== undefined) throw refusal;
const answer = dataGetSchema.parse(
await this.#request(
"/agentenv",
{ jsonrpc: "2.0", id: 1, method: "data/get", params: { paths } },
signal,
{
limit: {
bytes: Math.min(
guestFileAnswerBytes(bytes, paths.length),
guestFileAnswerWorkingBytes(ceiling),
),
what: `The computer's answer for ${paths.join(", ")}`,
},
},
),
);
if ("error" in answer)
throw new Error(
`data/get failed: ${this.redacted(answer.error.message)}`,
);
const retained = retainGuestFiles(
answer.result.parts.flatMap((part) => {
const encoded = part.file?.bytes;
return encoded === undefined
? []
: [
{
path: Object.values(part.metadata ?? {})[0]?.path,
bytes: Buffer.from(encoded, "base64"),
},
];
}),
bytes,
paths.length,
ceiling,
);
if (!retained.ok) throw retained.error;
return retained.files;
}
/** Run a command in the computer's desktop session as its user; output is cut at the computer's cap. */
async exec(
argv: readonly string[],
timeoutS: number,
signal?: AbortSignal,
): Promise<ComputerCommandResult> {
const result = execSchema.parse(
await this.#request(
"/agentenv/ext/computer-exec/run",
{ argv, request_id: randomUUID(), timeout_s: timeoutS },
signal,
{
limit: {
bytes: EXEC_RESPONSE_BYTES,
what: `The computer's answer to ${argv[0] ?? "a command"}`,
},
timeoutMs: (timeoutS + 30) * 1000,
},
),
);
return {
...result,
stdout: this.redacted(result.stdout),
stderr: this.redacted(result.stderr),
};
}
/**
* The parts the computer's one app answers a data method with: `data/get`,
* read within a quarter of the V8 heap since it carries the app's files, or
* `data/reset`.
*/
async appData(
method: "data/get" | "data/reset",
signal?: AbortSignal,
): Promise<readonly unknown[]> {
const card = cardSchema.parse(
await this.#request("/.well-known/agent-env.json", undefined, signal),
);
const [app, ...others] = card.children_environments ?? [];
if (app === undefined || others.length > 0)
throw new Error(
`The computer lists ${String(card.children_environments?.length ?? 0)} apps; REA reads a computer with one app.`,
);
const path = new URL(app.url ?? `/svc/${app.name}/agentenv`, this.#baseUrl)
.pathname;
const answer = rpcSchema.parse(
await this.#request(
path,
{ jsonrpc: "2.0", id: 1, method, params: {} },
signal,
{
limit: {
bytes: heapQuarter(),
what: `The ${app.name} app's ${method} answer`,
},
},
),
);
if ("error" in answer)
throw new AppDataRefusal(method, this.redacted(answer.error.message));
return method === "data/get"
? z.object({ parts: z.array(z.unknown()) }).parse(answer.result).parts
: [];
}
/** Wait until the computer's readiness checks pass, as after a reset relaunches its app. */
async waitUntilReady(signal: AbortSignal): Promise<void> {
await this.#waitReady(signal);
}
/** End REA's MCP session and bring the computer down; a failed release can be retried. */
async close(): Promise<ComputerRelease> {
await this.#client.close().catch(() => undefined);
return this.#helper.release();
}
async #waitReady(signal: AbortSignal): Promise<void> {
const deadline = Date.now() + READY_TIMEOUT_MS;
for (;;) {
const status = readinessSchema.parse(
await this.#request(
"/agentenv/ext/computer-ready/status",
undefined,
signal,
),
);
if (status.ready) return;
const waiting = status.checks.filter((check) => !check.ok);
const failed = waiting.find((check) => check.final === true);
if (failed !== undefined || Date.now() > deadline)
throw new ComputerStartFailure(
failed === undefined
? `The computer was not ready within ${String(READY_TIMEOUT_MS / 1000)} s; waiting on ${waiting.map((check) => check.name).join(", ")}.`
: `The computer's ${failed.name} check failed: ${failed.detail ?? "no detail"}.`,
"Read the helper's output; `agent-env run <bundle> --task smoke --no-emulate` reproduces the deploy.",
{ reason: "not_ready" },
);
await delay(500, undefined, { signal });
}
}
/** `text` with the control token replaced, as every answer REA reads from the computer's routes and commands. */
redacted(text: string): string {
return this.#token === null
? text
: text.replaceAll(this.#token, "[control token]");
}
/** A framework-only route of the computer, with its control token; its answer is read within `bounds.limit`. */
async #request(
path: string,
body: unknown,
signal: AbortSignal | undefined,
bounds: RequestBounds = {},
): Promise<unknown> {
const limit = bounds.limit ?? {
bytes: CONTROL_RESPONSE_BYTES,
what: `The computer's answer to ${path}`,
};
const timeout = AbortSignal.timeout(bounds.timeoutMs ?? REQUEST_TIMEOUT_MS);
const url = `${this.#baseUrl}${path}`;
const init = {
method: body === undefined ? "GET" : "POST",
headers: {
...(this.#token === null
? {}
: { authorization: `Bearer ${this.#token}` }),
...(body === undefined ? {} : { "content-type": "application/json" }),
},
...(body === undefined ? {} : { body: JSON.stringify(body) }),
};
const deadline =
signal === undefined ? timeout : AbortSignal.any([signal, timeout]);
const response: HttpReply =
bounds.timeoutMs === undefined
? await fetch(url, { ...init, redirect: "error", signal: deadline })
: await awaitHttpReply(new URL(url), init, deadline);
const text = await readBounded(response, limit.bytes, limit.what);
if (response.status < 200 || response.status > 299)
throw new Error(
`The computer answered ${path} with HTTP ${String(response.status)}: ${this.redacted(text.slice(0, 512))}`,
);
return JSON.parse(text) as unknown;
}
}