/**
 * openSession: the persistent headless session runner - ONE process, many
 * turns (A-001). `send` writes a descriptor-encoded user record to stdin and starts a
 * turn when idle or hands the text to the harness when a turn is live
 * (the harness queues it). `result` lines delimit turns; identity dedupe (D-022)
 * spans the whole session. Lifecycle is bounded end to end: close() ends
 * stdin, escalates SIGTERM->SIGKILL if the child ignores EOF, and pipes
 * held open past exit close out after grace - a session can always be
 * ended. Pending sends that die with the session are surfaced, never
 * silently dropped. Structured lifecycle events (session open/close, turn
 * start/end, send dispositions, drops) are always-on evidence with
 * sessionId + turnId correlation.
 *
 * The supervising policies - stall clock, signal escalation, stderr
 * classification, question detection at close - are the turn supervisor's
 * (RFC-02 change 5), composed once per session and begun per turn. This
 * runner owns only what is particular to sessions: turn delimiting, the
 * send correlation, and the close-versus-abandon distinction.
 */
import { buildSessionArgv, buildTurnEnv } from "../interpretation/argv.js";
import { type CapabilityResult, capabilitiesOf } from "../interpretation/capabilities.js";
import { composeEscalatedPrompt, type QuestionMode } from "../interpretation/question.js";
import {
  decodeSessionRecord,
  encodeIdentityProbe,
  encodeSessionInput,
  hasNativeReceipts,
  resolveSessionInput,
  SessionInputRefusalError,
} from "../interpretation/session-input.js";
import type { SkillsSelection } from "../interpretation/skills-selection.js";
import type { HarnessDescriptor, SessionInputContract } from "../knowledge/descriptor.js";
import { AsyncChannel } from "./channel.js";
import { decodeLine, decodeParsed, freshDecodeState, settleProvisionalError } from "./decode.js";
import type { RunnerDeps, SpawnedProcess } from "./deps.js";
import {
  DROPPABLE_KINDS,
  type EscalationDetection,
  type ExitCause,
  type HarnessEvent,
} from "./events.js";
import type { FailureSummary } from "./failure.js";
import {
  failureFromLimit,
  failureFromLineOverflow,
  failureFromNativeRejection,
  failureFromStderrTail,
  failureFromTerminalError,
  failureFromTransport,
  isLimitFailure,
  reduceFailures,
} from "./failure.js";
import { LineBuffer, RUN_LINE_MAX } from "./lines.js";
import { PIPE_GRACE_MS, redactArgv } from "./stream-turn.js";
import { StderrTail, superviseTurn } from "./supervisor.js";

/** Grace after stdin EOF before concluding the child will not exit on its
 * own and escalating signals. */
export const CLOSE_GRACE_MS = 5_000;

const PRETURN_MAX = 256;

export interface SessionSendResult {
  readonly disposition: "started" | "rejected";
  /** Present when rejected. `write-failed` is a broken stdin pipe, which is
   * a different remedy from a session the caller already closed - the two
   * must stay distinguishable. */
  readonly reason?: "closed" | "native-rejected" | "write-failed";
  /** Resolves when a harness with command receipts accepts or rejects the input. */
  readonly settled?: Promise<SessionSendResult>;
}

/** One turn's event stream, tagged with the id of the send that opened it.
 * `inputId` is present for every turn a consumer send opened, which today is
 * every turn; a turn opened by anything else (none exists yet) omits it. */
export interface SessionTurn extends AsyncIterable<HarnessEvent> {
  readonly inputId?: string;
  /** `${sessionId}:turn-${n}`, matching the runner's turn_start log. */
  readonly turnId?: string;
}

/** A send's payload: the consumer's correlation id travels with the text
 * from the moment it arrives to the turn it opens and, on death, to the
 * loss report. */
export interface SessionInput {
  readonly id: string;
  readonly text: string;
}

export interface SessionHandle {
  /** One inner iterable per turn, each ending in a turn-scoped `done`.
   * Breaking out of THIS iterable closes the session; breaking out of a
   * single turn's iterable only stops reading that turn. */
  readonly turns: AsyncIterable<SessionTurn>;
  send(input: SessionInput): SessionSendResult;
  close(): Promise<void>;
}

export interface OpenSessionOptions {
  readonly sessionId: string;
  readonly model?: string;
  /** Working directory for the spawned harness. */
  readonly cwd?: string;
  /** question mode for session (ask/assume/none), default "ask" */
  readonly questions?: QuestionMode;
  /** Provider selector (pi); refused on a harness without one. */
  readonly provider?: string;
  /** Named-agent selection (map #288); refused on a harness without a
   * spec, the same way a one-shot turn refuses. */
  readonly agent?: string;
  readonly contextWindow?: number;
  /** Effort level for the session spawn, validated per harness/model the
   * way a one-shot launch validates it. Undefined means the caller made
   * no effort decision: the session runs at the harness's own default
   * effort (sessions carry no profile entry). */
  readonly effort?: string;
  /** True when resuming an existing conversation; controls which flag
   * (resumeFlag vs idFlag) buildSessionArgv renders. */
  readonly isResume?: boolean;
  /** Persistent-memory dimension (ratified 2026-08-26): the session spawn
   * honors it the way a one-shot launch does - false renders the
   * descriptor's disable (claude: CLAUDE_CODE_DISABLE_AUTO_MEMORY env var;
   * pi: nothing, no built-in memory). Undefined means the caller made no
   * memory decision; openSession adds no env overlay then. Resumed
   * sessions get it too - it is a spawn property, not a turn option, so
   * the launch-only rule never silently re-enables memory on a session. */
  readonly memory?: boolean;
  /** The host's own version, reported to harnesses that ask at session
   * open (codex app-server's initialize handshake carries clientInfo). */
  readonly clientVersion?: string;
  /** Caller-directed skills allowlist (issue #332), resolved by the CLI
   * against the registry root. Passed to buildSessionArgv, which renders
   * it through the same owner a one-shot launch uses; a harness that
   * cannot enforce an allowlist refuses here, before any spawn. */
  readonly skills?: SkillsSelection;
}

export class SessionClosedError extends Error {
  constructor() {
    super("session is closed; sends have nowhere to go");
    this.name = "SessionClosedError";
  }
}

export class SessionSpawnError extends Error {
  constructor(harness: string, cause: unknown) {
    super(
      `could not spawn a ${harness} session: ${cause instanceof Error ? cause.message : String(cause)}`,
    );
    this.name = "SessionSpawnError";
  }
}

/** True for harnesses whose session input waits for the spawn-probe
 * identity before any send can be written (popeye mint the session via
 * the create probe; codex needs the thread id on every request). pi's
 * prompt-wire is identity-silent and writes at send time. The single
 * predicate `send` uses to buffer before identity and the #345 arm uses
 * to settle buffered sends when the probe fails. */
const buffersSendsBeforeIdentity = (input: SessionInputContract): boolean =>
  input.kind === "popeye-rpc-prompt" || input.kind === "codex-jsonrpc";

export const openSession = (
  h: HarnessDescriptor,
  opts: OpenSessionOptions,
  deps: RunnerDeps,
): SessionHandle => {
  const log = deps.log ?? (() => {});
  const argv = buildSessionArgv(h, {
    sessionId: opts.sessionId,
    ...(opts.model !== undefined ? { model: opts.model } : {}),
    ...(opts.effort !== undefined ? { effort: opts.effort } : {}),
    ...(opts.provider !== undefined ? { provider: opts.provider } : {}),
    ...(opts.agent !== undefined ? { agent: opts.agent } : {}),
    ...(opts.contextWindow !== undefined ? { contextWindow: opts.contextWindow } : {}),
    ...(opts.isResume !== undefined ? { isResume: opts.isResume } : {}),
    ...(opts.skills !== undefined ? { skills: opts.skills } : {}),
  });
  let sessionInput: SessionInputContract;
  try {
    sessionInput = resolveSessionInput(h);
  } catch (cause) {
    if (!(cause instanceof SessionInputRefusalError)) throw cause;
    log({
      event: "session_input_refused",
      harness: h.name,
      issue: cause.issue,
      sessionId: opts.sessionId,
    });
    throw cause;
  }

  // Descriptor-derived spawn env (claude's memory disable) - same merge
  // rule as streamTurn: rendered at spawn, reported in the open log line.
  const turnEnv =
    opts.memory !== undefined ? buildTurnEnv(h, { memory: opts.memory }, "launch") : {};
  const envKeys = Object.keys(turnEnv).length > 0 ? Object.keys(turnEnv) : undefined;

  let proc: SpawnedProcess;
  try {
    // A session needs a writable stdin regardless of the descriptor's
    // one-shot stdin policy - that policy governs turns, not sessions.
    proc = deps.spawn(argv, {
      stdin: "pipe",
      ...(opts.cwd !== undefined ? { cwd: opts.cwd } : {}),
      ...(envKeys !== undefined ? { env: turnEnv } : {}),
    });
  } catch (cause) {
    log({ event: "session_spawn_failed", sessionId: opts.sessionId, harness: h.name });
    throw new SessionSpawnError(h.name, cause);
  }
  const stdin = proc.stdin;
  if (stdin === undefined) {
    // The child exists but there is no way to ever send to it - end it
    // before throwing, or it runs orphaned forever.
    deps.signal(proc, "SIGTERM");
    log({ event: "session_spawn_failed", sessionId: opts.sessionId, harness: h.name });
    throw new SessionSpawnError(h.name, new Error("spawner opened no stdin pipe"));
  }

  log({
    event: "session_open",
    sessionId: opts.sessionId,
    harness: h.name,
    argv: redactArgv(argv),
    ...(envKeys !== undefined ? { envKeys } : {}),
  });

  const turnsChannel = new AsyncChannel<SessionTurn>();
  const state = freshDecodeState(opts.sessionId, h.name);
  const questionMode: QuestionMode = opts.questions ?? "ask";
  const sessionInputMode = h.sessionMode;
  const stderrTail = new StderrTail();
  let turnCounter = 0;
  let activeTurn: AsyncChannel<HarnessEvent> | null = null;
  let activeTurnId = "";
  // The id of the send that opened the active turn (the inputId the turn
  // was tagged with). Used by the #344 command-failed arm to decide
  // whether the harness refused the command that opened the turn, vs a
  // later input (steer, codex request with no turn). Cleared at each turn
  // end.
  let activeTurnInputId: string | undefined;
  let activeTurnOpenerAccepted = false;
  let activeNativeTurnId: string | null = null;
  const pendingIds: string[] = [];
  const pendingLengths: number[] = [];
  const pendingNativeReceipts = new Map<string, (result: SessionSendResult) => void>();
  // The id of the codex turn currently running NATIVELY (result.turn.id /
  // turn/steer's {turnId}), the steer precondition for the next mid-turn
  // send. Null while no native turn is open; cleared at each turn end.
  // close() waits here while a turn is open. Ending the child's stdin
  // mid-turn is fatal on pi: rpc treats EOF as "finish up and exit", so the
  // prompt it has buffered never runs and the turn ends clean with no
  // output (issue #99). claude happens to drain a queued turn after EOF,
  // which is why this was invisible there. The README promises that a close
  // after a send lets the turn finish; this is what keeps that promise.
  let turnSettled: (() => void) | null = null;
  const preTurnEvents: HarnessEvent[] = [];
  let dead = false;
  let closing = false;
  // Popeye mints the session via the create probe: sends before the
  // response carry no session id and refuse. Buffer them; identity flushes.
  const popeyePending: Array<{ input: SessionInput }> = [];
  let finalized = false;
  let exitCode: number | null = null;
  let resultError = false;
  let turnLimitSeen = false;
  let turnFailures: FailureSummary[] = [];
  let turnDenialCount = 0;
  let pumpError: unknown = null;
  // issue #44: whether the active turn ended by asking, and what its close
  // detected.
  let turnAsked = false;
  let turnEscalationDetection: EscalationDetection = "none";
  let identityAnnounced = false;
  // A stall killed the process on purpose, so the signal death it caused
  // reports as "stall", not "killed".
  let stalled = false;
  // A refused session open (#345) or a crash with no open turn (#347)
  // has no turn to carry its failure. First wins; session_close carries it.
  let sessionFailure: FailureSummary | undefined;

  // The supervisor spans the session; each turn begins and ends on it. A
  // session turn can hang with the process alive and the pipes open, which
  // no exit code reports; the stall clock is what bounds that wait.
  const sup = superviseTurn(h, questionMode, {
    clock: deps.clock,
    stallMs: deps.stallMs,
    signal: (sig) => deps.signal(proc, sig),
    emit: (event) => routeEvent(event),
    fail: (summary) => pushFailure(summary),
    tail: stderrTail,
    onStall: () => {
      stalled = true;
      log({
        event: "stall",
        sessionId: opts.sessionId,
        turnId: activeTurnId,
        harness: h.name,
        reason: "inactivity",
        budgetMs: deps.stallMs,
      });
      // The turn is owed its own terminal event before the process dies;
      // the exit path then closes the session with the same cause.
      void pushFailure(failureFromTransport("stalled: inactivity"));
      endTurn({ kind: "done", exitCode: null, cause: "stall" });
    },
    onQuestion: (options) => {
      log({
        event: "question",
        sessionId: opts.sessionId,
        turnId: activeTurnId,
        harness: h.name,
        options,
      });
    },
  });

  const writeUser = (
    input: SessionInput,
    busy: boolean,
  ): { readonly settled?: Promise<SessionSendResult>; readonly written: boolean } => {
    let settleNative: ((result: SessionSendResult) => void) | undefined = pendingNativeReceipts.get(
      input.id,
    );
    // A send buffered before identity pre-registers its receipt; the flush
    // then writes with the registration already in place.
    const settled =
      settleNative === undefined && hasNativeReceipts(sessionInput.kind)
        ? new Promise<SessionSendResult>((resolve) => {
            settleNative = resolve;
            pendingNativeReceipts.set(input.id, resolve);
          })
        : undefined;
    try {
      stdin.write(
        encodeSessionInput(
          sessionInput,
          composeEscalatedPrompt(input.text, questionMode, "session"),
          {
            busy,
            id: input.id,
            // Popeye prompt frames and codex requests carry the minted
            // session/thread id; the create/thread-open response binds it
            // before any send is written.
            ...((sessionInput.kind === "popeye-rpc-prompt" ||
              sessionInput.kind === "codex-jsonrpc") &&
            state.lastSeenId !== null
              ? { sessionId: state.lastSeenId }
              : {}),
            // Codex steer precondition: the tracked id of the running
            // native turn. Unknown (receipt still in flight) renders as
            // the empty string, which the harness itself rejects - the
            // send reports rejected instead of being written blindly.
            ...(sessionInput.kind === "codex-jsonrpc" && activeNativeTurnId !== null
              ? { activeTurnId: activeNativeTurnId }
              : {}),
          },
        ),
      );
      return { ...(settled === undefined ? {} : { settled }), written: true };
    } catch {
      pendingNativeReceipts.delete(input.id);
      settleNative?.({ disposition: "rejected", reason: "write-failed" });
      // A broken stdin pipe ends the session: there is no way to drive the
      // child any more. Surface it as its own event, stop accepting sends,
      // and END the child - marking it dead here instead would suppress the
      // very signal that stops it. The exit path then finalizes as usual.
      void routeEvent({ kind: "error", message: "send failed: session stdin is gone" });
      closing = true;
      sup.escalate();
      return { written: false };
    }
  };

  /** A decoded failure event minus its kind: what done.failure carries. */
  const summaryOf = (event: HarnessEvent & { kind: "failure" }): FailureSummary => {
    const { kind: _kind, ...summary } = event;
    return summary;
  };

  const startTurn = (inputId?: string): void => {
    turnLimitSeen = false;
    turnFailures = [];
    turnDenialCount = 0;
    turnAsked = false;
    turnEscalationDetection = "none";
    activeTurn = new AsyncChannel<HarnessEvent>();
    activeTurnId = `${opts.sessionId}:turn-${++turnCounter}`;
    activeTurnInputId = inputId;
    activeTurnOpenerAccepted = false;
    // Tag the turn with the id of the send that opened it, so the consumer
    // correlates an input to its turn by reading the tag, not by
    // shadowing the runner's delivery order.
    (activeTurn as { inputId?: string; turnId?: string }).inputId = inputId;
    (activeTurn as { inputId?: string; turnId?: string }).turnId = activeTurnId;
    log({ event: "turn_start", sessionId: opts.sessionId, turnId: activeTurnId });
    for (const held of preTurnEvents.splice(0)) {
      if (held.kind === "failure") turnFailures.push(summaryOf(held));
      if (held.kind === "error" && held.denial !== undefined) turnDenialCount++;
      activeTurn.push(held);
    }
    turnsChannel.push(activeTurn as SessionTurn);
    sup.beginTurn();
  };

  const endTurn = (
    done: Omit<HarnessEvent & { kind: "done" }, "escalation" | "denialCount">,
  ): void => {
    if (activeTurn === null) return;
    sup.disarm();
    // issue #44: the supervisor scans the last assistant message for the
    // hcn-question block; the question event lands in the turn stream
    // right before its done. Asking is a successful turn: the session
    // semantic is "blocked on answer, session alive" - the done stays
    // TURN-scoped (exitCode null in sessions) and the caller answers with
    // the next send().
    const close = sup.close();
    turnEscalationDetection = close.detection;
    turnAsked = close.asked;
    if (turnAsked && done.cause === "clean") done = { ...done, cause: "awaiting-input" };
    // RFC-01 every turn end carries the escalation record
    let fullDone: HarnessEvent & { kind: "done" } = {
      ...done,
      ...(turnDenialCount > 0 ? { denialCount: turnDenialCount } : {}),
      escalation: { mode: questionMode, detection: turnEscalationDetection },
    } as HarnessEvent & { kind: "done" };
    // Every failure was already emitted as an event through pushFailure;
    // the turn's done carries the reduced summary, as streamTurn's does.
    const reduced = reduceFailures(turnFailures);
    if (reduced !== undefined) {
      if (fullDone.cause === "clean") fullDone = { ...fullDone, cause: "failed", failure: reduced };
      else fullDone = { ...fullDone, failure: reduced };
    }
    activeTurn.push(fullDone);
    activeTurn.close();
    log({
      event: "turn_end",
      sessionId: opts.sessionId,
      turnId: activeTurnId,
      cause: fullDone.cause,
    });
    activeTurn = null;
    activeTurnInputId = undefined;
    activeTurnOpenerAccepted = false;
    activeNativeTurnId = null;
    resultError = false;
    if (turnSettled !== null) {
      const release = turnSettled;
      turnSettled = null;
      release();
    }
    if (dead || closing) return;
    const nextId = pendingIds.shift();
    if (nextId === undefined) return;
    pendingLengths.shift();
    // The text was already written to the harness when the send arrived;
    // the id was held to correlate the next turn.
    startTurn(nextId);
  };

  const routeEvent = (event: HarnessEvent): Promise<void> => {
    if (event.kind === "limit") {
      state.limitSeen = true;
      turnLimitSeen = true;
    }
    sup.noteEvent(event);
    if (activeTurn !== null) {
      if (event.kind === "error" && event.denial !== undefined) turnDenialCount++;
      // Awaited by the pumps: past the channel's high water mark this
      // blocks the pump, so OS pipe backpressure reaches the child.
      return activeTurn.push(event);
    }
    // Between-turn events wait for the next turn, bounded: past the cap
    // droppable events go first, then oldest.
    preTurnEvents.push(event);
    if (preTurnEvents.length > PRETURN_MAX) {
      const droppableAt = preTurnEvents.findIndex((e) => DROPPABLE_KINDS.has(e.kind));
      preTurnEvents.splice(droppableAt === -1 ? 0 : droppableAt, 1);
    }
    return Promise.resolve();
  };

  const pushFailure = (f: FailureSummary): Promise<void> => {
    turnFailures.push(f);
    return routeEvent({ kind: "failure", ...f });
  };

  /** A decoded event other than a failure: a limit also records the
   * failure it stands for, and so does a terminal error, the way
   * streamTurn does. A content-path identity re-emit carrying a later
   * model attestation also re-announces the session identity (same
   * sessionId), so last-takes consumers read the observed model. */
  const routeDecoded = async (event: HarnessEvent): Promise<void> => {
    if (event.kind === "failure") {
      await pushFailure(summaryOf(event));
      return;
    }
    await routeEvent(event);
    if (event.kind === "limit") await pushFailure(failureFromLimit(event.code));
    if (event.kind === "error" && event.terminal === true) {
      const failure = failureFromTerminalError(h, event.message, deps.clock);
      if (isLimitFailure(failure)) {
        state.limitSeen = true;
        turnLimitSeen = true;
      }
      await pushFailure(failure);
    }
  };

  /** The probe answered: bind the announced id under the descriptor's
   * authority, or surface a rotation. The re-announce below keeps a live
   * session's identity current: when a later model attestation arrives on
   * content records, the turn's identity is re-emitted with the observed
   * model filled in (same sessionId, dedupe-by-id consumers see one) -
   * display only, never validation or refusal input. */
  const reannounceIdentity = (): Promise<void> => {
    if (!identityAnnounced || state.observedModel === null) return Promise.resolve();
    if (state.observedModel === state.emittedModel) return Promise.resolve();
    state.emittedModel = state.observedModel;
    const authority = sessionInputMode?.idFlag === null ? "harness-minted" : "caller-assigned";
    return routeEvent({
      kind: "identity",
      sessionId: state.lastSeenId ?? opts.sessionId,
      authority,
      capabilities: capabilitiesWithObserved(),
    });
  };
  const capabilitiesWithObserved = (): CapabilityResult => {
    const caps = capabilitiesOf(h, opts.model ?? "", "headless-session");
    if (state.observedModel === null) return caps;
    return {
      ...caps,
      escalation: {
        ...caps.escalation,
        observedOn: {
          harness: caps.escalation.observedOn?.harness ?? h.name,
          model: state.observedModel,
          version: caps.escalation.observedOn?.version ?? h.verifiedAgainst,
          date: caps.escalation.observedOn?.date ?? "",
        },
      },
    };
  };
  const announceIdentity = async (announced: string): Promise<void> => {
    if (identityAnnounced) return;
    // Kinds that mint identity via the spawn probe (popeye create, codex
    // thread open) buffer sends that arrived before the response; the
    // flush replays them with the minted id bound.
    const flushPreIdentity = (): void => {
      for (const pending of popeyePending.splice(0)) {
        // Re-check busyness at flush time: every pre-identity send
        // recorded idle, but an earlier flush entry may own the turn.
        const busy = activeTurn !== null;
        const written = writeUser(pending.input, busy);
        if (!written.written) continue;
        if (busy && sessionInput.kind !== "codex-jsonrpc") {
          pendingIds.push(pending.input.id);
          pendingLengths.push(pending.input.text.length);
        } else if (!busy) {
          // Codex defers its turn open to the native receipt; pi/popeye
          // open at write time. Either way the id is now in flight.
          if (sessionInput.kind !== "codex-jsonrpc") startTurn(pending.input.id);
        }
      }
    };
    if (sessionInputMode?.idFlag === null) {
      // Harness-MINTED identity (pi rpc: `--session` refuses unknown ids,
      // so fresh sessions omit the flag; popeye and codex mint through the
      // spawn probe). The minted id IS the identity;
      // opts.sessionId stays the caller-side handle.
      identityAnnounced = true;
      state.lastSeenId = announced;
      flushPreIdentity();
      await routeEvent({
        kind: "identity",
        sessionId: announced,
        authority: "harness-minted",
        capabilities: capabilitiesWithObserved(),
      });
      return;
    }
    if (announced === opts.sessionId) {
      identityAnnounced = true;
      await routeEvent({
        kind: "identity",
        sessionId: announced,
        authority: "caller-assigned",
        capabilities: capabilitiesWithObserved(),
      });
      return;
    }
    await routeEvent({
      kind: "error",
      message: `identity rotated: session announced ${JSON.stringify(announced)} but ${opts.sessionId} was requested`,
    });
  };

  const pumpStdout = async (): Promise<void> => {
    // An over-long line is reported when it ends; the pump turns each report
    // into a failure after the lines of the same chunk.
    const overflows: FailureSummary[] = [];
    const lines = new LineBuffer(RUN_LINE_MAX, (overflow) =>
      overflows.push(failureFromLineOverflow("stdout", overflow)),
    );
    // issue #44: a harness that is identity-silent at startup gets the
    // probe its descriptor declares; interpretation encodes it (ADR 0005).
    // Codex's probe is two JSON-RPC lines: initialize (required first,
    // verified 0.159.2) then the thread open - thread/start fresh,
    // thread/resume with the requested id when resuming.
    const probe = encodeIdentityProbe(h, {
      sessionId: opts.sessionId,
      isResume: opts.isResume,
      model: opts.model,
      clientVersion: opts.clientVersion,
    });
    if (probe !== null) {
      try {
        stdin.write(probe);
      } catch {
        // stdin already gone; the exited handler will surface the death.
      }
    }
    const handleLine = async (line: string): Promise<void> => {
      let parsed: Record<string, unknown> | null = null;
      try {
        parsed = JSON.parse(line) as Record<string, unknown>;
      } catch {
        // Not a record: the shared decoder reads the one signal a plain
        // line can carry (a wall) and nothing else.
        for (const event of decodeLine(h, line, state, opts.model ?? "")) {
          await routeDecoded(event);
        }
        return;
      }
      // Interpretation says what the record means; this runner only routes
      // on the kind (ADR 0005: no harness field names here).
      const record = decodeSessionRecord(h, parsed);
      switch (record.kind) {
        case "command-accepted": {
          if (record.nativeTurnId !== undefined) activeNativeTurnId = record.nativeTurnId;
          // A harness whose turn open is a protocol REQUEST (codex) only
          // starts the hcn turn once the receipt confirms it; a failed
          // open then leaves no stuck empty turn. Prompt-wire harnesses
          // (pi) start the turn at write time, so this is a no-op for
          // them - the turn already exists. Starting BEFORE the receipt
          // settles keeps the turn line ahead of the disposition line.
          if (activeTurn === null) startTurn(record.inputId);
          if (record.inputId === activeTurnInputId) activeTurnOpenerAccepted = true;
          const settle = pendingNativeReceipts.get(record.inputId);
          pendingNativeReceipts.delete(record.inputId);
          settle?.({ disposition: "started" });
          return;
        }
        case "identity":
          await announceIdentity(record.sessionId);
          return;
        case "probe-failed": {
          await routeEvent({ kind: "error", message: record.message });
          if (buffersSendsBeforeIdentity(sessionInput)) {
            const failure = failureFromNativeRejection(h, record.message, deps.clock);
            if (sessionFailure === undefined) sessionFailure = failure;
            await pushFailure(failure);
            // Codex rejects stranded sends without opening a phantom turn;
            // popeye keeps them so the first turn carries the refusal.
            if (sessionInput.kind === "codex-jsonrpc") {
              for (const pending of popeyePending.splice(0)) {
                const settle = pendingNativeReceipts.get(pending.input.id);
                pendingNativeReceipts.delete(pending.input.id);
                settle?.({ disposition: "rejected", reason: "native-rejected" });
              }
            }
            void close(false);
          }
          return;
        }
        case "command-failed": {
          if (record.inputId !== undefined) {
            const settle = pendingNativeReceipts.get(record.inputId);
            pendingNativeReceipts.delete(record.inputId);
            const pendingIndex = pendingIds.indexOf(record.inputId);
            if (pendingIndex >= 0) {
              pendingIds.splice(pendingIndex, 1);
              pendingLengths.splice(pendingIndex, 1);
            }
            settle?.({ disposition: "rejected", reason: "native-rejected" });
            // Issue #344: the harness refused the command that opened the
            // turn before acknowledging it. End it so close() cannot hang;
            // a refused later input, even with a reused id, leaves it running.
            if (
              activeTurn !== null &&
              activeTurnInputId === record.inputId &&
              !activeTurnOpenerAccepted
            ) {
              await routeEvent({ kind: "error", message: record.message });
              const failure = failureFromNativeRejection(h, record.message, deps.clock);
              await pushFailure(failure);
              endTurn({ kind: "done", exitCode: null, cause: "clean" });
              return;
            }
          }
          await routeEvent({ kind: "error", message: record.message });
          return;
        }
        case "ignored":
          return;
        case "turn-end": {
          // The turn-end record still feeds identity dedupe (claude includes
          // session_id on result - a rotation announced there must not be
          // missed) and content decoding, which already surfaces a failed
          // result as an error event; the flag here only classifies the
          // done cause.
          for (const event of decodeParsed(h, parsed, state, opts.model ?? "")) {
            await routeDecoded(event);
          }
          // The turn is over: a provisional terminal error (pi stopReason
          // error) that no successful assistant message in this turn
          // superseded settles into the turn's verdict now. The events are
          // routed before endTurn so the failure lands inside the closing
          // turn, never in the next one.
          for (const event of settleProvisionalError(state)) {
            await routeDecoded(event);
          }
          if (record.isError) resultError = true;
          endTurn({
            kind: "done",
            exitCode: null,
            cause: turnLimitSeen ? "limit" : resultError ? "crash" : "clean",
          });
          return;
        }
        case "content": {
          for (const event of decodeParsed(h, parsed, state, opts.model ?? "")) {
            await routeDecoded(event);
          }
          // A model attestation decoded on this record postdates the probe
          // identity: re-announce so the turn's identity carries it.
          await reannounceIdentity();
          return;
        }
        default: {
          const exhaustive: never = record;
          return exhaustive;
        }
      }
    };
    for await (const chunk of proc.stdout) {
      sup.rearm();
      for (const line of lines.push(chunk)) await handleLine(line);
      for (const failure of overflows.splice(0)) await pushFailure(failure);
    }
    const rest = lines.flush();
    if (rest !== null) await handleLine(rest);
    for (const failure of overflows.splice(0)) await pushFailure(failure);
  };

  const pumpStderr = async (): Promise<void> => {
    const overflows: FailureSummary[] = [];
    const lines = new LineBuffer(RUN_LINE_MAX, (overflow) =>
      overflows.push(failureFromLineOverflow("stderr", overflow)),
    );
    for await (const chunk of proc.stderr) {
      sup.rearm();
      for (const line of lines.push(chunk)) await sup.stderrLine(line);
      for (const failure of overflows.splice(0)) await pushFailure(failure);
    }
    // Flushed only for an over-long final line's report: an unterminated
    // final stderr line never reached the tail scan, and still does not.
    lines.flush();
    for (const failure of overflows.splice(0)) await pushFailure(failure);
  };

  const pumping = Promise.allSettled([pumpStdout(), pumpStderr()]).then((settlements) => {
    pumpError = settlements.find((settlement) => settlement.status === "rejected")?.reason ?? null;
  });

  let shutdownComplete: () => void = () => {};
  const shutdown = new Promise<void>((resolve) => {
    shutdownComplete = resolve;
  });

  let pipesOpenAtExit = false;
  const finalize = (): void => {
    if (finalized) return;
    finalized = true;
    if (pumpError !== null) {
      void routeEvent({ kind: "error", message: `session pump failed: ${String(pumpError)}` });
    }
    // Issue #345: a refused identity probe lands here with popeyePending
    // already drained by the probe-failed arm and sessionFailure set.
    // For a clean exit, that becomes "failed" - the harness did not crash,
    // it refused to open, and the cause must say so. Crash, killed, stall
    // and limit are stronger signals and stay.
    const baseCause: ExitCause = stalled
      ? "stall"
      : state.limitSeen
        ? "limit"
        : exitCode === 0
          ? "clean"
          : exitCode === null
            ? "killed"
            : "crash";
    const cause: ExitCause =
      sessionFailure !== undefined && baseCause === "clean" ? "failed" : baseCause;
    for (const settle of pendingNativeReceipts.values()) {
      settle({ disposition: "rejected", reason: "closed" });
    }
    pendingNativeReceipts.clear();
    if (popeyePending.length > 0) {
      const dropped = popeyePending.splice(0);
      // Popeye already reported started, so its buffered send needs a turn
      // to carry the failure. Codex receipts were rejected by the sweep above.
      if (sessionInput.kind === "popeye-rpc-prompt") {
        if (activeTurn === null) {
          const first = dropped[0];
          if (first !== undefined) startTurn(first.input.id);
        }
        const ids = dropped.map((pending) => pending.input.id);
        void routeEvent({
          kind: "error",
          message: `${ids.length} buffered send(s) died before session identity: ${ids.join(", ")}`,
        });
      }
    }
    if (pendingIds.length > 0) {
      const droppedIds = [...pendingIds];
      const droppedLengths = [...pendingLengths];
      void routeEvent({
        kind: "error",
        message: `${pendingIds.length} pending send(s) died with the session: ${droppedIds.join(", ")}`,
      });
      log({
        event: "sends_dropped",
        sessionId: opts.sessionId,
        count: pendingIds.length,
        ids: droppedIds,
        lengths: droppedLengths,
      });
      pendingIds.length = 0;
      pendingLengths.length = 0;
    }
    // A turn still open when the process died settles its provisional
    // terminal error here (pi stopReason error the turn never superseded).
    // Both the failure record and the events enqueue synchronously -
    // routeDecoded defers its failure half past a microtask boundary, which
    // would land it after endTurn's done - so finalize records the failure
    // directly, before endTurn reduces the turn's verdict.
    if (activeTurn !== null) {
      for (const event of settleProvisionalError(state)) {
        const failure = failureFromTerminalError(h, event.message, deps.clock);
        if (isLimitFailure(failure)) {
          state.limitSeen = true;
          turnLimitSeen = true;
        }
        turnFailures.push(failure);
        void routeEvent(event);
        void routeEvent({ kind: "failure", ...failure });
      }
      // Issue #341: an open turn at a crash exit gets run mode's
      // stderr-tail verdict. Unlike run mode, done.cause stays crash for
      // every class.
      if (cause === "crash" && turnFailures.length === 0) {
        void pushFailure(failureFromStderrTail(h, exitCode, stderrTail.snapshot()));
      }
    } else if (sessionFailure === undefined && (cause === "crash" || cause === "limit")) {
      // Between-turn walls are parked events, not stderr tail. Never reuse
      // turnFailures here: it still holds the last completed turn's verdict.
      sessionFailure = reduceFailures(
        preTurnEvents.filter((event) => event.kind === "failure").map(summaryOf),
      );
      if (sessionFailure === undefined && cause === "crash") {
        sessionFailure = failureFromStderrTail(h, exitCode, stderrTail.snapshot());
      }
    }

    endTurn({ kind: "done", exitCode, cause });
    if (preTurnEvents.some((e) => !DROPPABLE_KINDS.has(e.kind))) {
      log({
        event: "preturn_events_dropped",
        sessionId: opts.sessionId,
        kinds: preTurnEvents.map((e) => e.kind),
      });
    }
    turnsChannel.close();
    log({
      event: "session_close",
      sessionId: opts.sessionId,
      exitCode,
      cause,
      ...(sessionFailure !== undefined ? { failure: sessionFailure } : {}),
      ...(pipesOpenAtExit ? { pipesOpenAtExit } : {}),
      ...(cause === "crash" || cause === "killed" ? { stderrTail: stderrTail.snapshot() } : {}),
    });
    shutdownComplete();
  };

  void proc.exited.then((code) => {
    dead = true;
    exitCode = code;
    // A close() waiting on an open turn must not outlive the child.
    if (turnSettled !== null) {
      const release = turnSettled;
      turnSettled = null;
      release();
    }
    // The process is gone: no signal, no pending escalation, no stall
    // clock - a later fire would flip a finished turn to stall.
    sup.settle();
    // Pipes held open past exit (a grandchild) must not hang the session.
    const pipeGrace = deps.clock.setTimeout(() => {
      pipesOpenAtExit = true;
      activeTurn?.releaseBackpressure();
      proc.disposeOutput();
      void pumping.then(() => finalize());
    }, PIPE_GRACE_MS);
    void pumping.then(() => {
      deps.clock.clearTimeout(pipeGrace);
      finalize();
    });
  });

  // Two intents share this path and must not be conflated. A CLOSE is the
  // consumer asking politely: an open turn gets to finish first, because
  // the consumer is still there to receive it. ABANDONMENT is the consumer
  // walking away from the turns iterable: nobody is left to receive a turn,
  // so stdin ends at once and the child is reaped, not drained.
  const close = async (drain = true): Promise<void> => {
    if (closing) return shutdown;
    closing = true;
    // Let an open turn reach its end record before stdin goes away. The
    // wait is bounded twice over: the stall watchdog ends a silent turn,
    // and the grace below still escalates a child that never exits.
    if (drain && activeTurn !== null && !dead) {
      await new Promise<void>((resolve) => {
        turnSettled = resolve;
      });
    }
    try {
      stdin.end();
    } catch {
      // stdin may already be gone - escalation below still bounds close.
    }
    // A child that does not exit on stdin EOF gets signalled after grace.
    const closeGrace = deps.clock.setTimeout(() => sup.escalate(), CLOSE_GRACE_MS);
    await shutdown;
    deps.clock.clearTimeout(closeGrace);
  };

  return {
    // Breaking out of the turns iterable is abandonment - treat it as
    // close() so the child is never left running undrained (the streamTurn
    // C1 scar, adapted to a handle-shaped API).
    turns: (async function* () {
      try {
        for await (const turn of turnsChannel) yield turn;
      } finally {
        if (!closing && !dead) void close(false);
      }
    })(),
    send(input: SessionInput): SessionSendResult {
      if (dead || closing) throw new SessionClosedError();
      const wasBusy = activeTurn !== null;
      // Popeye sends before the create response - and codex sends before
      // the thread-open response - wait for identity; the flush replays
      // them with the minted session/thread id bound.
      if (buffersSendsBeforeIdentity(sessionInput) && state.lastSeenId === null) {
        popeyePending.push({ input });
        if (sessionInput.kind === "codex-jsonrpc") {
          // The receipt registration exists from now, so the flushed
          // write settles THIS promise.
          const settled = new Promise<SessionSendResult>((resolve) => {
            pendingNativeReceipts.set(input.id, resolve);
          });
          return { disposition: "started", settled };
        }
        return { disposition: "started" };
      }
      const written = writeUser(input, wasBusy);
      if (!written.written) {
        log({
          event: "send",
          sessionId: opts.sessionId,
          turnId: activeTurnId,
          inputId: input.id,
          disposition: "rejected",
          reason: "write-failed",
        });
        return { disposition: "rejected", reason: "write-failed" };
      }
      if (wasBusy) {
        // Codex and Pi steer into the running turn; queued inputs on the
        // other protocols identify a later native turn.
        if (sessionInput.kind !== "codex-jsonrpc" && sessionInput.kind !== "pi-rpc-prompt") {
          pendingIds.push(input.id);
          pendingLengths.push(input.text.length);
        }
      } else if (sessionInput.kind !== "codex-jsonrpc") {
        // Codex opens its hcn turn when the native receipt confirms the
        // turn/start request (command-accepted below); writing the turn
        // open at send time would leave a stuck empty turn on a refusal.
        startTurn(input.id);
      }
      log({
        event: "send",
        sessionId: opts.sessionId,
        turnId: activeTurnId,
        inputId: input.id,
        disposition: "started",
      });
      return {
        disposition: "started",
        ...(written.settled === undefined ? {} : { settled: written.settled }),
      };
    },
    close,
  };
};