feat: controls for unattended workers — usage-limit detection, end-attempt budgets, wall-time limits

packages/protocol/src/types.ts

@@ -440,6 +440,12 @@ export interface SessionInfo {
   turnQueued?: true;
   /** Remaining turns budget for autonomous mode (undefined = unbounded, 0 = exhausted). */
   turnsRemaining?: number;
+  /**
+   * An unattended worker (onBudgetExhausted "end_attempt") spent its
+   * autonomous tool budget: its tool calls are refused until a person sets a
+   * mode or sends a message, or its controller sets a new budget. Absent = not.
+   */
+  budgetExhausted?: true;
   /** Files pinned to the session — prepended to every turn's prompt. */
   pinnedFiles?: string[];
   /** SPIFFE/WIMSE URI of the primary session agent (falls back to anonymous:session:<id>). */

src/daemon/process-tree.ts

@@ -0,0 +1,179 @@
+/**
+ * Killing what a backend left running (Session.terminateBackend, supervisor.ts).
+ *
+ * Tearing a backend down kills ITS process; what that process had started —
+ * the shell a tool runs in, and whatever that shell runs — does not always go
+ * with it. Measured against the Claude Code CLI (SDK 0.3.281, a fake Messages
+ * API driving a real `Bash: sleep` call):
+ *
+ *   - abort (SIGTERM via the SDK) and Query.interrupt(): the CLI kills the
+ *     tool's shell. Nothing survives.
+ *   - the CLI SIGKILLed (the SDK's own escalation 5s after a SIGTERM the CLI
+ *     didn't act on): the tool's `sleep` SURVIVES, re-parented. Each Bash
+ *     call runs in its OWN process group (pgid = the shell's pid), so killing
+ *     the CLI's process group would not reach it either.
+ *   - a command that detaches itself (`setsid …`, `nohup … &`): survives every
+ *     path — it left the tree before anything here looks at it.
+ *
+ * codex and gemini-cli are stopped with a bare SIGKILL (jsonrpc-stdio.ts), so
+ * their running commands are in the second case on every teardown.
+ *
+ * So the stop is a TREE kill, not a group kill: snapshot every descendant of
+ * the backend's own process(es) before the teardown, and after it SIGKILL any
+ * of them still alive — with everything under them, including children they
+ * spawned after the snapshot. A survivor is matched by pid, command line AND
+ * start time, so a pid the OS reused in between is never touched. What this cannot reach, and does
+ * not claim to: anything that detached itself before the snapshot (only a
+ * cgroup or sandbox boundary bounds that), and a backend whose own process
+ * ids are unknown (qwen-code: @qwen-code/sdk spawns its CLI internally).
+ *
+ * Uses `ps` (Linux, macOS). Unavailable `ps` or Windows: empty snapshots, so
+ * terminateBackend degrades to the plain teardown.
+ */
+
+import { execFile } from "node:child_process";
+import { readFileSync } from "node:fs";
+
+export interface ProcessEntry {
+  pid: number;
+  ppid: number;
+  args: string;
+  /**
+   * When the process started, as an opaque comparable token — Linux:
+   * /proc/<pid>/stat field 22 (clock ticks since boot); macOS: ps `lstart`.
+   * With pid and command line, the identity a survivor must still match.
+   * Absent when it could not be read (the process is gone, or no /proc).
+   */
+  start?: string;
+}
+
+const PS_TIMEOUT_MS = 3_000;
+const DARWIN = process.platform === "darwin";
+
+/** `ps` with no COLUMNS (it would truncate `args` even with -ww on some systems). */
+function psEnv(): Record<string, string> {
+  const env: Record<string, string> = {};
+  for (const [k, v] of Object.entries(process.env)) if (v !== undefined && k !== "COLUMNS") env[k] = v;
+  return env;
+}
+
+/** Every process visible to this user: pid, parent pid, full command line (and start on macOS). */
+export function listProcesses(): Promise<ProcessEntry[]> {
+  if (process.platform === "win32") return Promise.resolve([]);
+  const columns = DARWIN ? "pid=,ppid=,lstart=,args=" : "pid=,ppid=,args=";
+  return new Promise((resolve) => {
+    execFile(
+      "ps",
+      ["-A", "-ww", "-o", columns],
+      { timeout: PS_TIMEOUT_MS, maxBuffer: 32 * 1024 * 1024, env: psEnv() },
+      (err, stdout) => {
+        if (err) {
+          resolve([]);
+          return;
+        }
+        resolve(parsePs(stdout, { lstart: DARWIN }));
+      },
+    );
+  });
+}
+
+/**
+ * Parse `ps -o pid=,ppid=[,lstart=],args=` output. `lstart` is five fields
+ * ("Sat Oct 24 10:08:43 2026") kept verbatim as the start token. Exported for tests.
+ */
+export function parsePs(stdout: string, opts: { lstart?: boolean } = {}): ProcessEntry[] {
+  const out: ProcessEntry[] = [];
+  const line = opts.lstart ? /^\s*(\d+)\s+(\d+)\s+(\S+\s+\S+\s+\S+\s+\S+\s+\S+)\s+(.*)$/ : /^\s*(\d+)\s+(\d+)\s+(.*)$/;
+  for (const text of stdout.split("\n")) {
+    const m = line.exec(text);
+    if (!m) continue;
+    if (opts.lstart) out.push({ pid: Number(m[1]), ppid: Number(m[2]), start: m[3]!.replace(/\s+/g, " "), args: m[4]!.trim() });
+    else out.push({ pid: Number(m[1]), ppid: Number(m[2]), args: m[3]!.trim() });
+  }
+  return out;
+}
+
+/** Linux: /proc/<pid>/stat field 22 (starttime). The comm field may hold spaces and parens, so count from the last ")". */
+export function procStartTime(pid: number): string | undefined {
+  try {
+    const stat = readFileSync(`/proc/${pid}/stat`, "utf8");
+    const fields = stat.slice(stat.lastIndexOf(")") + 2).split(" ");
+    // fields[0] is field 3 (state); field 22 is fields[19].
+    return fields[19];
+  } catch {
+    return undefined;
+  }
+}
+
+/** Fill `start` where ps didn't (Linux) — only for the entries that matter. */
+function withStart(entries: ProcessEntry[]): ProcessEntry[] {
+  if (process.platform !== "linux") return entries;
+  return entries.map((p) => (p.start !== undefined ? p : { ...p, ...optional(procStartTime(p.pid)) }));
+}
+function optional(start: string | undefined): { start?: string } {
+  return start === undefined ? {} : { start };
+}
+
+/** `roots` (those present) and all their descendants in `table`. Never this process. */
+export function treeOf(table: readonly ProcessEntry[], roots: readonly number[]): ProcessEntry[] {
+  const children = new Map<number, ProcessEntry[]>();
+  const byPid = new Map<number, ProcessEntry>();
+  for (const p of table) {
+    byPid.set(p.pid, p);
+    const list = children.get(p.ppid);
+    if (list) list.push(p);
+    else children.set(p.ppid, [p]);
+  }
+  const out: ProcessEntry[] = [];
+  const seen = new Set<number>([process.pid]);
+  const queue = roots.filter((pid) => byPid.has(pid) && !seen.has(pid));
+  for (const pid of queue) seen.add(pid);
+  while (queue.length > 0) {
+    const pid = queue.shift()!;
+    out.push(byPid.get(pid)!);
+    for (const child of children.get(pid) ?? []) {
+      if (seen.has(child.pid)) continue;
+      seen.add(child.pid);
+      queue.push(child.pid);
+    }
+  }
+  return out;
+}
+
+/** The process tree under `roots` right now, each entry with its start time. */
+export async function snapshotTree(roots: readonly number[]): Promise<ProcessEntry[]> {
+  if (roots.length === 0) return [];
+  return withStart(treeOf(await listProcesses(), roots));
+}
+
+/** Same process: pid, command line, and (when both are known) start time. */
+function sameProcess(then: ProcessEntry, now: ProcessEntry): boolean {
+  if (then.pid !== now.pid || then.args !== now.args) return false;
+  return then.start === undefined || now.start === undefined || then.start === now.start;
+}
+
+/**
+ * SIGKILL what is left of `snapshot`: every entry still running as the SAME
+ * process (pid + command line + start time — a reused pid is spared), and
+ * every descendant those have now, including children spawned after the
+ * snapshot was taken. One `ps` call. Returns the pids signalled.
+ */
+export async function killSurvivors(snapshot: readonly ProcessEntry[]): Promise<number[]> {
+  if (snapshot.length === 0) return [];
+  const table = await listProcesses();
+  const now = new Map(table.map((p) => [p.pid, p]));
+  const confirmed = withStart(snapshot.map((p) => now.get(p.pid)).filter((p): p is ProcessEntry => p !== undefined))
+    .filter((live) => sameProcess(snapshot.find((p) => p.pid === live.pid)!, live))
+    .map((p) => p.pid);
+  const killed: number[] = [];
+  for (const p of treeOf(table, confirmed)) {
+    if (p.pid === process.pid) continue;
+    try {
+      process.kill(p.pid, "SIGKILL");
+      killed.push(p.pid);
+    } catch {
+      // already gone, or not ours to signal
+    }
+  }
+  return killed;
+}

src/daemon/providers/acp/index.ts

@@ -38,7 +38,7 @@ import type {
   CatalogEntry,
 } from "../interface.js";
 import { renderHistorySeed, type CanonicalTurn, type HistorySeedResult } from "../canonical.js";
-import { buildGeminiCliEnv } from "../env.js";
+import { buildGeminiCliEnv, withExtraEnv } from "../env.js";
 import { StdioJsonRpcProcess } from "../jsonrpc-stdio.js";
 import { MEMORY_MCP_SERVER_NAME, type MemoryMcpMount } from "../../memory/mcp-http.js";
 import { BLACKBOARD_MCP_SERVER_NAME } from "../../blackboard/mcp-http.js";
@@ -89,6 +89,8 @@ export interface GeminiAcpProviderInit {
    */
   resolveAccount?: () => ResolvedAccount | null;
   onModels?: (models: ReadonlyArray<CatalogEntry>) => void;
+  /** Caller env for the child process (ProviderSessionInit.extraEnv), merged last via withExtraEnv. */
+  extraEnv?: Readonly<Record<string, string>>;
 }
 
 export class GeminiAcpProvider implements SessionProvider {
@@ -109,6 +111,7 @@ export class GeminiAcpProvider implements SessionProvider {
   #command: string;
   #argsPrefix: string[];
   #resolveAccount: (() => ResolvedAccount | null) | null;
+  readonly #extraEnv: Readonly<Record<string, string>> | undefined;
   #workspaceId: string;
   #memoryMcp: MemoryMcpMount | null;
   /** Role-scoped blackboard mount; token minted+revoked by the SessionManager. */
@@ -144,6 +147,7 @@ export class GeminiAcpProvider implements SessionProvider {
     this.#backingSessionId = init.initialBackingId;
     this.#command = init.command;
     this.#resolveAccount = init.resolveAccount ?? null;
+    this.#extraEnv = init.extraEnv;
     this.#argsPrefix = init.argsPrefix ?? [];
     this.#workspaceId = init.workspaceId ?? init.sessionId;
     this.#memoryMcp = init.memoryMcp ?? null;
@@ -248,6 +252,11 @@ export class GeminiAcpProvider implements SessionProvider {
     await this.teardown();
   }
 
+  async backendProcessIds(): Promise<number[]> {
+    const pid = this.#proc?.pid;
+    return pid ? [pid] : [];
+  }
+
   async teardown(): Promise<void> {
     this.#turnQueue?.close();
     this.#turnQueue = null;
@@ -315,7 +324,11 @@ export class GeminiAcpProvider implements SessionProvider {
         cwd: opts.workdir,
         // The account's own GEMINI_CLI_HOME (an overlay selecting its auth type).
         // ...and, for a thinking effort, an overlay of that whose settings set it (#417).
-        env: applyGeminiCliEffort(opts.effort, applyGeminiCliAccount(this.#resolveAccount?.() ?? null, buildGeminiCliEnv())),
+        // Then the caller's extra env, which can't override either (withExtraEnv).
+        env: withExtraEnv(
+          applyGeminiCliEffort(opts.effort, applyGeminiCliAccount(this.#resolveAccount?.() ?? null, buildGeminiCliEnv())),
+          this.#extraEnv,
+        ),
         onNotification: (method, params) => this.#onNotification(method, params),
         onServerRequest: (method, params) => this.#onServerRequest(method, params),
         onExit: ({ code, signal, stderrTail }) => {