import { spawn } from "node:child_process";
import { createHash } from "node:crypto";
import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { StringDecoder } from "node:string_decoder";
import { tmpdir } from "node:os";
import { basename, join } from "node:path";
import { fileURLToPath } from "node:url";
import { sanitizeTerminalText } from "./job-status.js";
import type { LaunchOptions } from "./launch-options.js";
import { CAPTURED_TEXT_MAX_BYTES, truncateUtf8 } from "./output.js";
import { getChildLaunchToolAllowlist } from "./profile-capabilities.js";
import {
  REPORT_DEDUPE_MAX_BYTES,
  REPORT_DEDUPE_MAX_ITEMS,
  REPORT_ID_MAX_BYTES,
  REPORT_MAX_BYTES,
  REPORT_RECORD_FIXED_BYTES,
  type AccessMode,
  type AgentProfile,
  type ReportKind,
  type SubagentReport,
  type TextTruncation,
  type UsageStats,
} from "./types.js";

const SESSION_READINESS_TIMEOUT_MS = 10_000;
const SESSION_CLOSE_GRACE_MS = 5_000;
const CHILD_SHUTDOWN_COMMAND = "simple-subagent-shutdown";
export const RESERVED_CHILD_SHUTDOWN_MESSAGE = "This message is reserved for internal use.";

/** Matches the command only where Pi would parse a leading slash command, with optional arguments. */
export const isReservedChildShutdownMessage = (message: string): boolean => {
  const trimmed = message.trimStart();
  const prefix = `/${CHILD_SHUTDOWN_COMMAND}`;
  if (!trimmed.startsWith(prefix)) return false;
  const next = trimmed[prefix.length];
  return next === undefined || /\s/u.test(next);
};

const SESSION_STDIN_FAILURE_MESSAGE = "Child Pi RPC stdin failed";
const SESSION_CHANNEL_FAILURE_MESSAGE = "Child Pi RPC channel failed";
/** Includes room for a 50 KiB assistant message, bounded diagnostic metadata, and its JSONL envelope. */
const RPC_WIRE_RECORD_MAX_BYTES = 128 * 1024;
const SAFE_TOOL_NAMES = new Set(["read", "grep", "find", "ls", "bash", "edit", "write", "subagent_report"]);
const CHILD_REPORTING_GUIDANCE = [
  "Child reporting protocol:",
  "- Use subagent_report with kind=progress only at meaningful milestones, not routine chatter.",
  "- When blocked, call subagent_report with kind=help_request alone, then stop/settle.",
  "- Do not combine a help_request with other tool calls.",
].join("\n");
let nextCommandId = 0;

const hashedReportId = (reportId: string): string =>
  `sha256:${createHash("sha256").update(reportId).digest("hex")}`;

export const normalizeReportId = (reportId: string): string => {
  if (Buffer.byteLength(reportId, "utf8") > REPORT_ID_MAX_BYTES) return hashedReportId(reportId);
  const safe = sanitizeTerminalText(reportId).replace(/\s+/gu, " ").trim();
  return safe === reportId ? safe : hashedReportId(reportId);
};

export const reportRecordBytes = (report: SubagentReport): number =>
  Buffer.byteLength(report.sessionId, "utf8")
  + Buffer.byteLength(report.reportId, "utf8")
  + Buffer.byteLength(report.message, "utf8")
  + REPORT_RECORD_FIXED_BYTES;

/** Insertion-ordered recent IDs. Old IDs may be accepted again after either bound evicts them. */
export class BoundedReportIds {
  private readonly ids = new Map<string, number>();
  private bytes = 0;

  add(reportId: string): boolean {
    if (this.ids.has(reportId)) return false;
    const bytes = Buffer.byteLength(reportId, "utf8");
    while (
      this.ids.size >= REPORT_DEDUPE_MAX_ITEMS
      || this.bytes + bytes > REPORT_DEDUPE_MAX_BYTES
    ) {
      const oldest = this.ids.entries().next().value as [string, number] | undefined;
      if (!oldest) break;
      this.ids.delete(oldest[0]);
      this.bytes -= oldest[1];
    }
    this.ids.set(reportId, bytes);
    this.bytes += bytes;
    return true;
  }

  clear(): void {
    this.ids.clear();
    this.bytes = 0;
  }
}

type ChildCommand = "prompt" | "steer" | "abort";

interface PendingCommand {
  command: ChildCommand;
  startsNewGeneration?: boolean;
  timer: unknown;
  resolve: () => void;
  reject: (error: Error) => void;
}

interface StdoutParserState {
  decoder: StringDecoder;
  buffer: string;
  wireBytes: number;
  discardUntilLineFeed: boolean;
}

interface ParsedStdoutRecord {
  value?: unknown;
  malformed?: true;
}

interface AssistantOutputCapture {
  text: string;
  originalBytes: number;
  exhausted: boolean;
  previewPublished: boolean;
}

interface AssistantTelemetryCapture {
  usage: UsageStats;
  model?: string;
}

interface GenerationCapture {
  active: boolean;
  output: string;
  outputTruncation?: TextTruncation;
  stopReason?: string;
  errorMessage?: string;
  errorTruncation?: TextTruncation;
  malformedEventCount: number;
}

interface StderrCapture {
  text: string;
  originalBytes: number;
  truncation?: TextTruncation;
}

const emptyUsage = (): UsageStats => ({ input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, turns: 0 });

const asRecord = (value: unknown): Record<string, unknown> | undefined =>
  value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : undefined;

const asNumber = (value: unknown): number => typeof value === "number" ? value : 0;

const assistantModelIdentity = (assistant: Record<string, unknown>): string | undefined => {
  const provider = assistant.provider;
  const model = assistant.responseModel ?? assistant.model;
  if (typeof provider !== "string" || provider.length === 0 || typeof model !== "string" || model.length === 0) return undefined;

  // Incomplete records keep the last verified identity instead of coercing untrusted values.
  return `${provider}/${model}`;
};

const normalizeReport = (value: unknown): { kind: ReportKind; message: string } | undefined => {
  const payload = asRecord(value);
  if (!payload || Object.keys(payload).length !== 2) return undefined;
  if ((payload.kind !== "progress" && payload.kind !== "help_request") || typeof payload.message !== "string" || payload.message.length === 0) return undefined;

  return { kind: payload.kind, message: truncateUtf8(payload.message, REPORT_MAX_BYTES).text };
};

export interface SessionOpenOptions {
  cwd: string;
  profile: AgentProfile;
  accessMode: AccessMode;
  launchOptions: LaunchOptions;
}

export interface SessionExit {
  exitCode: number;
  signal?: NodeJS.Signals;
  expected: boolean;
  error?: string;
  errorTruncation?: TextTruncation;
  stderr: string;
  stderrTruncation?: TextTruncation;
}

export interface SessionResult {
  output: string;
  stderr: string;
  usage: UsageStats;
  model?: string;
  stopReason?: string;
  errorMessage?: string;
  errorTruncation?: TextTruncation;
  malformedEventCount: number;
  outputTruncation?: TextTruncation;
  stderrTruncation?: TextTruncation;
}

export type SessionEvent =
  | { type: "output"; text: string; truncation?: TextTruncation }
  | { type: "progress"; text: string; timestamp: number }
  | { type: "telemetry"; usage: UsageStats; model?: string }
  | { type: "report"; reportId: string; kind: ReportKind; message: string; timestamp: number }
  | { type: "settled"; result: SessionResult };

export interface RunningSubagentSession {
  prompt(message: string, startsNewGeneration?: boolean): Promise<void>;
  steer(message: string): Promise<void>;
  abort(): Promise<void>;
  close(): Promise<void>;
  subscribe(listener: (event: SessionEvent) => void): () => void;
  readonly closed: Promise<SessionExit>;
}

export interface SessionRunner {
  open(options: SessionOpenOptions): Promise<RunningSubagentSession>;
}

/** Internal signal to lifecycle owners that a failed open still has a child awaiting close. */
export class TrackedSessionOpenError extends Error {
  override readonly name = "TrackedSessionOpenError";

  constructor(message: string, readonly processExit: Promise<void>) {
    super(message);
  }
}

/** Internal signal to lifecycle owners that command acceptance is unknown and the child is terminating. */
export class SessionChannelFailureError extends Error {
  override readonly name = "SessionChannelFailureError";

  constructor(kind: "command_timeout" | "stdin_failure") {
    super(kind === "stdin_failure" ? SESSION_STDIN_FAILURE_MESSAGE : SESSION_CHANNEL_FAILURE_MESSAGE);
  }
}

export interface SessionSpawnOptions {
  cwd: string;
  detached: boolean;
  shell: false;
  stdio: ["pipe", "pipe", "pipe"];
}

export interface SessionProcessStream {
  on(event: "data", listener: (data: Buffer) => void): unknown;
  removeListener(event: "data", listener: (data: Buffer) => void): unknown;
}

export interface SessionProcessInput {
  write(data: string, callback?: (error?: Error | null) => void): boolean;
  on(event: "error", listener: (error: Error) => void): unknown;
  removeListener(event: "error", listener: (error: Error) => void): unknown;
}

export interface SpawnedSessionProcess {
  readonly pid?: number;
  stdin: SessionProcessInput;
  stdout: SessionProcessStream;
  stderr: SessionProcessStream;
  on(event: "close", listener: (code: number | null, signal: NodeJS.Signals | null) => void): unknown;
  on(event: "error", listener: (error: Error) => void): unknown;
  removeListener(event: "close", listener: (code: number | null, signal: NodeJS.Signals | null) => void): unknown;
  removeListener(event: "error", listener: (error: Error) => void): unknown;
  kill(signal: NodeJS.Signals): boolean;
}

export interface PiRpcSessionRunnerDependencies {
  spawnProcess?(command: string, args: readonly string[], options: SessionSpawnOptions): SpawnedSessionProcess;
  platform?: NodeJS.Platform;
  signalProcessTree?(target: number, signal: NodeJS.Signals): boolean | Promise<boolean>;
  setTimer?(callback: () => void, delay: number): unknown;
  clearTimer?(timer: unknown): void;
  fileExists?(path: string): boolean;
  now?(): number;
}

const defaultSpawnProcess = (command: string, args: readonly string[], options: SessionSpawnOptions): SpawnedSessionProcess =>
  spawn(command, args, options) as unknown as SpawnedSessionProcess;

const defaultSignalProcessTree = (
  platform: NodeJS.Platform,
  target: number,
  signal: NodeJS.Signals,
): boolean | Promise<boolean> => {
  if (platform !== "win32") {
    try {
      return process.kill(target, signal);
    } catch {
      return false;
    }
  }

  return new Promise<boolean>((resolve) => {
    let settled = false;
    const settle = (delivered: boolean) => {
      if (settled) return;
      settled = true;
      resolve(delivered);
    };
    try {
      const args = ["/PID", String(target), "/T"];
      if (signal === "SIGKILL") args.push("/F");
      const taskkill = spawn("taskkill", args, { shell: false, stdio: "ignore", windowsHide: true });
      taskkill.once("error", () => settle(false));
      taskkill.once("close", (code) => settle(code === 0));
    } catch {
      settle(false);
    }
  });
};

const getPiInvocation = (args: string[], fileExists: (path: string) => boolean): { command: string; args: string[] } => {
  const currentScript = process.argv[1];
  if (currentScript && !currentScript.startsWith("/$bunfs/root/") && fileExists(currentScript)) {
    return { command: process.execPath, args: [currentScript, ...args] };
  }

  const execName = basename(process.execPath).toLowerCase();
  if (!/^(node|bun)(\.exe)?$/.test(execName)) return { command: process.execPath, args };
  return { command: "pi", args };
};

const writePrompt = (profile: AgentProfile): { dir: string; path: string } => {
  let dir: string | undefined;
  try {
    dir = mkdtempSync(join(tmpdir(), "simple-subagents-"));
    const safeName = profile.name.replace(/[^\w.-]+/g, "_");
    const path = join(dir, `prompt-${safeName}.md`);
    writeFileSync(path, `${profile.systemPrompt}\n\n${CHILD_REPORTING_GUIDANCE}`, { encoding: "utf8", mode: 0o600 });
    return { dir, path };
  } catch (error) {
    if (dir) {
      try {
        rmSync(dir, { recursive: true, force: true });
      } catch {
        // Preserve the prompt creation error without exposing prompt data.
      }
    }
    throw error;
  }
};

export class PiRpcSessionRunner implements SessionRunner {
  private readonly spawnProcess: (command: string, args: readonly string[], options: SessionSpawnOptions) => SpawnedSessionProcess;
  private readonly platform: NodeJS.Platform;
  private readonly signalProcessTree: (target: number, signal: NodeJS.Signals) => boolean | Promise<boolean>;
  private readonly setTimer: (callback: () => void, delay: number) => unknown;
  private readonly clearTimer: (timer: unknown) => void;
  private readonly fileExists: (path: string) => boolean;
  private readonly now: () => number;
  private readonly pendingCommands = new WeakMap<SpawnedSessionProcess, Map<string, PendingCommand>>();
  private readonly subscribers = new WeakMap<SpawnedSessionProcess, Set<(event: SessionEvent) => void>>();
  private readonly outputCaptures = new WeakMap<SpawnedSessionProcess, AssistantOutputCapture>();
  private readonly telemetryCaptures = new WeakMap<SpawnedSessionProcess, AssistantTelemetryCapture>();
  private readonly generationCaptures = new WeakMap<SpawnedSessionProcess, GenerationCapture>();
  private readonly activeReasoning = new WeakMap<SpawnedSessionProcess, boolean>();
  private readonly reportedToolCallIds = new WeakMap<SpawnedSessionProcess, BoundedReportIds>();

  constructor(dependencies: PiRpcSessionRunnerDependencies = {}) {
    this.spawnProcess = dependencies.spawnProcess ?? defaultSpawnProcess;
    this.platform = dependencies.platform ?? process.platform;
    this.signalProcessTree = dependencies.signalProcessTree
      ?? ((target, signal) => defaultSignalProcessTree(this.platform, target, signal));
    this.setTimer = dependencies.setTimer ?? ((callback, delay) => setTimeout(callback, delay));
    this.clearTimer = dependencies.clearTimer ?? ((timer) => clearTimeout(timer as NodeJS.Timeout));
    this.fileExists = dependencies.fileExists ?? existsSync;
    this.now = dependencies.now ?? Date.now;
  }

  open(options: SessionOpenOptions): Promise<RunningSubagentSession> {
    const { modelArgument, thinkingArgument } = options.launchOptions;
    const args = ["--mode", "rpc", "--no-session", "--no-extensions", "--extension", fileURLToPath(new URL("./child-extension.ts", import.meta.url))];
    if (modelArgument) args.push("--model", modelArgument);
    if (thinkingArgument) args.push("--thinking", thinkingArgument);

    const tools = getChildLaunchToolAllowlist(options.profile, options.accessMode);
    args.push("--tools", tools.join(","));

    return new Promise<RunningSubagentSession>((resolve, reject) => {
      let child: SpawnedSessionProcess | undefined;
      let prompt: { dir: string; path: string } | undefined;
      let timer: unknown;
      let settled = false;
      let handleStdinFailure = () => {};
      const onStdinError = () => handleStdinFailure();
      const stdoutParser: StdoutParserState = { decoder: new StringDecoder("utf8"), buffer: "", wireBytes: 0, discardUntilLineFeed: false };
      let stderrDecoder = new StringDecoder("utf8");
      let stderrCapture: StderrCapture = { text: "", originalBytes: 0 };
      let stderrFinished = false;
      const appendStderr = (text: string, sourceBytes: number) => {
        stderrCapture.originalBytes += sourceBytes;
        const safeChunk = truncateUtf8(text, CAPTURED_TEXT_MAX_BYTES).text;
        stderrCapture.text = truncateUtf8(stderrCapture.text + safeChunk, CAPTURED_TEXT_MAX_BYTES).text;
        if (stderrCapture.originalBytes > CAPTURED_TEXT_MAX_BYTES) {
          stderrCapture.truncation = {
            originalBytes: stderrCapture.originalBytes,
            keptBytes: Buffer.byteLength(stderrCapture.text, "utf8"),
          };
        }
      };
      const finishStderr = () => {
        if (!stderrFinished) {
          stderrFinished = true;
          appendStderr(stderrDecoder.end(), 0);
        }
        return stderrCapture.text;
      };
      const getStderrCapture = (): StderrCapture => ({ ...stderrCapture });
      const resetStderrCapture = () => {
        stderrCapture = { text: "", originalBytes: 0 };
      };
      const discardPersistentData = () => {
        stdoutParser.decoder = new StringDecoder("utf8");
        stdoutParser.buffer = "";
        stdoutParser.wireBytes = 0;
        stdoutParser.discardUntilLineFeed = false;
        stderrDecoder = new StringDecoder("utf8");
        stderrCapture = { text: "", originalBytes: 0 };
        stderrFinished = true;
      };
      const cleanupPrompt = () => {
        if (!prompt) return;
        try {
          rmSync(prompt.dir, { recursive: true, force: true });
        } catch {
          // Cleanup must not prevent readiness from settling.
        }
      };
      const settle = (error?: Error, childExitConfirmed = false) => {
        if (settled) return;
        settled = true;
        this.clearTimer(timer);
        let failure = error;
        if (error) {
          handleStdinFailure = () => {};
          const processExit = child && !childExitConfirmed
            ? this.terminateTrackedProcess(child, removeStdinListener)
            : undefined;
          removeReadinessListeners();
          removeDataListeners();
          if (!processExit) removeStdinListener();
          discardPersistentData();
          if (processExit) failure = new TrackedSessionOpenError(error.message, processExit);
        } else {
          removeReadinessListeners();
        }
        cleanupPrompt();
        if (failure) reject(failure);
        else if (child) resolve(this.session(
          child,
          finishStderr,
          getStderrCapture,
          discardPersistentData,
          removePersistentListeners,
          (handler) => { handleStdinFailure = handler; },
        ));
        else reject(new Error("Child Pi RPC process was not started"));
      };
      const removeDataListeners = () => {
        if (!child) return;
        child.stdout.removeListener("data", onStdout);
        child.stderr.removeListener("data", onStderr);
      };
      const removeStdinListener = () => {
        child?.stdin.removeListener("error", onStdinError);
      };
      const removePersistentListeners = () => {
        removeDataListeners();
        removeStdinListener();
      };
      const removeReadinessListeners = () => {
        if (!child) return;
        child.removeListener("close", onReadinessClose);
        child.removeListener("error", onReadinessError);
      };
      const onStdout = (data: Buffer) => {
        for (const record of this.parseStdout(stdoutParser, data)) {
          if (record.malformed) {
            if (child) this.incrementMalformedEventCount(child);
            continue;
          }
          const response = record.value;
          if (!settled && isReadinessResponse(response)) {
            if (isSuccessfulStateResponse(response)) settle();
            else settle(new Error("Child Pi RPC readiness protocol failed"));
            continue;
          }
          if (child) {
            this.handleCommandResponse(child, response, resetStderrCapture);
            this.reduceEvent(child, response, getStderrCapture);
          }
        }
      };
      const onStderr = (data: Buffer) => appendStderr(stderrDecoder.write(data), data.byteLength);
      const onReadinessClose = (code: number | null, signal: NodeJS.Signals | null) => {
        const exitReason = signal ? `signal ${signal}` : `code ${code ?? 1}`;
        settle(new Error(`Child Pi RPC process exited unexpectedly (${exitReason})`), true);
      };
      const onReadinessError = (error: Error) => {
        settle(new Error(`Child Pi RPC process failed: ${truncateUtf8(error.message, CAPTURED_TEXT_MAX_BYTES).text}`), true);
      };
      handleStdinFailure = () => settle(new Error(SESSION_STDIN_FAILURE_MESSAGE));

      try {
        prompt = writePrompt(options.profile);
        if (prompt) args.push("--append-system-prompt", prompt.path);

        const invocation = getPiInvocation(args, this.fileExists);
        child = this.spawnProcess(invocation.command, invocation.args, {
          cwd: options.cwd,
          detached: this.platform !== "win32",
          shell: false,
          stdio: ["pipe", "pipe", "pipe"],
        });
        child.stdin.on("error", onStdinError);
        child.stdout.on("data", onStdout);
        child.stderr.on("data", onStderr);
        child.on("close", onReadinessClose);
        child.on("error", onReadinessError);
        timer = this.setTimer(() => settle(new Error("Timed out waiting for child Pi RPC readiness")), SESSION_READINESS_TIMEOUT_MS);
        this.writeSessionInput(child, `${JSON.stringify({ id: "readiness", type: "get_state" })}\n`, handleStdinFailure);
      } catch (error) {
        settle(error instanceof Error ? error : new Error(String(error)));
      }
    });
  }

  private signalSessionProcess(child: SpawnedSessionProcess, signal: NodeJS.Signals): boolean | Promise<boolean> {
    const signalChild = (): boolean => {
      try {
        return child.kill(signal);
      } catch {
        return false;
      }
    };
    const pid = child.pid;
    if (pid === undefined || !Number.isSafeInteger(pid) || pid <= 0) return signalChild();

    const target = this.platform === "win32" ? pid : -pid;
    try {
      const treeDelivery = this.signalProcessTree(target, signal);
      if (typeof treeDelivery === "boolean") return treeDelivery || signalChild();
      return treeDelivery.then(
        (delivered) => delivered || signalChild(),
        () => signalChild(),
      );
    } catch {
      return signalChild();
    }
  }

  private writeSessionInput(child: SpawnedSessionProcess, data: string, onFailure: () => void): void {
    let failed = false;
    const fail = () => {
      if (failed) return;
      failed = true;
      onFailure();
    };
    try {
      child.stdin.write(data, (error) => {
        if (error) fail();
      });
    } catch {
      fail();
    }
  }

  private terminateTrackedProcess(child: SpawnedSessionProcess, onExitCleanup: () => void = () => {}): Promise<void> {
    return new Promise<void>((resolve) => {
      let exited = false;
      let killTimer: unknown;
      const removeListeners = () => {
        child.removeListener("close", onClose);
        child.removeListener("error", onError);
      };
      const onClose = () => {
        if (exited) return;
        exited = true;
        if (killTimer !== undefined) this.clearTimer(killTimer);
        removeListeners();
        onExitCleanup();
        resolve();
      };
      const onError = () => {
        // An error event does not prove that the already-spawned process exited.
      };
      const signal = (value: NodeJS.Signals) => {
        try {
          const delivered = this.signalSessionProcess(child, value);
          if (typeof delivered !== "boolean") {
            void delivered.catch(() => {
              // Liveness remains tracked until the process emits close.
            });
          }
        } catch {
          // Liveness remains tracked until the process emits close.
        }
      };

      child.on("close", onClose);
      child.on("error", onError);
      signal("SIGTERM");
      if (exited) return;
      killTimer = this.setTimer(() => {
        if (!exited) signal("SIGKILL");
      }, SESSION_CLOSE_GRACE_MS);
    });
  }

  private session(
    child: SpawnedSessionProcess,
    finishStderr: () => string,
    getStderrCapture: () => StderrCapture,
    discardPersistentData: () => void,
    removePersistentListeners: () => void,
    setStdinFailureHandler: (handler: () => void) => void,
  ): RunningSubagentSession {
    this.pendingCommands.set(child, new Map());
    this.subscribers.set(child, new Set());
    this.outputCaptures.set(child, { text: "", originalBytes: 0, exhausted: false, previewPublished: false });
    this.telemetryCaptures.set(child, { usage: emptyUsage() });
    this.generationCaptures.set(child, this.emptyGenerationCapture());
    this.reportedToolCallIds.set(child, new BoundedReportIds());
    let exited = false;
    let closeTimer: unknown;
    let closePromise: Promise<void> | undefined;
    let closeAttemptSettled = false;
    let resourcesReleased = false;
    let stdinFailed = false;
    let channelFailure: SessionChannelFailureError | undefined;
    let processError: ReturnType<typeof truncateUtf8> | undefined;
    let resolveClosed: (exit: SessionExit) => void = () => {};
    let resolveCloseAttempt: () => void = () => {};
    let rejectCloseAttempt: (error: Error) => void = () => {};
    const closed = new Promise<SessionExit>((resolve) => {
      resolveClosed = resolve;
    });
    const clearCloseTimer = () => {
      if (closeTimer === undefined) return;
      this.clearTimer(closeTimer);
      closeTimer = undefined;
    };
    const rejectPendingCommands = (error: Error) => {
      const pendingCommands = this.pendingCommands.get(child);
      if (!pendingCommands) return;
      this.pendingCommands.delete(child);
      const pending = [...pendingCommands.values()];
      pendingCommands.clear();
      for (const command of pending) {
        this.clearTimer(command.timer);
        command.reject(error);
      }
    };
    const releaseSessionResources = (error: Error, discardData: boolean) => {
      if (resourcesReleased) return;
      resourcesReleased = true;
      rejectPendingCommands(error);
      removePersistentListeners();
      if (discardData) discardPersistentData();
      this.subscribers.get(child)?.clear();
      this.subscribers.delete(child);
      this.outputCaptures.delete(child);
      this.telemetryCaptures.delete(child);
      this.generationCaptures.delete(child);
      this.activeReasoning.delete(child);
      this.reportedToolCallIds.delete(child);
    };
    const removeExitListeners = () => {
      child.removeListener("close", onClose);
      child.removeListener("error", onError);
    };
    const removeMinimalExitListeners = () => {
      child.stdin.removeListener("error", onLateStdinError);
      child.removeListener("close", onLateClose);
      child.removeListener("error", onLateError);
    };
    const onLateClose = (code: number | null, signal: NodeJS.Signals | null) => {
      if (exited) return;
      exited = true;
      removeMinimalExitListeners();
      resolveClosed({
        exitCode: code ?? 1,
        ...(signal ? { signal } : {}),
        expected: true,
        stderr: "",
      });
    };
    const onLateError = () => {
      // Only close confirms that the process is no longer live.
    };
    const onLateStdinError = () => {
      // The close attempt already failed, but the process can still emit EPIPE before exit.
    };
    const retainMinimalExitObservation = () => {
      removeExitListeners();
      child.stdin.on("error", onLateStdinError);
      child.on("close", onLateClose);
      child.on("error", onLateError);
    };
    const failCloseAttempt = (error: Error) => {
      if (exited || closeAttemptSettled) return;
      closeAttemptSettled = true;
      clearCloseTimer();
      releaseSessionResources(error, true);
      processError = undefined;
      retainMinimalExitObservation();
      rejectCloseAttempt(error);
    };
    const sendSignal = (signal: NodeJS.Signals) => {
      const complete = (delivered: boolean) => {
        if (exited || closeAttemptSettled || delivered) return;
        failCloseAttempt(new Error(`Failed to send ${signal} to child Pi RPC process`));
      };
      try {
        const delivered = this.signalSessionProcess(child, signal);
        if (typeof delivered === "boolean") complete(delivered);
        else void delivered.then(complete, () => complete(false));
      } catch {
        complete(false);
      }
    };
    const schedule = (callback: () => void) => {
      if (exited || closeAttemptSettled) return;
      clearCloseTimer();
      closeTimer = this.setTimer(callback, SESSION_CLOSE_GRACE_MS);
    };
    const onClose = (code: number | null, signal: NodeJS.Signals | null) => {
      if (exited) return;
      exited = true;
      clearCloseTimer();
      child.stdin.removeListener("error", onLateStdinError);
      finishStderr();
      const capturedStderr = getStderrCapture();
      const expected = closePromise !== undefined && !stdinFailed;
      let capturedError: ReturnType<typeof truncateUtf8> | undefined;
      if (!expected) {
        const exitReason = signal ? `signal ${signal}` : `code ${code ?? 1}`;
        capturedError = processError ?? truncateUtf8(
          `Child Pi RPC process exited unexpectedly (${exitReason})`,
          CAPTURED_TEXT_MAX_BYTES,
        );
      }
      const commandError = new Error(capturedError?.text ?? "Child Pi RPC process exited during close");
      releaseSessionResources(commandError, false);
      removeExitListeners();
      resolveClosed({
        exitCode: code ?? 1,
        ...(signal ? { signal } : {}),
        expected,
        ...(capturedError ? { error: capturedError.text } : {}),
        ...(capturedError?.truncation ? { errorTruncation: capturedError.truncation } : {}),
        stderr: capturedStderr.text,
        ...(capturedStderr.truncation ? { stderrTruncation: capturedStderr.truncation } : {}),
      });
      if (closePromise && !closeAttemptSettled) {
        closeAttemptSettled = true;
        resolveCloseAttempt();
      }
    };
    const onError = (error: Error) => {
      // Retain a bounded diagnostic, but only close confirms process exit.
      processError ??= truncateUtf8(error.message, CAPTURED_TEXT_MAX_BYTES);
    };
    const failChannel = (error: SessionChannelFailureError) => {
      if (exited || channelFailure) return;
      channelFailure = error;
      stdinFailed = true;
      processError = truncateUtf8(error.message, CAPTURED_TEXT_MAX_BYTES);
      clearCloseTimer();
      releaseSessionResources(error, true);
      child.stdin.on("error", onLateStdinError);
      void this.terminateTrackedProcess(child);
    };
    const onStdinFailure = () => {
      if (closePromise) return;
      failChannel(new SessionChannelFailureError("stdin_failure"));
    };
    const sendKill = () => {
      if (exited) return;
      sendSignal("SIGKILL");
      schedule(() => failCloseAttempt(new Error("Child Pi RPC process did not exit after SIGKILL")));
    };
    const sendTerm = () => {
      if (exited) return;
      sendSignal("SIGTERM");
      schedule(sendKill);
    };
    const close = (): Promise<void> => {
      if (closePromise) return closePromise;
      if (exited) {
        closePromise = Promise.resolve();
        return closePromise;
      }
      closePromise = new Promise<void>((resolve, reject) => {
        resolveCloseAttempt = resolve;
        rejectCloseAttempt = reject;
      });
      if (channelFailure) return closePromise;
      this.sendInternalShutdown(child, onStdinFailure);
      schedule(sendTerm);
      return closePromise;
    };

    setStdinFailureHandler(onStdinFailure);
    child.on("close", onClose);
    child.on("error", onError);
    const send = (command: ChildCommand, message?: string, startsNewGeneration?: boolean): Promise<void> =>
      channelFailure
        ? Promise.reject(channelFailure)
        : this.sendCommand(child, command, onStdinFailure, failChannel, message, startsNewGeneration);
    const rejectReservedMessage = (): Promise<void> => Promise.reject(new Error(RESERVED_CHILD_SHUTDOWN_MESSAGE));
    return {
      prompt: (message, startsNewGeneration = true) =>
        isReservedChildShutdownMessage(message) ? rejectReservedMessage() : send("prompt", message, startsNewGeneration),
      steer: (message) =>
        isReservedChildShutdownMessage(message) ? rejectReservedMessage() : send("steer", message),
      abort: () => send("abort"),
      close,
      subscribe: (listener) => {
        const subscribers = this.subscribers.get(child);
        if (!subscribers) return () => {};
        subscribers.add(listener);
        return () => subscribers.delete(listener);
      },
      closed,
    };
  }

  private emptyGenerationCapture(): GenerationCapture {
    return { active: false, output: "", malformedEventCount: 0 };
  }

  private beginPrompt(
    child: SpawnedSessionProcess,
    resetStderrCapture: () => void,
    startsNewGeneration: boolean,
  ): void {
    const generation = this.generationCaptures.get(child);
    if (!generation || generation.active) return;
    this.generationCaptures.set(child, { ...this.emptyGenerationCapture(), active: true });
    this.outputCaptures.set(child, { text: "", originalBytes: 0, exhausted: false, previewPublished: false });
    this.telemetryCaptures.set(child, { usage: emptyUsage() });
    this.activeReasoning.delete(child);
    if (startsNewGeneration) this.reportedToolCallIds.get(child)?.clear();
    resetStderrCapture();
  }

  private incrementMalformedEventCount(child: SpawnedSessionProcess): void {
    const generation = this.generationCaptures.get(child);
    if (generation?.active) generation.malformedEventCount += 1;
  }

  private settleGeneration(child: SpawnedSessionProcess, getStderrCapture: () => StderrCapture): void {
    const generation = this.generationCaptures.get(child);
    const telemetry = this.telemetryCaptures.get(child);
    if (!generation?.active || !telemetry) return;

    const stderr = getStderrCapture();
    const result: SessionResult = {
      output: generation.output,
      stderr: stderr.text,
      usage: { ...telemetry.usage },
      model: telemetry.model,
      stopReason: generation.stopReason,
      errorMessage: generation.errorMessage,
      errorTruncation: generation.errorTruncation,
      malformedEventCount: generation.malformedEventCount,
      outputTruncation: generation.outputTruncation,
      stderrTruncation: stderr.truncation,
    };
    generation.active = false;
    this.emit(child, { type: "settled", result });
  }

  private reduceEvent(child: SpawnedSessionProcess, value: unknown, getStderrCapture: () => StderrCapture): void {
    if (value === null || typeof value !== "object" || Array.isArray(value)) return;
    const event = value as Record<string, unknown>;
    const capture = this.outputCaptures.get(child);
    if (!capture) return;

    if (event.type === "agent_settled") {
      this.settleGeneration(child, getStderrCapture);
      return;
    }

    if (event.type === "message_update") {
      const assistantEvent = event.assistantMessageEvent;
      if (assistantEvent === null || typeof assistantEvent !== "object" || Array.isArray(assistantEvent)) return;
      const deltaEvent = assistantEvent as Record<string, unknown>;
      if (deltaEvent.type === "thinking_start") {
        this.activeReasoning.set(child, true);
        this.emitProgress(child, "Model reasoning");
        return;
      }
      if (deltaEvent.type === "thinking_delta") {
        if (!this.activeReasoning.get(child)) {
          this.activeReasoning.set(child, true);
          this.emitProgress(child, "Model reasoning");
        }
        return;
      }
      if (deltaEvent.type === "thinking_end") {
        this.activeReasoning.delete(child);
        return;
      }
      if (deltaEvent.type !== "text_delta" || typeof deltaEvent.delta !== "string") return;

      capture.originalBytes += Buffer.byteLength(deltaEvent.delta, "utf8");
      if (!capture.exhausted) {
        const captured = truncateUtf8(capture.text + deltaEvent.delta, CAPTURED_TEXT_MAX_BYTES);
        capture.text = captured.text;
        capture.exhausted = captured.truncation !== undefined;
      }
      capture.previewPublished = true;
      this.emitOutput(child, capture.text, capture.originalBytes);
      return;
    }

    if (event.type === "tool_execution_start" || event.type === "tool_execution_update" || event.type === "tool_execution_end") {
      if (event.type === "tool_execution_start" && event.toolName === "subagent_report" && typeof event.toolCallId === "string" && event.toolCallId.length > 0) {
        const report = normalizeReport(event.args);
        const reportedToolCallIds = this.reportedToolCallIds.get(child);
        if (report && reportedToolCallIds) {
          const reportId = normalizeReportId(event.toolCallId);
          if (reportedToolCallIds.add(reportId)) {
            this.emit(child, { type: "report", reportId, ...report, timestamp: this.now() });
          }
        }
      }
      const toolName = typeof event.toolName === "string" && SAFE_TOOL_NAMES.has(event.toolName) ? event.toolName : "tool";
      let action: "Started" | "Updated" | "Completed";
      if (event.type === "tool_execution_start") {
        action = "Started";
      } else if (event.type === "tool_execution_update") {
        action = "Updated";
      } else {
        action = "Completed";
      }
      this.emitProgress(child, `${action} ${toolName}`);
      return;
    }

    if (event.type !== "message_end") return;
    const message = event.message;
    if (message === null || typeof message !== "object" || Array.isArray(message)) return;
    const assistant = message as Record<string, unknown>;
    if (assistant.role !== "assistant") return;
    this.activeReasoning.delete(child);
    this.emitTelemetry(child, assistant);
    const generation = this.generationCaptures.get(child);
    if (generation?.active) {
      generation.stopReason = typeof assistant.stopReason === "string" ? assistant.stopReason : generation.stopReason;
      generation.errorMessage = undefined;
      generation.errorTruncation = undefined;
      if (typeof assistant.errorMessage === "string") {
        const error = truncateUtf8(assistant.errorMessage, CAPTURED_TEXT_MAX_BYTES);
        generation.errorMessage = error.text;
        generation.errorTruncation = error.truncation;
      }
    }
    if (!Array.isArray(assistant.content)) return;

    const textParts: string[] = [];
    for (const part of assistant.content) {
      if (part === null || typeof part !== "object" || Array.isArray(part)) continue;
      const content = part as Record<string, unknown>;
      if (content.type === "text" && typeof content.text === "string") textParts.push(content.text);
    }
    if (textParts.length > 0 || capture.previewPublished) {
      const text = textParts.join("");
      const captured = truncateUtf8(text, CAPTURED_TEXT_MAX_BYTES);
      if (generation?.active && textParts.length > 0) {
        generation.output = captured.text;
        generation.outputTruncation = captured.truncation;
      }
      this.emitOutput(child, captured.text, Buffer.byteLength(text, "utf8"));
    }
    capture.text = "";
    capture.originalBytes = 0;
    capture.exhausted = false;
    capture.previewPublished = false;
  }

  private emitTelemetry(child: SpawnedSessionProcess, assistant: Record<string, unknown>): void {
    const capture = this.telemetryCaptures.get(child);
    if (!capture) return;

    capture.usage.turns += 1;
    const usage = asRecord(assistant.usage);
    if (usage) {
      capture.usage.input += asNumber(usage.input);
      capture.usage.output += asNumber(usage.output);
      capture.usage.cacheRead += asNumber(usage.cacheRead);
      capture.usage.cacheWrite += asNumber(usage.cacheWrite);
      const cost = asRecord(usage.cost);
      capture.usage.cost += cost ? asNumber(cost.total) : asNumber(usage.cost);
    }
    const modelIdentity = assistantModelIdentity(assistant);
    if (modelIdentity !== undefined) capture.model = modelIdentity;

    this.emit(child, {
      type: "telemetry",
      usage: { ...capture.usage },
      ...(capture.model === undefined ? {} : { model: capture.model }),
    });
  }

  private emitOutput(child: SpawnedSessionProcess, text: string, originalBytes: number): void {
    const keptBytes = Buffer.byteLength(text, "utf8");
    this.emit(child, {
      type: "output",
      text,
      ...(originalBytes > keptBytes ? { truncation: { originalBytes, keptBytes } } : {}),
    });
  }

  private emitProgress(child: SpawnedSessionProcess, text: string): void {
    this.emit(child, { type: "progress", text, timestamp: this.now() });
  }

  private emit(child: SpawnedSessionProcess, event: SessionEvent): void {
    for (const listener of this.subscribers.get(child) ?? []) {
      try {
        listener(event);
      } catch {
        // Subscriber failures must not disrupt RPC parsing or other subscribers.
      }
    }
  }

  /** Bypasses public prompt validation because this command is owned by the session lifecycle. */
  private sendInternalShutdown(child: SpawnedSessionProcess, onStdinFailure: () => void): void {
    this.writeSessionInput(child, `${JSON.stringify({
      id: `prompt-${++nextCommandId}`,
      type: "prompt",
      message: `/${CHILD_SHUTDOWN_COMMAND}`,
    })}\n`, onStdinFailure);
  }

  private sendCommand(
    child: SpawnedSessionProcess,
    command: ChildCommand,
    onStdinFailure: () => void,
    onChannelFailure: (error: SessionChannelFailureError) => void,
    message?: string,
    startsNewGeneration?: boolean,
  ): Promise<void> {
    const id = `${command}-${++nextCommandId}`;
    const pendingCommands = this.pendingCommands.get(child);
    if (!pendingCommands) return Promise.reject(new Error("Child Pi RPC process has exited"));
    const commands = pendingCommands;
    return new Promise<void>((resolve, reject) => {
      const timer = this.setTimer(() => {
        if (!commands.has(id)) return;
        onChannelFailure(new SessionChannelFailureError("command_timeout"));
      }, SESSION_READINESS_TIMEOUT_MS);
      commands.set(id, { command, startsNewGeneration, timer, resolve, reject });
      this.writeSessionInput(
        child,
        `${JSON.stringify({ id, type: command, ...(message === undefined ? {} : { message }) })}\n`,
        onStdinFailure,
      );
    });
  }

  private parseStdout(parser: StdoutParserState, data: Buffer): ParsedStdoutRecord[] {
    const records: ParsedStdoutRecord[] = [];
    let offset = 0;
    for (;;) {
      if (offset >= data.length) return records;
      if (parser.discardUntilLineFeed) {
        const lineFeed = data.indexOf(0x0a, offset);
        if (lineFeed < 0) return records;
        parser.discardUntilLineFeed = false;
        parser.decoder = new StringDecoder("utf8");
        offset = lineFeed + 1;
        continue;
      }

      const lineFeed = data.indexOf(0x0a, offset);
      const end = lineFeed < 0 ? data.length : lineFeed;
      const chunk = data.subarray(offset, end);
      if (parser.wireBytes + chunk.byteLength > RPC_WIRE_RECORD_MAX_BYTES) {
        // Never decode or parse oversized raw records, including complete records in one chunk.
        parser.buffer = "";
        parser.wireBytes = 0;
        parser.decoder = new StringDecoder("utf8");
        records.push({ malformed: true });
        if (lineFeed < 0) {
          parser.discardUntilLineFeed = true;
          return records;
        }
        offset = lineFeed + 1;
        continue;
      }

      parser.buffer += parser.decoder.write(chunk);
      parser.wireBytes += chunk.byteLength;
      if (lineFeed < 0) return records;

      const rawLine = parser.buffer + parser.decoder.end();
      parser.buffer = "";
      parser.wireBytes = 0;
      parser.decoder = new StringDecoder("utf8");
      const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine;
      if (line.trim()) {
        try {
          records.push({ value: JSON.parse(line) as unknown });
        } catch {
          // Malformed output is discarded and never retained.
          records.push({ malformed: true });
        }
      }
      offset = lineFeed + 1;
    }
  }

  private handleCommandResponse(child: SpawnedSessionProcess, value: unknown, resetStderrCapture?: () => void): void {
    if (value === null || typeof value !== "object" || Array.isArray(value)) return;
    const response = value as Record<string, unknown>;
    if (response.type !== "response" || typeof response.id !== "string" || typeof response.command !== "string" || typeof response.success !== "boolean") return;

    const pendingCommands = this.pendingCommands.get(child);
    if (!pendingCommands) return;
    const pending = pendingCommands.get(response.id);
    if (!pending || pending.command !== response.command) return;
    pendingCommands.delete(response.id);
    this.clearTimer(pending.timer);
    if (response.success) {
      if (pending.command === "prompt" && resetStderrCapture) {
        this.beginPrompt(child, resetStderrCapture, pending.startsNewGeneration !== false);
      }
      pending.resolve();
    } else pending.reject(new Error(`Child rejected ${pending.command} command`));
  }
}

const isReadinessResponse = (value: unknown): boolean => {
  if (value === null || typeof value !== "object" || Array.isArray(value)) return false;
  return (value as Record<string, unknown>).id === "readiness";
};

const isSuccessfulStateResponse = (value: unknown): boolean => {
  if (value === null || typeof value !== "object" || Array.isArray(value)) return false;
  const response = value as Record<string, unknown>;
  return response.id === "readiness"
    && response.type === "response"
    && response.command === "get_state"
    && response.success === true;
};