/**
 * Teammate spawner — spawns named resident child Pi processes in RPC mode.
 *
 * A teammate is a long-lived worker: it receives prompts on its control
 * stream (stdin), streams JSON events on stdout, and suspends between wake
 * ups without consuming tokens. The harness delivers new prompts via
 * deliverPrompt (idle wake-up) and steering lines via sendWorkerSteer
 * (mid-turn delivery). Wake-up sequences are uncapped: no turn-count or
 * wall-clock limit terminates a working teammate; anomalies surface as
 * leader notifications instead.
 */

import { randomUUID } from "node:crypto";
import { WORKER_BUILTIN_TOOLS } from "./worker-tools.ts";
export { WORKER_BUILTIN_TOOLS } from "./worker-tools.ts";
import * as fs from "node:fs";
import { tmpdir } from "node:os";
import * as path from "node:path";
import { fileURLToPath } from "node:url";
import {
  DEFAULT_TERMINATION_GRACE_MS,
  extractTextContent,
  resolvePiCli,
  spawnPiChild,
  terminateChildProcess,
} from "@fradser/pi-kit";
import type { ChildProcess } from "node:child_process";
import type { WorkerUsage } from "./types.ts";
import { writeWorkContext } from "./work-context.ts";
import type { SessionContext } from "@earendil-works/pi-coding-agent";

const OUTPUT_CAP = 16_000;
/** A single JSONL record and an unterminated record may not exceed this many bytes. */
export const MAX_JSONL_LINE_BYTES = 8 * 1024 * 1024;
/** Live text, thinking, or tool arguments are bounded per assistant message. */
export const MAX_TURN_OUTPUT_BYTES = 16 * 1024 * 1024;
/** Diagnostic tail only; this is not a resident-process lifetime quota. */
const DIAGNOSTIC_TAIL_BYTES = 64 * 1024;

export interface WorkerProcessResult {
  pid: number;
  exitCode: number | null;
  signal: NodeJS.Signals | null;
  stdout: string;
  stderr: string;
  usage?: WorkerUsage;
}

/** Live state extracted from a teammate's RPC output stream. */
export interface WorkerProgressUpdate {
  text: string;
  activeTool?: string;
  liveThinking?: string;
  /** Assistant turns observed in the current wake-up sequence. */
  turns: number;
  /** True after the current sequence's final response. */
  finalResponse?: boolean;
  /** True once the stream delivered recognized model activity (text, thinking,
   *  tool-call, or tool execution events). Independent of usage totals so
   *  providers that omit usage are not misclassified as silent. */
  modelOutputSeen?: boolean;
  /** A prompt-less child acknowledged native startup readiness. */
  ready?: boolean;
  /** Lifetime accumulated usage parsed from message_end events. */
  usage?: WorkerUsage;
  controlError?: string;
}

/** A crashed teammate is one that closed without a normal zero exit. */
export function isCleanExit(result: Pick<WorkerProcessResult, "exitCode" | "signal">): boolean {
  return result.exitCode === 0 && result.signal === null;
}

/** Tools every teammate receives regardless of its role definition. */
export const WORKER_CAPABILITY_TOOLS: readonly string[] = [
  "agent_event",
  "work",
];

/** Effective tool allowlist for one teammate: the role's requested tools plus
 *  the capability set, deduplicated in request order. Roles without a tools
 *  field get exactly the capability set — leaders should see that narrow
 *  grant before the first wake so missing read/bash is obvious at spawn time. */
export function resolveWorkerTools(requested?: string[]): string[] {
  const requestedOnly = (requested ?? []).filter((tool) => !WORKER_CAPABILITY_TOOLS.includes(tool));
  return [...new Set([...requestedOnly, ...WORKER_CAPABILITY_TOOLS])];
}

/** Pi built-in tool ids a bare teammate process registers. The child runs
 *  --no-extensions, so this list (plus the capability set) is the complete
 *  grantable universe; anything else is silently dropped by the child's
 *  --tools allowlist filter. */
// Canonical IDs are shared with public schema and guidance in worker-tools.ts.

/** Every tool id a teammate can actually receive: pi built-ins plus the capability set. */
export const WORKER_TOOL_UNIVERSE: readonly string[] = [...WORKER_BUILTIN_TOOLS, ...WORKER_CAPABILITY_TOOLS];

/** Requested tool ids outside the teammate universe — exactly the ones the
 *  child's --tools allowlist would silently drop. */
export function unknownWorkerTools(requested?: string[]): string[] {
  return [...new Set((requested ?? []).filter((tool) => !WORKER_TOOL_UNIVERSE.includes(tool)))];
}

// ── Process registry ──────────────────────────────────────────────

/** Live children by teammate name — powers steering, prompting, and shutdown. */
const workers = new Map<string, ChildProcess>();
const closedWorkers = new WeakSet<ChildProcess>();

function observeWorkerClose(name: string, child: ChildProcess): void {
  child.once("close", () => {
    closedWorkers.add(child);
    if (workers.get(name) === child) workers.delete(name);
  });
}

/** True only after Node has emitted the child process close event. */
export function isWorkerCloseObserved(name: string): boolean {
  const child = workers.get(name);
  if (child) return closedWorkers.has(child);
  // Not registered anymore: either never spawned or already unregistered by close.
  return true;
}

export { terminateChildProcess };

/** Terminate a living teammate and wait until its child process has closed. */
export type TeammateTermination = { outcome: "missing" } | { outcome: "closed" } | { outcome: "unconfirmed" };

export async function terminateTeammate(name: string, graceMs = DEFAULT_TERMINATION_GRACE_MS): Promise<TeammateTermination> {
  const child = workers.get(name);
  if (!child) return { outcome: "missing" };
  const closed = closedWorkers.has(child) || await terminateChildProcess(child, graceMs);
  return { outcome: closed ? "closed" : "unconfirmed" };
}

export async function terminateAllTeammates(graceMs = DEFAULT_TERMINATION_GRACE_MS): Promise<Array<{ name: string; confirmedClosed: boolean }>> {
  const entries = [...workers.entries()];
  return Promise.all(entries.map(async ([name, child]) => ({
    name,
    confirmedClosed: closedWorkers.has(child) || await terminateChildProcess(child, graceMs),
  })));
}

// ── Control stream ────────────────────────────────────────────────

interface PendingControlResponse {
  child: ChildProcess;
  command: string;
  timeout: ReturnType<typeof setTimeout>;
  resolve: (success: boolean) => void;
}

type PendingFreshDelivery = {
  message: string;
  streamingBehavior: "steer" | "followUp";
};

interface PendingFreshAssignment {
  child: ChildProcess;
  message: string;
  resetId?: string;
  queued: PendingFreshDelivery[];
  resolve: (success: boolean) => void;
}

const CONTROL_RESPONSE_TIMEOUT_MS = 5_000;
const pendingControlResponses = new Map<string, PendingControlResponse>();
const pendingFreshAssignments = new Map<string, PendingFreshAssignment>();

function writeToControlStream(child: ChildProcess, line: unknown): boolean {
  if (!child.stdin || child.stdin.destroyed || !child.stdin.writable) return false;
  child.stdin.write(`${JSON.stringify(line)}\n`);
  return true;
}

/**
 * Deliver a new wake-up prompt to an idle teammate's control stream.
 * This starts a fresh assistant sequence in the child process.
 */
export function deliverPrompt(name: string, message: string): boolean {
  const child = workers.get(name);
  if (!child) return false;
  const sent = writeToControlStream(child, { type: "prompt", id: randomUUID(), message, streamingBehavior: "followUp" });
  if (sent) beginSequence(name);
  return sent;
}

/** Per-name stream states so prompt delivery can reset sequence boundaries. */
const streamStates = new Map<string, StreamState>();

function beginSequence(name: string): void {
  const state = streamStates.get(name);
  if (state) {
    state.finalResponse = false;
    state.text = "";
    state.thinking = "";
    clearActiveTools(state);
    state.turnBytes = 0;
    state.outputLimitError = undefined;
  }
  baselines.set(name, streamTurns.get(name) ?? 0);
}

function sendOrQueueDuringFreshReset(
  name: string,
  message: string,
  streamingBehavior: PendingFreshDelivery["streamingBehavior"],
): boolean {
  const child = workers.get(name);
  if (!child) return false;
  const pending = pendingFreshAssignments.get(name);
  if (pending?.child === child) {
    pending.queued.push({ message, streamingBehavior });
    return true;
  }
  return writeToControlStream(child, { type: "prompt", id: randomUUID(), message, streamingBehavior });
}

/** Send a mid-turn steer to a working teammate; no peer mailbox is involved. */
export function sendWorkerSteer(name: string, message: string): boolean {
  return sendOrQueueDuringFreshReset(name, message, "steer");
}

/** Peer traffic yields to leader steering without depending on roster freshness. */
export function sendWorkerFollowUp(name: string, message: string): boolean {
  return sendOrQueueDuringFreshReset(name, message, "followUp");
}

function settleFreshAssignment(name: string, success: boolean): void {
  const pending = pendingFreshAssignments.get(name);
  if (!pending) return;
  pendingFreshAssignments.delete(name);
  if (pending.resetId) {
    const control = pendingControlResponses.get(pending.resetId);
    if (control) clearTimeout(control.timeout);
    pendingControlResponses.delete(pending.resetId);
  }
  pending.resolve(success);
}

function deliverFreshPrompts(name: string, pending: PendingFreshAssignment): boolean {
  if (!writeToControlStream(pending.child, {
    type: "prompt",
    id: randomUUID(),
    message: pending.message,
    streamingBehavior: "followUp",
  })) return false;
  beginSequence(name);
  for (const delivery of pending.queued) {
    if (!writeToControlStream(pending.child, {
      type: "prompt",
      id: randomUUID(),
      message: delivery.message,
      streamingBehavior: delivery.streamingBehavior,
    })) break;
  }
  return true;
}

function startFreshAssignmentReset(name: string): void {
  const pending = pendingFreshAssignments.get(name);
  const state = streamStates.get(name);
  if (!pending || pending.resetId || state?.finalResponse !== true || workers.get(name) !== pending.child) return;
  const id = randomUUID();
  pending.resetId = id;
  const timeout = setTimeout(() => settleFreshAssignment(name, false), CONTROL_RESPONSE_TIMEOUT_MS);
  timeout.unref?.();
  pendingControlResponses.set(id, {
    child: pending.child,
    command: "new_session",
    timeout,
    resolve: (reset) => {
      const current = pendingFreshAssignments.get(name);
      if (current !== pending || !reset || workers.get(name) !== pending.child) {
        settleFreshAssignment(name, false);
        return;
      }
      settleFreshAssignment(name, deliverFreshPrompts(name, pending));
    },
  });
  if (!writeToControlStream(pending.child, { type: "new_session", id })) {
    settleFreshAssignment(name, false);
  }
}

/** True while a new Assignment Attempt is waiting for reset or prompt delivery. */
export function isFreshAssignmentPending(name: string): boolean {
  return pendingFreshAssignments.has(name);
}

/** Queue a fresh Pi session and deliver a new Assignment Attempt after settlement. */
export function deliverFreshAssignment(name: string, message: string): Promise<boolean> {
  const child = workers.get(name);
  if (!child || pendingFreshAssignments.has(name)) return Promise.resolve(false);
  return new Promise((resolve) => {
    pendingFreshAssignments.set(name, { child, message, queued: [], resolve });
    startFreshAssignmentReset(name);
  });
}

// ── Stream parsing ────────────────────────────────────────────────

type JsonEvent = {
  id?: string;
  type?: string;
  command?: string;
  success?: boolean;
  error?: string;
  data?: { cancelled?: boolean; isStreaming?: boolean; isCompacting?: boolean; pendingMessageCount?: number };
  toolCallId?: string;
  toolName?: string;
  args?: unknown;
  assistantMessageEvent?: {
    type?: string;
    delta?: string;
  };
  message?: {
    role?: string;
    stopReason?: string;
    content?: Array<{ type?: string; text?: string }>;
    usage?: {
      input?: number;
      output?: number;
      cacheRead?: number;
      cacheWrite?: number;
      totalTokens?: number;
      cost?: { total?: number };
    };
  };
};

interface StreamState {
  text: string;
  thinking: string;
  toolcallArgs: string;
  activeTool?: string;
  activeTools: Map<string, string>;
  /** Lifetime count of completed assistant messages. */
  turns: number;
  finalResponse?: boolean;
  /** Set by any recognized model/stream activity; never by a bare empty
   *  message_end artifact. The stall classifier uses this, not usage. */
  modelOutputSeen?: boolean;
  usage?: WorkerUsage;
  controlError?: string;
  turnBytes: number;
  outputLimitError?: string;
}

function createStreamState(): StreamState {
  return { text: "", thinking: "", toolcallArgs: "", activeTools: new Map(), turns: 0, turnBytes: 0 };
}

function clearActiveTools(state: StreamState): void {
  state.toolcallArgs = "";
  state.activeTool = undefined;
  state.activeTools.clear();
}

function toolExecutionLabel(toolName: string | undefined, args: unknown): string {
  const serialized = typeof args === "string" ? args : JSON.stringify(args ?? {});
  return truncate(toolcallLabel(serialized) ?? toolName ?? "tool", OUTPUT_CAP);
}

/** Human-readable label from a partially streamed tool-call argument JSON. */
function toolcallLabel(rawArgs: string): string | undefined {
  const trimmed = rawArgs.trim();
  if (!trimmed.startsWith("{")) return undefined;
  try {
    const args = JSON.parse(trimmed) as Record<string, unknown>;
    const command = args.command;
    if (typeof command === "string" && command.trim()) return `bash: ${normalizeInline(command)}`;
    const filePath = args.path;
    if (typeof filePath === "string" && filePath.trim()) return `file: ${normalizeInline(path.basename(filePath.trim()))}`;
    const subject = args.subject;
    if (typeof subject === "string" && subject.trim()) return `message: ${normalizeInline(subject)}`;
    const query = args.query;
    if (typeof query === "string" && query.trim()) return `search: ${normalizeInline(query)}`;
    const to = args.to;
    if (typeof to === "string" && to.trim() && typeof args.subject === "string") return `send: ${normalizeInline(args.subject)}`;
  } catch {
    // Incomplete JSON mid-stream — retry on the next delta.
  }
  return undefined;
}

function normalizeInline(text: string): string {
  return text.replace(/\s+/g, " ").trim();
}

function appendTurnText(state: StreamState, field: "text" | "thinking" | "toolcallArgs", delta: string): boolean {
  const nextBytes = state.turnBytes + Buffer.byteLength(delta, "utf8");
  if (nextBytes > MAX_TURN_OUTPUT_BYTES) {
    state.outputLimitError = `Resident worker ${field} exceeded ${MAX_TURN_OUTPUT_BYTES} bytes in one turn.`;
    return false;
  }
  state.turnBytes = nextBytes;
  state[field] += delta;
  return true;
}

function applyStreamLine(state: StreamState, line: string, child?: ChildProcess): boolean {
  if (!line.trim()) return false;
  let event: JsonEvent;
  try {
    event = JSON.parse(line) as JsonEvent;
  } catch {
    return false;
  }
  if (event.type === "response" && event.id) {
    const pending = pendingControlResponses.get(event.id);
    if (pending && child === pending.child && pending.command === event.command) {
      clearTimeout(pending.timeout);
      pendingControlResponses.delete(event.id);
      pending.resolve(event.success === true && event.data?.cancelled !== true
        && (pending.command !== "get_state" || (event.data?.isStreaming === false
          && event.data?.isCompacting !== true && (event.data?.pendingMessageCount ?? 0) === 0)));
    }
  }
  if (event.type === "response" && event.success === false) {
    state.controlError = truncate(`RPC ${event.command ?? "command"} rejected: ${event.error ?? "unknown error"}`, 1000);
    return true;
  }
  if (event.type === "agent_settled") {
    state.finalResponse = true;
    clearActiveTools(state);
    return true;
  }
  if (event.type === "agent_start") {
    state.finalResponse = false;
    state.text = "";
    state.thinking = "";
    clearActiveTools(state);
    state.turnBytes = 0;
    state.outputLimitError = undefined;
    return true;
  }
  if (event.type === "tool_execution_start") {
    state.modelOutputSeen = true;
    state.activeTools.set(event.toolCallId ?? `tool-${state.activeTools.size}`, toolExecutionLabel(event.toolName, event.args));
    state.activeTool = [...state.activeTools.values()].at(-1);
    return true;
  }
  if (event.type === "tool_execution_end") {
    state.modelOutputSeen = true;
    if (event.toolCallId) state.activeTools.delete(event.toolCallId);
    else state.activeTools.clear();
    state.activeTool = [...state.activeTools.values()].at(-1);
    return true;
  }
  if (event.type !== "message_update") {
    if (event.type !== "message_end" || event.message?.role !== "assistant") return false;
    state.turns++;
    clearActiveTools(state);
    state.thinking = "";
    const parts = extractTextContent(event.message.content, "");
    if (Buffer.byteLength(parts, "utf8") > MAX_TURN_OUTPUT_BYTES) {
      state.outputLimitError = `Resident worker message exceeded ${MAX_TURN_OUTPUT_BYTES} bytes.`;
      return true;
    }
    if (parts.trim()) state.text = parts;
    state.turnBytes = 0;
    // Usage stays diagnostics: only streamed content counts as model output,
    // so input-only or failed responses cannot bypass the zero-output tier.
    const u = event.message.usage;
    if (parts.trim()) state.modelOutputSeen = true;
    if (u) {
      state.usage = {
        input: (state.usage?.input ?? 0) + (u.input ?? 0),
        output: (state.usage?.output ?? 0) + (u.output ?? 0),
        cacheRead: (state.usage?.cacheRead ?? 0) + (u.cacheRead ?? 0),
        cacheWrite: (state.usage?.cacheWrite ?? 0) + (u.cacheWrite ?? 0),
        totalTokens: (state.usage?.totalTokens ?? 0) + (u.totalTokens ?? 0),
        cost: (state.usage?.cost ?? 0) + (u.cost?.total ?? 0),
      };
    }
    return true;
  }
  const sub = event.assistantMessageEvent;
  if (!sub) return false;
  switch (sub.type) {
    case "text_delta":
      state.modelOutputSeen = true;
      state.activeTool = undefined;
      appendTurnText(state, "text", sub.delta ?? "");
      return true;
    case "thinking_delta":
      state.modelOutputSeen = true;
      state.activeTool = undefined;
      appendTurnText(state, "thinking", sub.delta ?? "");
      return true;
    case "toolcall_start":
      state.modelOutputSeen = true;
      clearActiveTools(state);
      return true;
    case "toolcall_delta": {
      state.modelOutputSeen = true;
      appendTurnText(state, "toolcallArgs", sub.delta ?? "");
      const label = toolcallLabel(state.toolcallArgs);
      if (label) state.activeTool = label;
      return true;
    }
    case "toolcall_end":
      state.modelOutputSeen = true;
      // Execution events own the activity label until the result arrives.
      clearActiveTools(state);
      return true;
    default:
      return false;
  }
}

/** Parse the final assistant text and accumulated usage from captured stdout. */
export function parseTeammateOutput(stdout: string): { text: string; usage?: WorkerUsage } {
  const state = createStreamState();
  for (const line of stdout.split("\n")) applyStreamLine(state, line);
  return { text: state.text.trim(), usage: state.usage };
}

function truncate(text: string, cap: number): string {
  if (text.length <= cap) return text;
  return `${text.slice(0, cap)}\n... [truncated ${text.length - cap} chars]`;
}

// ── Resident spawn ────────────────────────────────────────────────

/** Turn-count baseline per teammate at its most recent delivered prompt. */
const baselines = new Map<string, number>();
const streamTurns = new Map<string, number>();

export interface ResidentSpawnOptions {
  /** Teammate name — the registry key for steering, prompting, and shutdown. */
  workerName: string;
  /** Optional kickoff prompt; omit to let the teammate idle immediately. */
  description?: string;
  model?: string;
  /** Execution-tool allowlist; capability tools are always appended. */
  tools?: string[];
  /** Leader session's thinking level, forwarded as the child's default. */
  thinking?: string;
  /** Detached active-context snapshot; undefined starts with fresh history. */
  context?: SessionContext["messages"];
  env?: Record<string, string | undefined>;
  cwd?: string;
  onUpdate?: (update: WorkerProgressUpdate) => void;
  onExit: (result: WorkerProcessResult) => void;
  onError?: (error: Error) => void;
}

export interface SpawnedResident {
  pid: number;
}

const WORKER_EXTENSION = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "index.ts");

/**
 * Spawn one resident child Pi process in RPC mode. Returns immediately with
 * the child pid; outcomes arrive via onExit / onError. An empty description
 * spawns an idle teammate that waits for its first delivered prompt.
 */
export function spawnResident(options: ResidentSpawnOptions): SpawnedResident | { error: string } {
  const cli = resolvePiCli();
  const args: string[] = [
    ...cli.args,
    "--mode", "rpc",
    "--no-extensions",
    "--extension", WORKER_EXTENSION,
  ];
  if (options.model) args.push("--model", options.model);
  if (options.thinking) args.push("--thinking", options.thinking);
  args.push("--tools", resolveWorkerTools(options.tools).join(","));

  let tempDir: string | undefined;
  const cleanupTempDir = () => {
    if (!tempDir) return;
    try {
      fs.rmSync(tempDir, { recursive: true, force: true });
    } catch {
      // Temporary cleanup must not mask the worker outcome.
    } finally {
      tempDir = undefined;
    }
  };

  let child: ChildProcess;
  try {
    if (options.context !== undefined) {
      tempDir = fs.mkdtempSync(path.join(tmpdir(), "pi-work-context-"));
      args.push("--session", writeWorkContext(options.cwd ?? process.cwd(), tempDir, options.context));
    } else {
      args.push("--no-session");
    }
    child = spawnPiChild(cli.command, args, {
      cwd: options.cwd,
      env: { ...process.env, ...options.env },
      stdio: ["pipe", "pipe", "pipe"],
    });
  } catch (error) {
    cleanupTempDir();
    return { error: error instanceof Error ? error.message : String(error) };
  }
  if (options.description && options.description.trim()) {
    writeToControlStream(child, { type: "prompt", id: randomUUID(), message: options.description });
    baselines.set(options.workerName, 0);
  }
  observeWorkerClose(options.workerName, child);
  workers.set(options.workerName, child);

  const stdoutChunks: string[] = [];
  const stderrChunks: string[] = [];
  const streamState = createStreamState();
  streamStates.set(options.workerName, streamState);
  let stdoutBuffer = "";
  let failureReason: string | undefined;
  let settled = false;

  const emitProgress = () => options.onUpdate?.({
    text: truncate(streamState.text, OUTPUT_CAP),
    activeTool: streamState.activeTool,
    liveThinking: truncate(streamState.thinking, OUTPUT_CAP),
    turns: Math.max(0, streamState.turns - (baselines.get(options.workerName) ?? 0)),
    finalResponse: streamState.finalResponse,
    modelOutputSeen: streamState.modelOutputSeen,
    usage: streamState.usage,
    controlError: streamState.controlError,
  });

  let termination: Promise<boolean> | undefined;
  const failWorker = (reason: string) => {
    failureReason ??= reason;
    stdoutBuffer = "";
    streamState.outputLimitError ??= reason;
    streamState.controlError = reason;
    emitProgress();
    termination ??= terminateChildProcess(child);
  };

  child.stdout?.on("data", (chunk: Buffer) => {
    if (failureReason) return;
    const text = chunk.toString();
    stdoutBuffer += text;
    const lines = stdoutBuffer.split("\n");
    stdoutBuffer = lines.pop() ?? "";
    let changed = false;
    for (const line of lines) {
      if (Buffer.byteLength(line, "utf8") > MAX_JSONL_LINE_BYTES) {
        failWorker(`Resident worker JSONL line exceeded ${MAX_JSONL_LINE_BYTES} bytes.`);
        return;
      }
      changed = applyStreamLine(streamState, line, child) || changed;
      if (streamState.finalResponse) startFreshAssignmentReset(options.workerName);
      if (streamState.outputLimitError) {
        failWorker(streamState.outputLimitError);
        return;
      }
    }
    if (Buffer.byteLength(stdoutBuffer, "utf8") > MAX_JSONL_LINE_BYTES) {
      failWorker(`Resident worker JSONL line exceeded ${MAX_JSONL_LINE_BYTES} bytes.`);
      return;
    }
    appendCapped(stdoutChunks, text, DIAGNOSTIC_TAIL_BYTES);
    streamTurns.set(options.workerName, streamState.turns);
    if (changed) emitProgress();
    if (!failureReason) streamState.controlError = undefined;
  });
  child.stderr?.on("data", (chunk: Buffer) => {
    if (failureReason) return;
    appendCapped(stderrChunks, chunk.toString(), DIAGNOSTIC_TAIL_BYTES);
  });

  child.on("error", (error) => {
    if (child.pid === undefined) cleanupTempDir();
    settled = true;
    // Keep the registry entry until close is observed so shutdown diagnostics
    // can distinguish an error/exit code from a confirmed close event.
    options.onError?.(error);
  });

  child.on("close", (code, signal) => {
    cleanupTempDir();
    if (workers.get(options.workerName) === child) workers.delete(options.workerName);
    streamStates.delete(options.workerName);
    const fresh = pendingFreshAssignments.get(options.workerName);
    if (fresh?.child === child) settleFreshAssignment(options.workerName, false);
    for (const [id, pending] of pendingControlResponses) {
      if (pending.child !== child) continue;
      clearTimeout(pending.timeout);
      pendingControlResponses.delete(id);
      pending.resolve(false);
    }
    baselines.delete(options.workerName);
    streamTurns.delete(options.workerName);
    // A spawn failure was already reported via onError (Node fires error then
    // close) — do not double-report through onExit.
    if (settled) return;
    const failed = failureReason !== undefined;
    const parsed = failed ? { text: "", usage: undefined } : parseTeammateOutput(stdoutChunks.join(""));
    options.onExit({
      pid: child.pid ?? 0,
      exitCode: code === 0 && failed ? 1 : code,
      signal,
      stdout: failed ? "" : truncate(parsed.text, OUTPUT_CAP),
      stderr: truncate([failureReason, stderrChunks.join("").trim()].filter(Boolean).join("\n"), OUTPUT_CAP),
      usage: parsed.usage,
    });
  });

  if (!options.description?.trim()) {
    const id = randomUUID();
    const timeout = setTimeout(() => {
      pendingControlResponses.delete(id);
      streamState.controlError = "Resident startup readiness was not acknowledged.";
      emitProgress();
    }, CONTROL_RESPONSE_TIMEOUT_MS);
    timeout.unref?.();
    pendingControlResponses.set(id, {
      child,
      command: "get_state",
      timeout,
      resolve: (ready) => {
        if (workers.get(options.workerName) !== child) return;
        if (!ready) {
          streamState.controlError = "Resident startup readiness acknowledgement was rejected or not idle.";
          emitProgress();
          return;
        }
        streamState.finalResponse = true;
        options.onUpdate?.({ text: "", turns: 0, finalResponse: true, ready: true });
        startFreshAssignmentReset(options.workerName);
      },
    });
    if (!writeToControlStream(child, { type: "get_state", id })) {
      clearTimeout(timeout);
      pendingControlResponses.delete(id);
      options.onError?.(new Error("Cannot request resident startup readiness."));
    }
  }
  return { pid: child.pid ?? 0 };
}

function appendCapped(chunks: string[], chunk: string, cap: number): void {
  chunks.push(chunk);
  let total = chunks.reduce((sum, value) => sum + Buffer.byteLength(value, "utf8"), 0);
  while (total > cap && chunks.length > 1) {
    total -= Buffer.byteLength(chunks.shift() ?? "", "utf8");
  }
  if (total > cap && chunks.length === 1) {
    chunks[0] = Buffer.from(chunks[0], "utf8").subarray(-cap).toString("utf8");
  }
}