feat(desktop): keep an agentenv computer up and drive it through REA's MCP tools (#9041)

* feat(desktop): keep an agentenv computer up and drive it through REA's MCP tools

- A Python helper keeps one agentenv computer up per build with
  run_bundle(keep=True) and reports its address, MCP URL, control token
  and sandbox id; it tears the computer down when stdin closes.
- AgentEnvComputer owns the helper, waits for readiness, holds one MCP
  session to the computer's computer_* tools, records through the
  computer-recording routes, and removes the sandbox's containers by
  label when agentenv cannot confirm teardown. The token never leaves it.
- start_build_computer starts a computer from a request (archive package,
  apt repository or .deb, opened on demand with REA in /opt/rea) or from
  a bundle folder; act_on_build_computer forwards screenshot, click,
  drag, scroll, type, key and wait and returns the driver's images
  unchanged; close_build_computer tears it down.

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

* fix(desktop): remove a build computer's container by name when agentenv cannot confirm teardown

agentenv's local provider runs an ordinary computer as agent-<sandbox id>
without the agentenv.sandbox label, so the label-only fallback found
nothing and reported a running computer as removed. The fallback now
lists containers by exact name and by label. The real test holds its
recording open long enough for a first frame, checks the container by
name, and runs rea inside the computer; the guide and tool descriptions
say the app opens from the Applications menu and that a screenshot after
opening a menu should come from a wait.

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

* fix(desktop): own failed build-computer starts, keep ids past oversized screenshots, bound recordings

- A start is owned from the helper's spawn. One that fails and whose
  teardown is unconfirmed stays held under a computer_id close can retry;
  closeAll refuses new starts, cancels and joins starts in progress, then
  closes every held computer.
- When the first screenshot is too large, the error carries REA's result
  with the computer_id, and start_build_computer takes screenshot region,
  format and quality.
- Recordings are refused before reading when the manifest's total exceeds
  the allowance, then fetched one segment at a time, each read bounded and
  checked against its size and sha256.
- The helper keeps the tail of its output (a new tail retention mode in
  ProviderProcessSupervisor), so a build's own error is not crowded out.
- A helper that reported no computer exits on its own before SIGTERM, so
  no phantom owner is kept.

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

* fix(desktop): bound a build computer's action log

Every action kept its full arguments for the computer's whole life, so a long
session grew REA's memory without limit. The log now keeps the latest 256
actions within 256 KiB, cuts each string to 1 KiB and each list to 64 items
(naming what it cut), and keeps the total and the last applied index of each
kind of action, so rules over the whole session hold after entries drop.

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

src/process/ProviderProcess.ts

@@ -206,6 +206,11 @@ export interface ProviderProcessSupervisorOptions {
   readonly captureStdout?: boolean;
   /** Retained output budget; overflow is drained and reported, not retained. */
   readonly maxDiagnosticBytes?: ProcessDiagnosticBudget;
+  /**
+   * Which bytes the budget keeps: the first ones (the default), or each
+   * stream's latest, where a failing command's last words are.
+   */
+  readonly diagnosticRetention?: "head" | "tail";
   /**
    * Single-line token ending at a non-ASCII-word byte or EOF, located before retention
    * applies, so its latest occurrence survives an exhausted budget.
@@ -833,13 +838,24 @@ export class ProviderProcessSupervisor {
       streamLimit === undefined
         ? bytes.byteLength
         : Math.max(0, streamLimit - capture.bytes);
-    const retained = Math.min(
-      bytes.byteLength,
-      aggregateRemaining,
-      streamRemaining,
-    );
-    if (retained < bytes.byteLength) this.#diagnosticTruncated = true;
-    if (retained > 0) capture.append(bytes.subarray(0, retained));
+    let truncated: boolean;
+    if (this.#options.diagnosticRetention === "tail") {
+      // The stream may give up its own earlier bytes to keep these.
+      const limit = Math.min(
+        capture.bytes + aggregateRemaining,
+        streamLimit ?? Number.MAX_SAFE_INTEGER,
+      );
+      truncated = capture.appendTail(bytes, limit) > 0;
+    } else {
+      const retained = Math.min(
+        bytes.byteLength,
+        aggregateRemaining,
+        streamRemaining,
+      );
+      if (retained > 0) capture.append(bytes.subarray(0, retained));
+      truncated = retained < bytes.byteLength;
+    }
+    if (truncated) this.#diagnosticTruncated = true;
     this.#options.onDiagnostic?.({
       type: "output",
       stream,
@@ -848,7 +864,7 @@ export class ProviderProcessSupervisor {
         stream === "stdout"
           ? this.#stdoutObservedBytes
           : this.#stderrObservedBytes,
-      ...(retained < bytes.byteLength ? { truncated: true } : {}),
+      ...(truncated ? { truncated: true } : {}),
     });
   }
 
@@ -868,6 +884,8 @@ export class ProviderProcessSupervisor {
 class ProcessOutputCapture {
   readonly #chunks: Buffer[] = [];
   #bytes = 0;
+  /** Observed bytes before the first retained one, which tail retention dropped. */
+  #start = 0;
 
   get bytes(): number {
     return this.#bytes;
@@ -876,6 +894,7 @@ class ProcessOutputCapture {
   reset(): void {
     this.#chunks.length = 0;
     this.#bytes = 0;
+    this.#start = 0;
   }
 
   append(chunk: Buffer): void {
@@ -883,6 +902,45 @@ class ProcessOutputCapture {
     this.#chunks.push(Buffer.from(chunk));
   }
 
+  /**
+   * Keep the latest `limit` bytes; returns how many observed bytes this
+   * dropped. A UTF-8 character the cut splits is dropped whole.
+   */
+  appendTail(chunk: Buffer, limit: number): number {
+    const before = this.#start;
+    const kept = chunk.subarray(Math.max(0, chunk.byteLength - limit));
+    this.append(kept);
+    this.#start += chunk.byteLength - kept.byteLength;
+    const cut = kept.byteLength < chunk.byteLength || this.#bytes > limit;
+    this.#dropFront(this.#bytes - limit, cut);
+    return this.#start - before;
+  }
+
+  /** Drop the first `count` retained bytes and, after a cut, the rest of a character it split. */
+  #dropFront(count: number, cut: boolean): void {
+    let remaining = Math.max(0, count);
+    for (
+      let first = this.#chunks[0];
+      first !== undefined;
+      first = this.#chunks[0]
+    ) {
+      let drop = Math.min(remaining, first.byteLength);
+      // UTF-8 continuation bytes are 0b10xxxxxx.
+      while (
+        cut &&
+        drop < first.byteLength &&
+        ((first[drop] ?? 0) & 0xc0) === 0x80
+      )
+        drop += 1;
+      if (drop === first.byteLength) this.#chunks.shift();
+      else if (drop > 0) this.#chunks[0] = Buffer.from(first.subarray(drop));
+      this.#bytes -= drop;
+      this.#start += drop;
+      remaining -= Math.min(remaining, drop);
+      if (drop < first.byteLength) return;
+    }
+  }
+
   snapshot(observedBytes: number): ProviderProcessSnapshot["stdout"] {
     return this.#snapshotBuffer(
       Buffer.concat(this.#chunks, this.#bytes),
@@ -894,10 +952,12 @@ class ProcessOutputCapture {
     start: number,
     end: number,
   ): { stream: ProviderProcessSnapshot["stdout"]; complete: boolean } {
-    const availableEnd = Math.min(end, this.#bytes);
-    const availableStart = Math.min(start, availableEnd);
+    // Positions are observed bytes; the retained ones start at #start.
+    const retainedEnd = this.#start + this.#bytes;
+    const availableEnd = Math.max(this.#start, Math.min(end, retainedEnd));
+    const availableStart = Math.min(Math.max(start, this.#start), availableEnd);
     const chunks: Buffer[] = [];
-    let offset = 0;
+    let offset = this.#start;
     for (const chunk of this.#chunks) {
       const chunkEnd = offset + chunk.byteLength;
       const from = Math.max(availableStart, offset);
@@ -909,7 +969,9 @@ class ProcessOutputCapture {
     const buffer = Buffer.concat(chunks, availableEnd - availableStart);
     return {
       stream: this.#snapshotBuffer(buffer, Math.max(0, end - start)),
-      complete: start === end || (start <= this.#bytes && end <= this.#bytes),
+      complete:
+        start === end ||
+        (start >= this.#start && start <= retainedEnd && end <= retainedEnd),
     };
   }