/**
 * Arrays of blocks, nested deltas and payload-discriminated records do not fit
 * the flat descriptor match/field spec used by identity decoding.
 */
import type {
  CompactionState,
  CompactionTrigger,
  HarnessName,
  LimitCode,
} from "../knowledge/descriptor.js";
import { asRecord, readPath as at } from "./shape.js";
import { type UsageEvent, usageEventsOf } from "./usage.js";

export interface ToolEvent {
  readonly kind: "tool";
  readonly name: string;
  readonly input?: unknown;
  readonly callId?: string;
}

export interface ToolResultEvent {
  readonly kind: "tool-result";
  readonly name?: string;
  readonly callId?: string;
  readonly result?: unknown;
  readonly isError?: boolean;
}

export interface ToolDenial {
  readonly tool: string;
  readonly reason: string;
  readonly callId?: string;
  /** JSON input preview, limited to 512 characters; it may be truncated. */
  readonly inputSummary?: string;
}

const inputSummaryOf = (input: unknown): { readonly inputSummary?: string } => {
  try {
    const text = JSON.stringify(input);
    return text === undefined ? {} : { inputSummary: text.slice(0, 512) };
  } catch {
    return {};
  }
};

const callIdOf = (id: unknown): { readonly callId?: string } =>
  typeof id === "string" ? { callId: id } : {};

export type ContentEvent =
  | ToolEvent
  | ToolResultEvent
  | UsageEvent
  | { readonly kind: "token"; readonly text: string }
  | { readonly kind: "message"; readonly role: string; readonly text: string }
  | { readonly kind: "progress"; readonly label: string }
  /** ADR 0009: the harness compacted its own context. Passes through
   * `decode.ts` unchanged, so this shape matches the HarnessEvent arm. */
  | {
      readonly kind: "compaction";
      readonly state: CompactionState;
      readonly trigger?: CompactionTrigger;
      readonly tokensBefore?: number;
      readonly tokensAfter?: number;
      readonly durationMs?: number;
      readonly detail?: string;
    }
  /** `terminal: true` marks an error that ended the turn (a failed result
   * record); the runner turns it into a task failure. Other errors are
   * informational and the turn goes on. `provisional: true` qualifies a
   * terminal error: the harness may still supersede it with a successful
   * assistant message in the same run (pi retries a stopReason-error
   * attempt in-process), so the decoder holds the verdict and settles it
   * at stream/turn end (see `settleProvisionalError`). */
  | {
      readonly kind: "error";
      readonly message: string;
      readonly terminal?: boolean;
      readonly provisional?: boolean;
      /** A refused tool call or rejected question:
       * the native tool name and the harness's reason, which may be
       * empty. Machine consumers branch on presence, never on prose. */
      readonly denial?: ToolDenial;
    }
  | { readonly kind: "budget"; readonly detail: string }
  /** A structured limit record on the stream (claude's rate_limit_event);
   * the runner turns it into a limit-class failure. `resetsAt` is epoch
   * milliseconds when the harness reported one. */
  | {
      readonly kind: "limit";
      readonly code: LimitCode;
      readonly detail: string;
      readonly resetsAt?: number;
    };

/** Text of an array of `{type:"text", text}` content blocks. */
const textOfBlocks = (content: unknown): string =>
  Array.isArray(content)
    ? content
        .map((b) => asRecord(b))
        .filter((b): b is Record<string, unknown> => b !== null && b.type === "text")
        .map((b) => (typeof b.text === "string" ? b.text : ""))
        .join("")
    : "";

const numberOr = (value: unknown): number | undefined =>
  typeof value === "number" && Number.isFinite(value) ? value : undefined;

/** claude reports `auto` and `manual`; anything else is left unset rather
 * than guessed, because a wrong trigger is worse than a missing one. */
const claudeTrigger = (value: unknown): CompactionTrigger | undefined =>
  value === "auto" || value === "manual" ? value : undefined;

/** ADR 0009. claude emits up to four records around one compaction, and this
 * decoder reads one record at a time with no state between them, so exactly
 * one record produces the end event. The boundary is that record, because it
 * is the only one carrying numbers. Returning `null` means "not a compaction
 * record", which falls through to the generic progress arm. */
const claudeCompaction = (r: Record<string, unknown>): ContentEvent[] | null => {
  if (r.subtype === "compact_boundary") {
    const meta = asRecord(r.compact_metadata) ?? {};
    const trigger = claudeTrigger(meta.trigger);
    return [
      {
        kind: "compaction",
        state: "compacted",
        ...(trigger !== undefined ? { trigger } : {}),
        // The live stream is snake_case; the saved transcript is camelCase
        // for the same data, and transcript/claude.ts reads that one.
        ...(numberOr(meta.pre_tokens) !== undefined
          ? { tokensBefore: numberOr(meta.pre_tokens) }
          : {}),
        ...(numberOr(meta.post_tokens) !== undefined
          ? { tokensAfter: numberOr(meta.post_tokens) }
          : {}),
        ...(numberOr(meta.duration_ms) !== undefined
          ? { durationMs: numberOr(meta.duration_ms) }
          : {}),
      },
    ];
  }
  if (r.subtype !== "status") return null;
  if (r.status === "compacting") return [{ kind: "compaction", state: "started" }];
  if (r.compact_result === "failed") {
    const detail = typeof r.compact_error === "string" ? r.compact_error : "";
    return [{ kind: "compaction", state: "failed", ...(detail !== "" ? { detail } : {}) }];
  }
  // The success status is deliberately silent: the boundary record that
  // follows carries the end, and emitting both would give two ends for one
  // compaction, so a caller counting boundaries would double-count.
  if (r.compact_result === "success") return [];
  return null;
};

const claude = (r: Record<string, unknown>): ContentEvent[] => {
  const events: ContentEvent[] = [];
  if (r.type === "assistant") {
    const content = at(r, "message.content");
    if (Array.isArray(content)) {
      for (const raw of content) {
        const block = asRecord(raw);
        if (block?.type === "tool_use" && typeof block.name === "string") {
          events.push({
            kind: "tool",
            name: block.name,
            input: block.input,
            ...callIdOf(block.id),
          });
        }
      }
    }
    const text = textOfBlocks(content);
    if (text !== "") events.push({ kind: "message", role: "assistant", text });
  } else if (r.type === "user") {
    const content = at(r, "message.content");
    if (Array.isArray(content)) {
      for (const raw of content) {
        const block = asRecord(raw);
        if (block?.type !== "tool_result") continue;
        events.push({
          kind: "tool-result",
          ...callIdOf(block.tool_use_id),
          ...(typeof block.name === "string" ? { name: block.name } : {}),
          ...(Object.hasOwn(block, "content") ? { result: block.content } : {}),
          ...(typeof block.is_error === "boolean" ? { isError: block.is_error } : {}),
        });
      }
    }
  } else if (r.type === "stream_event") {
    const delta = at(r, "event.delta");
    const d = asRecord(delta);
    if (d?.type === "text_delta" && typeof d.text === "string") {
      events.push({ kind: "token", text: d.text });
    }
  } else if (r.type === "system" && r.subtype !== "init" && typeof r.subtype === "string") {
    // Compaction records are lossless and replace the progress label they
    // used to carry (ADR 0009); everything else stays droppable progress.
    // Non-init system lines (hook_started etc.) surface as droppable
    // progress, as the pre-refactor decoder did. The init line is the
    // identity announcement, handled upstream.
    const compaction = claudeCompaction(r);
    if (compaction !== null) events.push(...compaction);
    else events.push({ kind: "progress", label: r.subtype });
  } else if (r.type === "result") {
    if (Array.isArray(r.permission_denials)) {
      for (const raw of r.permission_denials) {
        const denial = asRecord(raw);
        if (typeof denial?.tool_name !== "string" || denial.tool_name.trim() === "") continue;
        events.push({
          kind: "error",
          message: `Tool permission denied: ${denial.tool_name}`,
          denial: {
            tool: denial.tool_name,
            reason: "",
            ...callIdOf(denial.tool_use_id),
            ...inputSummaryOf(denial.tool_input),
          },
        });
      }
    }
    if (r.is_error === true) {
      // The native turn verdict is independent of its denial observations.
      // Keep the harness's text for auth and other native failure classification.
      const sub = typeof r.subtype === "string" ? r.subtype : "result error";
      const detail =
        typeof r.result === "string" && r.result.length > 0 ? ` (${r.result.slice(0, 512)})` : "";
      events.push({ kind: "error", message: `turn failed: ${sub}${detail}`, terminal: true });
    }
  } else if (r.type === "rate_limit_event") {
    // Only "rejected" is a limit; "allowed_warning" still serves the request.
    // overageStatus is a separate billing signal, not a rate limit.
    // resetsAt arrives in seconds; the event carries milliseconds. No wall
    // clock is read.
    const info = asRecord(r.rate_limit_info);
    const status = info?.status;
    if (status === "rejected") {
      const raw = info?.resetsAt;
      const milliseconds = typeof raw === "number" ? raw * 1000 : Number.NaN;
      const resetsAt = Number.isFinite(milliseconds) && milliseconds > 0 ? milliseconds : undefined;
      events.push({
        kind: "limit",
        code: "rate-limit",
        detail: `rate_limit_event status=${String(status)}`,
        ...(resetsAt !== undefined ? { resetsAt } : {}),
      });
    }
  }
  return events;
};

// codex item types that are NOT tool activity - everything else on
// item.started is a tool of some kind (command_execution, file_change,
// mcp_tool_call, web_search, ...), surfaced generically by item.type.
const CODEX_NON_TOOL = new Set([
  "agent_message",
  "error",
  "reasoning",
  "todo_list",
  "plan",
  "enteredReviewMode",
  "exitedReviewMode",
]);

// codex app-server session notifications carry ThreadItem types in
// camelCase (verified 0.159.2); the exec --json surface spells them in
// snake_case and the reader below is written against those names. The
// map is closed: a type absent here passes through unchanged.
const CODEX_APPSERVER_ITEM_TYPES: Readonly<Record<string, string>> = {
  agentMessage: "agent_message",
  commandExecution: "command_execution",
  fileChange: "file_change",
  mcpToolCall: "mcp_tool_call",
  todoList: "todo_list",
  webSearch: "web_search",
};

// Session-only item echoes that must not surface: the user's own input
// (userMessage), harness-internal hook prompt frames, and the compaction
// item (compaction stays unreported on codex - compactionReporting is
// null because the exec surface marks nothing; a session-only mapping
// would claim reporting the one-shot surface does not have).
const CODEX_APPSERVER_DROP_ITEMS = new Set(["userMessage", "hookPrompt", "contextCompaction"]);

/** One codex app-server JSON-RPC notification, restated as the exec-style
 * record the reader below already understands; null drops it. Field names
 * live here (ADR 0005) - the runner only ever sees decoded kinds. */
const codexNotificationRecord = (r: Record<string, unknown>): Record<string, unknown> | null => {
  const params = asRecord(r.params);
  if (params === null) return null;
  switch (r.method) {
    case "item/started":
    case "item/completed": {
      const rawItem = asRecord(params.item);
      if (rawItem === null) return null;
      const itemType = typeof rawItem.type === "string" ? rawItem.type : "";
      if (CODEX_APPSERVER_DROP_ITEMS.has(itemType)) return null;
      return {
        item: {
          ...rawItem,
          type: CODEX_APPSERVER_ITEM_TYPES[itemType] ?? itemType,
        },
        type: r.method === "item/started" ? "item.started" : "item.completed",
      };
    }
    case "error": {
      const error = asRecord(params.error);
      return {
        type: "error",
        ...(typeof error?.message === "string" ? { message: error.message } : {}),
        // A retriable error does not end the turn; only its final form does.
        willRetry: params.willRetry,
      };
    }
    case "turn/completed": {
      // The delimiter itself is classified by the session record reader;
      // content-wise only the FAILED verdict carries news (the terminal
      // error message), mirroring exec --json's turn.failed record.
      const turn = asRecord(params.turn);
      if (turn === null || turn.status !== "failed") return null;
      return { error: turn.error, type: "turn.failed" };
    }
    default:
      // turn/started, thread/started (identity rides the open response in
      // session mode), status changes and MCP/hook noise have no content here.
      return null;
  }
};

const codex = (raw: Record<string, unknown>): ContentEvent[] => {
  // Session mode feeds the same reader JSON-RPC notifications; restate
  // them first so both surfaces share one item vocabulary.
  const r =
    typeof raw.method === "string"
      ? (codexNotificationRecord(raw) ?? { type: "__dropped__" })
      : raw;
  if (r.type === "turn.failed" || r.type === "error") {
    const error = r.type === "error" ? r : asRecord(r.error);
    return [
      {
        kind: "error",
        message: typeof error?.message === "string" ? error.message : "Codex turn failed",
        // A retriable app-server error notification does not end the turn.
        terminal: r.willRetry !== true,
      },
    ];
  }
  const item = asRecord(r.item);
  if (item === null) return [];
  // Tool activity is emitted once on start so it is not double-counted
  // against the completion record. Any non-message/reasoning item is a tool;
  // name it by its item.type, carry the most useful field as input.
  if (
    r.type === "item.started" &&
    typeof item.type === "string" &&
    !CODEX_NON_TOOL.has(item.type)
  ) {
    const name = item.type === "command_execution" ? "shell" : item.type;
    const input = item.command ?? item.changes ?? item.query ?? item.invocation ?? undefined;
    return [{ kind: "tool", name, input, ...callIdOf(item.id) }];
  }
  if (r.type !== "item.completed") return [];
  if (item.type === "agent_message" && typeof item.text === "string") {
    return [{ kind: "message", role: "assistant", text: item.text }];
  }
  if (item.type === "error" && typeof item.message === "string") {
    // Codex uses item errors for notices too. The item completing does not
    // establish turn failure; retain the diagnostic without inventing one.
    return [{ kind: "error", message: item.message }];
  }
  if (typeof item.type !== "string" || CODEX_NON_TOOL.has(item.type)) return [];
  const exitCode = numberOr(item.exit_code ?? item.exitCode);
  const mcpResult = asRecord(item.result);
  let isError: boolean | undefined;
  if (
    item.status === "failed" ||
    item.status === "declined" ||
    item.error != null ||
    mcpResult?.isError === true
  ) {
    isError = true;
  } else if (item.type === "command_execution") {
    if (exitCode !== undefined) isError = exitCode !== 0;
  } else if (item.status === "completed" || mcpResult?.isError === false) {
    isError = false;
  }
  const nativeItem =
    asRecord(typeof raw.method === "string" ? at(raw, "params.item") : raw.item) ?? item;
  return [
    {
      kind: "tool-result",
      name: item.type === "command_execution" ? "shell" : item.type,
      ...callIdOf(item.id),
      result: nativeItem,
      ...(isError === undefined ? {} : { isError }),
    },
  ];
};

/**
 * pi's compaction `reason` onto ADR 0009's shared trigger vocabulary.
 * `threshold` is pi's automatic trigger - it fires when contextTokens
 * exceeds contextWindow - reserveTokens - so it normalizes to `auto`.
 * `manual` and `overflow` carry over by name.
 *
 * Only `threshold` was observed live on 0.87.0; the other two are read
 * from pi's own source and docs/rpc.md
 * (docs/research/2026-09-22-compaction-signals/pi).
 */
const piTrigger = (value: unknown): CompactionTrigger | undefined => {
  if (value === "threshold") return "auto";
  if (value === "manual" || value === "overflow") return value;
  return undefined;
};

/**
 * pi brackets every compaction with `compaction_start` and
 * `compaction_end`. Both carry `reason`; the end record carries the
 * payload. Returns null for a record that is not one of the two, so the
 * caller falls through to the rest of the pi decoder.
 *
 * Either record can arrive OUTSIDE an agent_start/agent_settled pair -
 * before the turn exists, or after a run completes - so nothing here keys
 * off turn boundaries. openSession buffers between-turn events for the
 * next turn, and `compaction` is not droppable, so it survives that wait.
 */
const piCompaction = (r: Record<string, unknown>): ContentEvent[] | null => {
  const trigger = piTrigger(r.reason);
  const withTrigger = trigger !== undefined ? { trigger } : {};
  if (r.type === "compaction_start") {
    return [{ kind: "compaction", state: "started", ...withTrigger }];
  }
  if (r.type !== "compaction_end") return null;
  // pi emits `result: undefined` on all three abort and failure paths, and
  // both modes serialize through JSON.stringify, which drops
  // undefined-valued keys. On the wire the key is therefore ABSENT, not
  // null - docs/rpc.md says null, the implementation and the serializer say
  // absent. asRecord answers the same for absent, null and non-object, so
  // this arm holds whichever of the two is true on a given build.
  const result = asRecord(r.result);
  if (result === null) {
    // `aborted` separates a stopped run from a failed one. Neither was
    // observed live; both are source readings, and each reports its own
    // state rather than a `compacted` that carries no numbers.
    return [
      { kind: "compaction", state: r.aborted === true ? "aborted" : "failed", ...withTrigger },
    ];
  }
  return [
    {
      kind: "compaction",
      state: "compacted",
      ...withTrigger,
      ...(numberOr(result.tokensBefore) !== undefined
        ? { tokensBefore: numberOr(result.tokensBefore) }
        : {}),
      // pi's own name for the after-count is an estimate, and hcn reports
      // it as the harness gave it rather than recomputing anything.
      ...(numberOr(result.estimatedTokensAfter) !== undefined
        ? { tokensAfter: numberOr(result.estimatedTokensAfter) }
        : {}),
    },
  ];
};

const pi = (r: Record<string, unknown>): ContentEvent[] => {
  // ADR 0009. Ahead of everything else: these two record types collide with
  // nothing below, and a null return falls straight through.
  const compaction = piCompaction(r);
  if (compaction !== null) return compaction;
  // One clean tool event per invocation (the toolcall_start/delta/end and
  // tool_execution_update stream is noise; tool_execution_start fires once).
  if (r.type === "tool_execution_start" && typeof r.toolName === "string") {
    return [{ kind: "tool", name: r.toolName, input: r.args, ...callIdOf(r.toolCallId) }];
  }
  if (r.type === "tool_execution_end") {
    if (
      typeof r.toolName !== "string" &&
      typeof r.toolCallId !== "string" &&
      !Object.hasOwn(r, "result")
    )
      return [];
    return [
      {
        kind: "tool-result",
        ...(typeof r.toolName === "string" ? { name: r.toolName } : {}),
        ...callIdOf(r.toolCallId),
        ...(Object.hasOwn(r, "result") ? { result: r.result } : {}),
        ...(typeof r.isError === "boolean" ? { isError: r.isError } : {}),
      },
    ];
  }
  if (r.type === "message_update") {
    const ev = asRecord(r.assistantMessageEvent);
    if (ev?.type === "text_delta" && typeof ev.delta === "string") {
      return [{ kind: "token", text: ev.delta }];
    }
    return [];
  }
  if (r.type === "message_end") {
    const message = asRecord(r.message);
    if (message?.role === "assistant") {
      // A turn that ends with stopReason "error" is a provider/auth/token
      // failure pi does NOT print to stderr and exits 0 for (verified with
      // an expired minimax token: empty content, stopReason error, clean
      // exit). Without this the failure is invisible - a silent empty turn.
      // It is also PROVISIONAL: pi retries inside one run (verified live on
      // pi 0.85.1, test/fixtures/harnesses/pi-terminated-recovered.ndjson:
      // resuming a session created under a different model/provider aborts
      // the first continuation once - errorMessage "terminated", pi's own
      // fetch-abort wording - then the next assistant message_end completes
      // the turn). The decoder therefore holds the terminal claim until no
      // successful assistant message can supersede it; the error-then-
      // nothing shape (minimax) still settles into a loud failure.
      if (message.stopReason === "error") {
        const msg =
          typeof message.errorMessage === "string"
            ? `pi turn ended with stopReason error: ${message.errorMessage}`
            : "pi turn ended with stopReason error (provider/auth failure)";
        return [
          {
            kind: "error",
            message: msg,
            terminal: true,
            provisional: true,
          },
        ];
      }
      const text = textOfBlocks(message.content);
      if (text !== "") return [{ kind: "message", role: "assistant", text }];
      // A reply that ended normally (stop) or on the output cap (length)
      // with no answer text - typically reasoning only - is otherwise a
      // clean run with no message. Report the stop reason and the token
      // counts so the caller can tell a reasoning-exhausted reply from an
      // empty one. Provisional like stopReason error: a later assistant
      // message with text in the same run supersedes it.
      if (message.stopReason === "stop" || message.stopReason === "length") {
        return [
          {
            kind: "error",
            message: `pi turn ended with stopReason ${message.stopReason} and no text (${piUsage(message.usage)})`,
            terminal: true,
            provisional: true,
          },
        ];
      }
    }
  }
  return [];
};

/** pi's token counts for a report: the fields present, in a fixed order. */
const piUsage = (value: unknown): string => {
  const usage = asRecord(value);
  const parts = (["input", "output", "reasoning"] as const).flatMap((field) => {
    const count = numberOr(usage?.[field]);
    return count === undefined ? [] : [`${field} ${count}`];
  });
  return parts.length === 0 ? "usage not reported" : `usage: ${parts.join(", ")}`;
};

const muse = (r: Record<string, unknown>): ContentEvent[] => {
  const payload = asRecord(r.payload);
  if (payload === null) return [];
  // Tool activity: muse emits a tool_result naming the tool in
  // correlation_facts.tool_name, with the command inside the result text.
  if (payload.kind === "tool_result") {
    const facts = asRecord(payload.correlation_facts);
    const name = typeof facts?.tool_name === "string" ? facts.tool_name : "tool";
    // Shell tools carry a JSON result with a `command`; file tools carry
    // plain prose. Surface the command when present, else leave input off
    // (the tool name is the signal that matters).
    let command: unknown;
    if (typeof payload.text === "string") {
      try {
        command = (JSON.parse(payload.text) as Record<string, unknown>).command;
      } catch {
        command = undefined;
      }
    }
    const outcome = facts?.outcome;
    const isError =
      outcome === "success"
        ? false
        : outcome === "error" ||
            outcome === "failure" ||
            outcome === "failed" ||
            outcome === "rejected"
          ? true
          : undefined;
    return [
      { kind: "tool", name, input: command, ...callIdOf(payload.call_id) },
      {
        kind: "tool-result",
        name,
        ...callIdOf(payload.call_id),
        ...(Object.hasOwn(payload, "text") ? { result: payload.text } : {}),
        ...(isError === undefined ? {} : { isError }),
      },
    ];
  }
  if (payload.kind === "run_output_delta" && typeof payload.text === "string") {
    return [{ kind: "token", text: payload.text }];
  }
  if (payload.kind === "run_terminal") {
    if (
      payload.terminal === "completed" &&
      typeof payload.text === "string" &&
      payload.text !== ""
    ) {
      return [{ kind: "message", role: "assistant", text: payload.text }];
    }
    if (payload.terminal === "failed") {
      const reason = typeof payload.reason === "string" ? payload.reason : "run failed";
      // The muse reader is the muse-specific seam; no descriptor field
      // carries budget phrasings yet, so the pattern lives here.
      if (/did not reach a terminal state within \d+ step/i.test(reason)) {
        return [{ kind: "budget", detail: reason }];
      }
      return [{ kind: "error", message: `muse run failed: ${reason}`, terminal: true }];
    }
  }
  return [];
};

const compactDetail = (value: unknown): string => {
  if (typeof value === "string") return value.slice(0, 512);
  try {
    return JSON.stringify(value).slice(0, 512);
  } catch {
    return String(value).slice(0, 512);
  }
};

/** Popeye `--mode hcn` contract (verified 0.1.0, probes 2026-09-24).
 * identity opens the stream and carries the harness-minted sessionId;
 * token carries text deltas; message carries the settled assistant text;
 * tool carries tool-call starts; compaction carries started/compacted;
 * error with terminal:true ended the turn; done/failure delimit it.
 * done and failure are turn delimiters, not content: []. */
const popeyeReader = (r: Record<string, unknown>): ContentEvent[] => {
  if (r.kind === "token" && typeof r.text === "string") {
    return [{ kind: "token", text: r.text }];
  }
  if (r.kind === "message" && r.role === "assistant" && typeof r.text === "string") {
    return r.text === "" ? [] : [{ kind: "message", role: "assistant", text: r.text }];
  }
  // No popeye hcn stream carries a tool record today (toolCalls stay
  // kernel-side per the P1 taxonomy); the branch decodes one if it lands.
  if (r.kind === "tool" && typeof r.name === "string") {
    return [{ kind: "tool", name: r.name, input: r.input }];
  }
  if (r.kind === "compaction" && (r.state === "started" || r.state === "compacted")) {
    return [{ kind: "compaction", state: r.state }];
  }
  if (r.kind === "error" && typeof r.message === "string") {
    return [
      r.terminal === true
        ? { kind: "error", message: r.message, terminal: true }
        : { kind: "error", message: r.message },
    ];
  }
  // RPC session records: progress assistantText streams as tokens; the
  // settled snapshot carries the turn's assistant entries as its message.
  if (r._tag === "assistantText" && typeof r.text === "string") {
    return [{ kind: "token", text: r.text }];
  }
  const result = asRecord(r.result);
  if (result !== null && result._tag === "snapshot" && Array.isArray(result.entries)) {
    const texts: string[] = [];
    for (const line of result.entries) {
      const entry = asRecord(line);
      const payload = asRecord(entry?.payload);
      if (entry?.kind === "message" && payload?.role === "assistant") {
        const text = typeof payload.content === "string" ? payload.content : "";
        if (text !== "") texts.push(text);
      }
    }
    if (texts.length > 0) return [{ kind: "message", role: "assistant", text: texts.join("\n") }];
  }
  return [];
};

/** Antigravity stream-json contract. Permission errors are decoded
 * only from structured tool_info.error fields. General model prose is not
 * evidence that a native permission rule denied a call. */ const antigravity = (
  r: Record<string, unknown>,
): ContentEvent[] => {
  if (r.event === "step_update") {
    const step = asRecord(r.step_update);
    if (step === null) return [];
    if (step.step_type === "agent_response" && typeof step.text_delta === "string") {
      return [{ kind: "token", text: step.text_delta }];
    }
    // ADR 0009 / #240. Antigravity announces compaction with one record and
    // no payload: a `checkpoint` step in state DONE. There is no ACTIVE
    // phase, no summary, no reason and no token counts - every compaction
    // in the 1.2.8 probe emitted exactly this shape and nothing else, so
    // `compacted` is the only state antigravity can report and no `started`
    // precedes it (docs/research/2026-09-22-compaction-signals/antigravity).
    //
    // DONE is the record's own completion evidence, and it is the only
    // outcome evidence it carries. The token count on the surrounding
    // records is deliberately NOT read as a signal: it drops with and
    // without compaction, so it would report compactions that never ran.
    if (step.step_type === "checkpoint") {
      if (step.state !== "DONE") return [];
      // Antigravity reports seconds and this event's unit is milliseconds,
      // so the value is converted and rounded to the millisecond. That
      // conversion is the whole of it; no other number is derived.
      const seconds = numberOr(step.duration_seconds);
      return [
        {
          kind: "compaction",
          state: "compacted",
          ...(seconds !== undefined ? { durationMs: Math.round(seconds * 1000) } : {}),
        },
      ];
    }
    if (step.step_type !== "tool" || (step.state !== "DONE" && step.state !== "ERROR")) return [];

    const info = asRecord(step.tool_info);
    const name =
      typeof info?.name === "string"
        ? info.name
        : typeof step.tool_name === "string"
          ? step.tool_name
          : "tool";
    const tool: ContentEvent = {
      kind: "tool",
      name,
      ...(info !== null && Object.hasOwn(info, "parameters") ? { input: info.parameters } : {}),
    };
    const nativeError = info?.error;
    const results: ContentEvent[] = [tool];
    if (info !== null && (Object.hasOwn(info, "output") || nativeError != null)) {
      results.push({
        kind: "tool-result",
        name,
        result: nativeError != null ? info : info.output,
        // An ERROR state or tool_info.error marks failure; DONE output
        // alone does not report whether a shell command succeeded.
        ...(nativeError != null || step.state === "ERROR" ? { isError: true } : {}),
      });
    }
    if (nativeError === undefined || nativeError === null) return results;

    const error = asRecord(nativeError);
    const type = typeof error?.type === "string" ? error.type : "";
    const reason = typeof error?.message === "string" ? error.message : compactDetail(nativeError);
    const isDenial = /permission|denied|rejected|not[_ -]?allowed/i.test(`${type} ${reason}`);
    return [
      ...results,
      {
        kind: "error",
        message: `antigravity tool ${name} failed: ${reason}`,
        ...(isDenial ? { denial: { tool: name, reason } } : {}),
      },
    ];
  }

  if (r.event !== "result") return [];
  const result = asRecord(r.result);
  if (result === null) {
    return [{ kind: "error", message: "antigravity result was malformed", terminal: true }];
  }
  const status = typeof result.status === "string" ? result.status : "INVALID";
  if (status !== "SUCCESS") {
    const detail = Object.hasOwn(result, "error")
      ? compactDetail(result.error)
      : "no native error detail";
    return [
      {
        kind: "error",
        message: `antigravity turn ended with status ${status}: ${detail}`,
        terminal: true,
      },
    ];
  }
  return typeof result.response === "string" && result.response !== ""
    ? [{ kind: "message", role: "assistant", text: result.response }]
    : [];
};

/** Opaque per-reader decoder state (cursor only in v1): pending tool
 * call ids, tombstoned ids evicted from pending that must never re-emit
 * a tool event, completed ids that must not re-emit on a duplicate
 * completion, and query args by toolCallId for completions that arrive
 * with no preceding start. Owned by interpretation, threaded by execution
 * without inspection; reset per turn. Each collection is bounded
 * (drop-oldest) so a pathological turn cannot grow memory without bound.
 * Completions land in their own set precisely so settling normal calls
 * never pushes an evicted id out of the tombstones. */
export interface CursorReaderState {
  readonly pending: readonly string[];
  readonly tombstones: readonly string[];
  readonly completed: readonly string[];
  readonly queryArgs: readonly { readonly toolCallId: string; readonly args: unknown }[];
}

export const freshCursorReaderState = (): CursorReaderState => ({
  pending: [],
  tombstones: [],
  completed: [],
  queryArgs: [],
});

/** Per-harness reader state: cursor state for cursor, null otherwise.
 * Execution holds it without inspection (ADR 0005). */
export type ReaderState = CursorReaderState | null;

const STATE_FACTORIES: Record<HarnessName, () => ReaderState> = {
  claude: () => null,
  codex: () => null,
  pi: () => null,
  muse: () => null,
  cursor: freshCursorReaderState,
  antigravity: () => null,
  popeye: () => null,
};

export const freshReaderState = (harness: HarnessName): ReaderState => STATE_FACTORIES[harness]();

const assistantText = (r: Record<string, unknown>): string => {
  const message = asRecord(r.message);
  if (message === null) return "";
  return textOfBlocks(message.content);
};

/** One collection bound for every cursor reader map: drop-oldest past it. */
const CURSOR_STATE_BOUND = 128;

const boundedAdd = (ids: readonly string[], id: string): readonly string[] =>
  [...ids.filter((known) => known !== id), id].slice(-CURSOR_STATE_BOUND);

const boundedForget = (ids: readonly string[], id: string): readonly string[] =>
  ids.filter((known) => known !== id);

/** The native *ToolCall entry of a tool_call record, if it names one. */
const toolCallEntry = (
  r: Record<string, unknown>,
): { readonly name: string; readonly call: Record<string, unknown> } | null => {
  const toolCall = asRecord(r.tool_call);
  if (toolCall === null) return null;
  for (const [key, value] of Object.entries(toolCall)) {
    if (key.endsWith("ToolCall")) {
      const call = asRecord(value);
      if (call !== null) return { name: key, call };
    }
  }
  return null;
};

/** Query args recorded under one toolCallId, both observed paths: the id
 * sits beside args for ask/createPlan and inside args for web
 * fetch/search. A future query kind works without a code change. */
const queryRecordOf = (
  query: Record<string, unknown>,
): { readonly toolCallId: string; readonly args: unknown } | null => {
  for (const value of Object.values(query)) {
    const inner = asRecord(value);
    if (inner === null) continue;
    if (typeof inner.toolCallId === "string") {
      return { toolCallId: inner.toolCallId, args: asRecord(inner.args) ?? inner };
    }
    const args = asRecord(inner.args);
    if (args !== null && typeof args.toolCallId === "string") {
      return { toolCallId: args.toolCallId, args };
    }
  }
  return null;
};

/** Input for a completion with no preceding start: the recorded query
 * args matched by toolCallId, else the completion's own args, else none. */
const unknownInputOf = (
  state: CursorReaderState,
  r: Record<string, unknown>,
  entry: { readonly call: Record<string, unknown> },
): unknown => {
  const toolCall = asRecord(r.tool_call);
  const id =
    typeof toolCall?.toolCallId === "string"
      ? toolCall.toolCallId
      : typeof r.call_id === "string"
        ? r.call_id
        : null;
  const recorded = id === null ? undefined : state.queryArgs.find((q) => q.toolCallId === id)?.args;
  return recorded ?? asRecord(entry.call.args) ?? undefined;
};

const denialEvent = (tool: string, reason: string): ContentEvent => ({
  kind: "error",
  message:
    reason === "" ? `cursor denied ${tool} (no reason given)` : `cursor denied ${tool}: ${reason}`,
  denial: { tool, reason },
});

type ToolOutcome =
  | { readonly kind: "success"; readonly isError?: boolean }
  | { readonly kind: "error"; readonly isError: true }
  | { readonly kind: "unknown" }
  | { readonly kind: "rejected"; readonly reason: string; readonly isError: true };

const toolOutcome = (call: Record<string, unknown>, toolName: string): ToolOutcome | null => {
  const result = asRecord(call.result);
  if (result === null || Object.keys(result).length === 0) return null;
  const rejected = asRecord(result.rejected);
  if (rejected !== null)
    return {
      kind: "rejected",
      isError: true,
      reason: typeof rejected.reason === "string" ? rejected.reason : "",
    };
  if (result.error != null) return { kind: "error", isError: true };
  if (Object.hasOwn(result, "success")) {
    const exitCode = numberOr(at(result, "success.exitCode"));
    if (toolName === "shellToolCall" && exitCode === undefined) return { kind: "success" };
    return { kind: "success", isError: exitCode !== undefined && exitCode !== 0 };
  }
  return { kind: "unknown" };
};

const cursorWithState = (
  r: Record<string, unknown>,
  state: CursorReaderState,
): { readonly events: ContentEvent[]; readonly state: CursorReaderState } => {
  if (r.type === "retry" || r.type === "connection") {
    const label = typeof r.subtype === "string" ? r.subtype : r.type;
    return { events: [{ kind: "progress", label }], state };
  }
  if (r.type === "system") {
    // The init line is the identity announcement, handled upstream. Named
    // liveness lines surface as droppable progress; any other system
    // subtype is ignored like any unknown shape.
    if (r.subtype === "task_notification" || r.subtype === "background_shell_timeout") {
      return { events: [{ kind: "progress", label: r.subtype }], state };
    }
    return { events: [], state };
  }
  if (r.type === "tool_call" && r.subtype === "started") {
    // One tool event per invocation, on start (the pi/codex precedent).
    // No canonical mapping in v1: the tools surface is null, so the
    // native *ToolCall key is data. Ids evicted from pending by the bound
    // tombstone, so a late completion for them cannot double-count.
    const entry = toolCallEntry(r);
    if (typeof r.call_id !== "string" || entry === null) return { events: [], state };
    const args = asRecord(entry.call.args);
    const pending = boundedAdd(state.pending, r.call_id);
    const evicted = state.pending.filter((id) => !pending.includes(id));
    return {
      events: [
        {
          kind: "tool",
          name: entry.name,
          callId: r.call_id,
          ...(args !== null ? { input: args } : {}),
        },
      ],
      state: {
        ...state,
        pending,
        tombstones: [...state.tombstones, ...evicted].slice(-CURSOR_STATE_BOUND),
      },
    };
  }
  if (r.type === "tool_call" && r.subtype === "completed") {
    const entry = toolCallEntry(r);
    const outcome = entry === null ? null : toolOutcome(entry.call, entry.name);
    if (typeof r.call_id !== "string" || entry === null || outcome === null) {
      return { events: [], state };
    }
    const settled: CursorReaderState = {
      ...state,
      pending: boundedForget(state.pending, r.call_id),
      completed: boundedAdd(state.completed, r.call_id),
    };
    const seen =
      state.pending.includes(r.call_id) ||
      state.tombstones.includes(r.call_id) ||
      state.completed.includes(r.call_id);
    const events: ContentEvent[] = [];
    if (!seen) {
      const input = unknownInputOf(state, r, entry);
      events.push({
        kind: "tool",
        name: entry.name,
        callId: r.call_id,
        ...(input !== undefined ? { input } : {}),
      });
    }
    if (!state.completed.includes(r.call_id)) {
      events.push({
        kind: "tool-result",
        name: entry.name,
        callId: r.call_id,
        result: entry.call.result,
        ...(outcome.kind === "unknown" || outcome.isError === undefined
          ? {}
          : { isError: outcome.isError }),
      });
    }
    if (outcome.kind === "rejected") events.push(denialEvent(entry.name, outcome.reason));
    return { events, state: settled };
  }
  if (r.type === "interaction_query") {
    // Handshake internals: record the query args by toolCallId for the
    // unknown-call_id rule above and emit nothing. Response records carry
    // no args and leave the state untouched.
    const query = asRecord(r.query);
    const found = query === null ? null : queryRecordOf(query);
    if (found === null) return { events: [], state };
    return {
      events: [],
      state: {
        ...state,
        queryArgs: [
          ...state.queryArgs.filter((q) => q.toolCallId !== found.toolCallId),
          found,
        ].slice(-CURSOR_STATE_BOUND),
      },
    };
  }
  if (r.type === "result") {
    // Any other subtype, or is_error, is a forward rule to a terminal
    // error through the task failure path.
    if (r.subtype === "success" && r.is_error !== true) return { events: [], state };
    const subtype = typeof r.subtype === "string" ? r.subtype : "result error";
    return {
      events: [{ kind: "error", message: `cursor turn failed: ${subtype}`, terminal: true }],
      state,
    };
  }
  if (r.type === "assistant") {
    const text = assistantText(r);
    if (text === "") return { events: [], state };
    // Partial deltas carry timestamp_ms without model_call_id (probes 11,
    // 43); full-text segments carry model_call_id, with or without a
    // timestamp (probes 13/15), and so does the end-of-turn flush, which
    // carries neither field (probes 11/11b). Timestamp presence alone is
    // NOT the discriminator: plain-stream tool segments carry
    // timestamp_ms and would otherwise decode as tokens.
    if (typeof r.timestamp_ms === "number" && typeof r.model_call_id !== "string") {
      return { events: [{ kind: "token", text }], state };
    }
    return { events: [{ kind: "message", role: "assistant", text }], state };
  }
  return { events: [], state };
};

type StatefulReader = (
  r: Record<string, unknown>,
  state: ReaderState,
) => { readonly events: ContentEvent[]; readonly state: ReaderState };

const stateless =
  (read: (r: Record<string, unknown>) => ContentEvent[]): StatefulReader =>
  (r, state) => ({ events: read(r), state });

const STATEFUL_READERS: Record<HarnessName, StatefulReader> = {
  claude: stateless(claude),
  codex: stateless(codex),
  pi: stateless(pi),
  muse: stateless(muse),
  cursor: (r, state) => cursorWithState(r, state ?? freshCursorReaderState()),
  antigravity: stateless(antigravity),
  popeye: stateless(popeyeReader),
};

/** Stateful entry point: every harness returns { events, state }.
 * decode.ts is the one production call site; contentEventsOf below keeps
 * the array shape for stateless callers. */
export const contentEventsWithState = (
  harness: HarnessName,
  raw: unknown,
  state: ReaderState,
): { readonly events: ContentEvent[]; readonly state: ReaderState } => {
  const record = asRecord(raw);
  if (record === null) return { events: [], state };
  const decoded = STATEFUL_READERS[harness](record, state);
  return { events: [...decoded.events, ...usageEventsOf(harness, record)], state: decoded.state };
};

export const contentEventsOf = (harness: HarnessName, raw: unknown): ContentEvent[] =>
  contentEventsWithState(harness, raw, freshReaderState(harness)).events;