import { execFile } from "node:child_process";
import { readFileSync } from "node:fs";
import { homedir } from "node:os";
import { join } from "node:path";
import type { Adapter, FetchLike } from "../adapter";
import type { AdapterResult, Item, Panel } from "../schema";
import { readAbandoned } from "../abandoned-store";

type FetchFn = FetchLike;

export type HarnessJson = null | boolean | number | string | HarnessJson[] | { [key: string]: HarnessJson };
export type HarnessCapabilities = Record<string, boolean | string | number | null>;

export interface HarnessEvent {
  id: string;
  seq?: number;
  source: string;
  kind: string;
  ts: string;
  taskId?: string | null;
  attemptId?: string | null;
  payload: Record<string, HarnessJson>;
}

export interface HarnessEventsResponse {
  events: HarnessEvent[];
  nextSince: string;
  capabilities: HarnessCapabilities;
  hasMore?: boolean;
}

export interface HarnessConfigField {
  value?: HarnessJson;
  source: string;
  immutable: boolean;
  mutationClass: "launch" | "mid-run" | "global";
  isSecret?: boolean;
  redacted?: boolean;
}

export interface HarnessRunConfigResponse {
  revision: string;
  fields: Record<string, HarnessConfigField>;
}

export interface HarnessEffectivePlanResponse {
  runId: string;
  revision: string;
  planHash: string;
  tasks: HarnessJson[];
  meta: Record<string, HarnessJson> | null;
}

export type HarnessDecisionOption = string | { value: string; meaning?: string; label?: string; recommended?: boolean };

export interface HarnessDecisionEntry {
  id: string;
  task: string | null;
  summary: string | null;
  options: HarnessDecisionOption[];
  requestedAt: string | null;
  status?: string;
  category?: HarnessJson;
  needs?: HarnessJson;
  why?: string;
  blast_radius?: HarnessJson;
  choice?: HarnessJson;
  answeredAt?: HarnessJson;
}

export interface HarnessDecisionsResponse {
  decisions: HarnessDecisionEntry[];
  capabilities: HarnessCapabilities;
}

export interface HarnessRateLimit {
  provider: string;
  state: string;
  resumeAt?: string | null;
  waitMs?: number | null;
  [key: string]: HarnessJson | undefined;
}

export interface HarnessRunDetailResponse {
  run: { ratelimits?: Record<string, HarnessRateLimit>; [key: string]: unknown };
}

export interface HarnessQueueEntry {
  id: string;
  repo: string;
  slug: string;
  preset: string | null;
  account: string | null;
  addedAt: string;
  status: string;
  attempt: number;
  runId: string | null;
  window: HarnessJson | null;
  nextEligibleAt: string | null;
  bypassRequestedAt: string | null;
  terminalAt: string | null;
  failure: HarnessJson | null;
}

export interface HarnessPlansPanelData {
  runs: HarnessPanelRun[];
  queue: HarnessQueueEntry[];
}

export interface HarnessTaskControlRequest {
  attemptId?: string;
  requestId: string;
}

export interface HarnessRunControlRequest {
  requestId?: string;
  [key: string]: HarnessJson | undefined;
}

export interface HarnessApiErrorBody {
  error?: string;
  message?: string;
  [key: string]: HarnessJson | undefined;
}

export class HarnessApiError extends Error {
  constructor(
    readonly status: number,
    readonly body: HarnessApiErrorBody,
    readonly path: string,
  ) {
    super(`harness control-api ${path} failed: HTTP ${status}`);
    this.name = "HarnessApiError";
  }
}

interface HarnessDagNode {
  id: string;
  kind: string;
  label: string;
  status: string;
  meta: { wave: number; state: string; deps: string[]; attempt: number; seat: string; branch: string; failClass?: unknown };
}

interface HarnessDagEdge {
  source: string;
  target: string;
}

interface HarnessPanelTask {
  id: string;
  status: string;
  state?: string;
  seat?: string;
  deps?: string[];
  attempt?: number;
  branch?: string;
  failClass?: string;
  alarmed?: boolean;
}

interface HarnessPanelRun {
  runId: string;
  seq: number;
  title: string;
  status: string;
  state: string;
  owner: string;
  currentTask: string | null;
  tasksTotal: number;
  tasksCompleted: number;
  pendingDecisions: number;
  attempts: number;
  degradedReason: string;
  updatedAt: string | null;
  repoRoot: string;
  registry?: { slug?: string; created?: string; projectName?: string; topic?: string };
  failClass?: string;
  abandoned: boolean;
  abandonedAt: string | null;
  alarmed?: boolean;
  waves: Array<{ wave: number; tasks: HarnessPanelTask[] }>;
  edges: HarnessDagEdge[];
}

interface HarnessRunSummary {
  runId: string;
  seq: number;
  title: string;
  status: string;
  degradedReason: string;
  state: string;
  currentTask: string | null;
  owner: string;
  tasksTotal: number;
  tasksCompleted: number;
  pendingDecisions: number;
  updatedAt: string;
  startedAt?: string | null;
  currentActivityStartedAt?: string | null;
  repoRoot?: string;
  /** Prior attempts of the same (repoRoot, slug); the engine projects them onto one row. */
  attempts?: number;
  attemptRunIds?: string[];
  registry?: { slug?: unknown; created?: unknown; projectName?: unknown; topic?: unknown; [key: string]: unknown };
  dag: { nodes: HarnessDagNode[]; edges: HarnessDagEdge[] };
}

interface HarnessRunsResponse {
  runs: HarnessRunSummary[];
}

interface HarnessQueueResponse {
  queue: HarnessQueueEntry[];
}

interface HarnessDecisionAnswerResponse {
  statusCode?: number;
  applied?: boolean;
  error?: string;
  [key: string]: unknown;
}

interface HarnessSteerResponse {
  ok: boolean;
  id: string | null;
  queued: boolean;
  restart: boolean;
}

interface HarnessTimelineTile {
  key: string;
  label: string;
  value: number;
}

interface HarnessTimelineAttribution {
  cat: string;
  totalMs: number;
}

interface HarnessTimelineRun {
  index: number;
  startTs: string;
  endTs: string;
  durationMs: number;
  outcome: string;
  landed: number;
  quarantined: number;
  skipped: number;
  gate0Fails: number;
  fixerAttempts: number;
}

interface HarnessTimelineSegment {
  t0: number;
  durMs: number;
  cat: string;
  taskId?: string;
  agentId?: string;
}

interface HarnessTimelineResponse {
  runId: string;
  tiles: HarnessTimelineTile[];
  attribution: HarnessTimelineAttribution[];
  runs: HarnessTimelineRun[];
  segments: HarnessTimelineSegment[];
}

export interface HarnessConnection {
  baseUrl: string;
  token: string;
}

interface CachedHarnessConnection {
  connection: HarnessConnection;
  gen: number;
}

class HarnessResponseBodyError extends Error {}

export interface HarnessAttemptLiveness {
  elapsedSecs: number;
  lastActivity: string;
  lastSignalAt: string;
}

export interface HarnessAttemptEntry {
  attemptId: string;
  task?: string;
  phase?: string;
  seat?: string;
  timeoutSecs: number;
  liveness: HarnessAttemptLiveness | null;
}

export interface HarnessAttemptsResponse {
  attempts: HarnessAttemptEntry[];
}

export interface HarnessAdapterOptions {
  id?: string;
  interval?: number;
  fetchImpl?: FetchFn;
  baseUrl?: string;
  token?: string;
  harnessHome?: string;
  requestTimeoutMs?: number;
  requestConcurrency?: number;
  now?: () => number;
  /** Restarts the control-api when a poll cannot reach a current one. Injectable for tests. */
  ensureImpl?: () => Promise<void>;
}

export interface HarnessAdapter extends Adapter {
  /** POST /runs/:runId/decisions/:decisionId — answers a pending decision. */
  answerDecision(runId: string, decisionId: string, choice: string): Promise<HarnessDecisionAnswerResponse>;
  /** POST /runs/:runId/tasks/:taskId/steer — gated for wave-4; the gateway calls this. */
  steerTask(runId: string, taskId: string, body: { text?: string; restart?: boolean }): Promise<HarnessSteerResponse>;
  getRunEvents(runId: string, query?: { since?: string; taskId?: string; raw?: string; lastEventId?: string }): Promise<HarnessEventsResponse>;
  openTaskStream(runId: string, taskId: string, query?: { since?: string; raw?: string; lastEventId?: string; signal?: AbortSignal }): Promise<Response>;
  getRunConfig(runId: string): Promise<HarnessRunConfigResponse>;
  getEffectivePlan(runId: string): Promise<HarnessEffectivePlanResponse>;
  getDecisions(runId: string): Promise<HarnessDecisionsResponse>;
  getRunDetail(runId: string): Promise<HarnessRunDetailResponse>;
  controlTask(runId: string, taskId: string, verb: "kill" | "pause" | "resume", body: HarnessTaskControlRequest): Promise<Record<string, HarnessJson>>;
  controlRun(runId: string, verb: "kill" | "pause" | "resume", body?: HarnessRunControlRequest): Promise<Record<string, HarnessJson>>;
  patchRunConfig(runId: string, patch: Record<string, HarnessJson>, revision: string): Promise<HarnessRunConfigResponse>;
}

/** Run statuses that require human intervention and therefore surface as an unanswered HALT item. */
const HALT_STATUSES = new Set(["failed", "degraded"]);
const DEFAULT_INTERVAL_MS = 5_000;
const DEFAULT_REQUEST_TIMEOUT_MS = 10_000;
const DEFAULT_REQUEST_CONCURRENCY = 8;
const HEARTBEAT_GAP_MS = 180_000;

async function mapWithConcurrency<T, R>(
  values: readonly T[],
  concurrency: number,
  map: (value: T) => Promise<R>,
): Promise<R[]> {
  const results = new Array<R>(values.length);
  let nextIndex = 0;
  const worker = async (): Promise<void> => {
    while (nextIndex < values.length) {
      const index = nextIndex++;
      results[index] = await map(values[index]!);
    }
  };
  const workerCount = Math.min(values.length, Math.max(1, Math.floor(concurrency)));
  await Promise.all(Array.from({ length: workerCount }, worker));
  return results;
}

function defaultHarnessHome(): string {
  return join(homedir(), ".harness");
}

function readConnectionFromDisk(harnessHome: string): HarnessConnection {
  const port = readFileSync(join(harnessHome, "control-api.port"), "utf8").trim();
  const token = readFileSync(join(harnessHome, "token"), "utf8").trim();
  return { baseUrl: `http://127.0.0.1:${port}`, token };
}

function groupNodesByWave(nodes: HarnessDagNode[]): Array<{ wave: number; tasks: HarnessPanelTask[] }> {
  const byWave = new Map<number, HarnessDagNode[]>();
  for (const node of nodes) {
    const wave = node.meta.wave;
    const bucket = byWave.get(wave);
    if (bucket) {
      bucket.push(node);
    } else {
      byWave.set(wave, [node]);
    }
  }
  return [...byWave.entries()]
    .sort(([left], [right]) => left - right)
    .map(([wave, tasks]) => ({
      wave,
      tasks: tasks.map((task) => ({
        id: task.id,
        status: task.status,
        state: task.meta.state,
        seat: task.meta.seat,
        deps: task.meta.deps,
        attempt: task.meta.attempt,
        branch: task.meta.branch,
        ...(typeof task.meta.failClass === "string" ? { failClass: task.meta.failClass } : {}),
      })),
    }));
}

function buildHaltItem(source: string, run: HarnessRunSummary): Item {
  return {
    id: `halt-${run.runId}`,
    source,
    project: typeof run.registry?.projectName === "string" ? run.registry.projectName || undefined : undefined,
    severity: "act",
    kind: "halt",
    title: `${run.title} halted`,
    detail: run.degradedReason || `run status: ${run.status}`,
    ts: run.updatedAt,
    actions: [],
  };
}

export function isAttemptAlarmed(attempt: HarnessAttemptEntry, nowMs: number): boolean {
  if (!attempt.liveness) return false;
  if (attempt.liveness.elapsedSecs > 2 * attempt.timeoutSecs) return true;
  const lastSignalMs = Date.parse(attempt.liveness.lastSignalAt);
  if (!Number.isFinite(lastSignalMs)) return false;
  return nowMs - lastSignalMs > HEARTBEAT_GAP_MS;
}

function newestTimestampMs(candidates: Array<number | null>): number | null {
  const valid = candidates.filter((value): value is number => value !== null && Number.isFinite(value));
  if (valid.length === 0) return null;
  return Math.max(...valid);
}

export function newestJournalEventMs(events: HarnessEvent[]): number | null {
  return newestTimestampMs(events.map((event) => {
    const parsed = Date.parse(event.ts);
    return Number.isFinite(parsed) ? parsed : null;
  }));
}

export function newestAttemptHeartbeatMs(attempts: HarnessAttemptEntry[]): number | null {
  return newestTimestampMs(attempts.map((attempt) => {
    if (!attempt.liveness) return null;
    const parsed = Date.parse(attempt.liveness.lastSignalAt);
    return Number.isFinite(parsed) ? parsed : null;
  }));
}

/** O4 per-run staleness: now − max(newest journal ts, newest attempt heartbeat ts). */
export function computeRunStalenessMs(
  nowMs: number,
  newestJournalMs: number | null,
  newestHeartbeatMs: number | null,
): number | null {
  const lastActivityMs = newestTimestampMs([newestJournalMs, newestHeartbeatMs]);
  if (lastActivityMs === null) return null;
  return nowMs - lastActivityMs;
}

export function isRunAlarmedByStaleness(stalenessMs: number | null): boolean {
  return stalenessMs !== null && stalenessMs > HEARTBEAT_GAP_MS;
}

const ACTIVE_TASK_STATUSES = new Set([
  "running",
  "working",
  "dispatching",
  "retrying",
  "verifying",
  "gating",
  "reviewing",
  "resolving",
]);

function markRunStalenessAlarm(panelRun: HarnessPanelRun): void {
  panelRun.alarmed = true;
  for (const wave of panelRun.waves) {
    for (const task of wave.tasks) {
      if (ACTIVE_TASK_STATUSES.has(task.status.trim().toLowerCase())) task.alarmed = true;
    }
  }
}

function buildLivenessAlertItem(
  source: string,
  run: HarnessRunSummary,
  attempt: HarnessAttemptEntry,
  nowMs: number,
): Item {
  const reasons: string[] = [];
  if (attempt.liveness!.elapsedSecs > 2 * attempt.timeoutSecs) {
    reasons.push(`elapsed ${attempt.liveness!.elapsedSecs}s exceeds 2× timeout (${attempt.timeoutSecs}s)`);
  }
  const lastSignalMs = Date.parse(attempt.liveness!.lastSignalAt);
  if (Number.isFinite(lastSignalMs) && nowMs - lastSignalMs > HEARTBEAT_GAP_MS) {
    reasons.push(`no heartbeat for ${Math.floor((nowMs - lastSignalMs) / 1000)}s`);
  }
  const task = attempt.task ?? attempt.attemptId;
  return {
    id: `liveness-${run.runId}-${attempt.attemptId}`,
    source,
    project: typeof run.registry?.projectName === "string" ? run.registry.projectName || undefined : undefined,
    severity: "warn",
    kind: "alert",
    title: `${run.title} attempt stalled`,
    detail: `task ${task}: ${reasons.join("; ")}`,
    ts: run.updatedAt === "" ? new Date(nowMs).toISOString() : run.updatedAt,
    actions: [],
  };
}

function buildAttemptsUnavailableItem(source: string, run: HarnessRunSummary, nowMs: number): Item {
  return {
    id: `attempts-unavailable-${run.runId}`,
    source,
    project: typeof run.registry?.projectName === "string" ? run.registry.projectName || undefined : undefined,
    severity: "warn",
    kind: "alert",
    title: `${run.title} attempt coverage unavailable`,
    detail: "attempt-liveness coverage unavailable for this run",
    ts: run.updatedAt === "" ? new Date(nowMs).toISOString() : run.updatedAt,
    actions: [],
  };
}

const TERMINAL_PLAN_STATUSES = new Set(["failed", "degraded", "killed", "succeeded", "completed", "done"]);
const ACTIVE_PLAN_STATUSES = new Set([
  "running",
  "executing",
  "gated",
  "queued",
  "attention",
  "blocked",
  "paused",
  "needs-you",
]);

function planActivityMs(run: HarnessRunSummary): number {
  const parsed = Date.parse(run.updatedAt);
  return Number.isFinite(parsed) ? parsed : -Infinity;
}

function isActiveHarnessRun(run: HarnessRunSummary): boolean {
  if (run.pendingDecisions > 0) return true;
  const status = run.status.trim().toLowerCase();
  if (TERMINAL_PLAN_STATUSES.has(status)) return false;
  if (run.tasksTotal > 0 && run.tasksCompleted >= run.tasksTotal) return false;
  return ACTIVE_PLAN_STATUSES.has(status);
}

/** Active/in-progress plans first, then most-recent updatedAt — shared panel order for overview + /plans. */
function sortRunsForDisplay(runs: HarnessRunSummary[]): HarnessRunSummary[] {
  return [...runs].sort((left, right) => {
    const leftActive = isActiveHarnessRun(left) ? 1 : 0;
    const rightActive = isActiveHarnessRun(right) ? 1 : 0;
    if (leftActive !== rightActive) return rightActive - leftActive;
    return planActivityMs(right) - planActivityMs(left);
  });
}

function buildPlansPanel(runs: HarnessRunSummary[], queue: HarnessQueueEntry[], ts: string): Panel & { data: HarnessPlansPanelData } {
  const ordered = sortRunsForDisplay(runs);
  const abandoned = readAbandoned();
  return {
    id: "plans",
    ts,
    data: {
      runs: ordered.map((run) => {
        const failedNodeFailClass = run.dag.nodes.find(
          (node) => node.status === "failed" && typeof node.meta.failClass === "string",
        )?.meta.failClass as string | undefined;
        const registry = run.registry
          ? Object.fromEntries(
              (["slug", "created", "projectName", "topic"] as const)
                .filter((key) => typeof run.registry?.[key] === "string")
                .map((key) => [key, run.registry?.[key]]),
            )
          : undefined;

        return {
          runId: run.runId,
          seq: run.seq,
          title: run.title,
          status: run.status,
          state: run.state,
          owner: run.owner,
          currentTask: run.currentTask,
          tasksTotal: run.tasksTotal,
          tasksCompleted: run.tasksCompleted,
          pendingDecisions: run.pendingDecisions,
          degradedReason: run.degradedReason,
          updatedAt: run.updatedAt === "" ? null : run.updatedAt,
          startedAt: typeof run.startedAt === "string" && run.startedAt !== "" ? run.startedAt : null,
          currentActivityStartedAt: typeof run.currentActivityStartedAt === "string" && run.currentActivityStartedAt !== "" ? run.currentActivityStartedAt : null,
          repoRoot: run.repoRoot ?? "",
          attempts: typeof run.attempts === "number" && run.attempts > 0 ? run.attempts : 1,
          ...(Array.isArray(run.attemptRunIds) ? { attemptRunIds: run.attemptRunIds } : {}),
          ...(registry ? { registry } : {}),
          ...(failedNodeFailClass ? { failClass: failedNodeFailClass } : {}),
          abandoned: run.runId in abandoned,
          abandonedAt: abandoned[run.runId]?.abandonedAt ?? null,
          waves: groupNodesByWave(run.dag.nodes),
          edges: run.dag.edges,
        };
      }),
      queue,
    } satisfies HarnessPlansPanelData,
  };
}

function buildForensicsPanel(runId: string, ts: string, timeline: HarnessTimelineResponse): Panel {
  return {
    id: `forensics:${runId}`,
    ts,
    data: {
      tiles: timeline.tiles,
      attribution: timeline.attribution,
      runs: timeline.runs,
      segments: timeline.segments,
    },
  };
}

export function createHarnessAdapter(opts: HarnessAdapterOptions = {}): HarnessAdapter {
  const id = opts.id ?? "harness";
  const interval = opts.interval ?? DEFAULT_INTERVAL_MS;
  const fetchImpl = opts.fetchImpl ?? fetch;
  const harnessHome = opts.harnessHome ?? defaultHarnessHome();
  const requestTimeoutMs = opts.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS;
  const requestConcurrency = opts.requestConcurrency ?? DEFAULT_REQUEST_CONCURRENCY;
  const now = opts.now ?? Date.now;

  const fixedConnection = opts.baseUrl !== undefined && opts.token !== undefined
    ? { connection: { baseUrl: opts.baseUrl, token: opts.token }, gen: 0 }
    : null;
  let connection: CachedHarnessConnection | null = fixedConnection;
  let nextGeneration = 0;

  function resolveConnection(): CachedHarnessConnection {
    if (!connection) {
      connection = { connection: readConnectionFromDisk(harnessHome), gen: ++nextGeneration };
    }
    return connection;
  }

  // The launch-time ensure script only runs when a plan starts, so a control-api
  // that died or kept serving a superseded engine version would otherwise stay
  // wrong until the next launch. Failures here are reported and ignored: the
  // retry below decides whether the poll survives.
  async function defaultEnsure(): Promise<void> {
    const version = readFileSync(join(harnessHome, "engine", "CURRENT"), "utf8").trim();
    const script = join(harnessHome, "engine", "versions", version, "v2", "bin", "ensure-control-api.sh");
    await new Promise<void>((resolve, reject) => {
      execFile("bash", [script], { timeout: 30_000, env: { ...process.env, V2_HARNESS_HOME: harnessHome } }, (error) => (error ? reject(error) : resolve()));
    });
  }
  const ensureImpl = opts.ensureImpl ?? defaultEnsure;

  async function healthyConnection(): Promise<CachedHarnessConnection> {
    let expected: string | null = null;
    try {
      expected = readFileSync(join(harnessHome, "engine", "CURRENT"), "utf8").trim();
    } catch {}
    const attempt = async (): Promise<CachedHarnessConnection> => {
      const candidate = resolveConnection();
      let health: Record<string, HarnessJson> | null = null;
      try {
        health = await requestVia<Record<string, HarnessJson>>(candidate, "/health");
      } catch (error) {
        // An HTTP error proves a live listener; anything else is a dead socket.
        if (!(error instanceof HarnessApiError)) throw error;
      }
      if (health && expected !== null && typeof health.version === "string" && health.version !== expected) {
        invalidateConnection(candidate);
        throw new Error(`harness control-api serves ${String(health.version)}, engine CURRENT is ${expected}`);
      }
      return candidate;
    };
    try {
      return await attempt();
    } catch (cause) {
      if (!fixedConnection) connection = null;
      try {
        await ensureImpl();
      } catch (error) {
        console.error(`harness control-api ensure failed (${String(cause)}): ${String(error)}`);
      }
      return attempt();
    }
  }

  function invalidateConnection(candidate: CachedHarnessConnection) {
    if (!fixedConnection && connection?.gen === candidate.gen) connection = null;
  }

  function isRecord(value: unknown): value is Record<string, HarnessJson> {
    return value !== null && typeof value === "object" && !Array.isArray(value);
  }

  function isHarnessJson(value: unknown): value is HarnessJson {
    if (value === null || typeof value === "boolean" || typeof value === "string") return true;
    if (typeof value === "number") return Number.isFinite(value);
    if (Array.isArray(value)) return value.every(isHarnessJson);
    return isRecord(value) && Object.values(value).every(isHarnessJson);
  }

  function isCapabilities(value: unknown): value is HarnessCapabilities {
    return isRecord(value) && Object.values(value).every((entry) => entry === null || typeof entry === "boolean" || typeof entry === "string" || (typeof entry === "number" && Number.isFinite(entry)));
  }

  const credentialKey = /(?:token|secret|password|api[ _-]?key|authorization|cookie|private[ _-]?key)/i;
  const MAX_REDACTION_DEPTH = 32;

  function redact(value: HarnessJson, depth = 0): HarnessJson {
    if (depth >= MAX_REDACTION_DEPTH) return "[REDACTED]";
    if (Array.isArray(value)) return value.map((item) => redact(item, depth + 1));
    if (!isRecord(value)) return value;
    return Object.fromEntries(Object.entries(value).map(([key, nested]) => [
      key, credentialKey.test(key) ? "[REDACTED]" : redact(nested, depth + 1),
    ]));
  }

  async function parseJson(path: string, res: Response): Promise<Record<string, HarnessJson>> {
    let raw: string;
    try {
      raw = await res.text();
    } catch {
      throw new HarnessResponseBodyError(`unable to read harness response for ${path}`);
    }
    let body: unknown;
    try {
      body = JSON.parse(raw);
    } catch {
      throw new Error(`invalid harness response for ${path}`);
    }
    if (!isRecord(body)) throw new Error(`invalid harness response for ${path}`);
    const safeBody = redact(body) as HarnessApiErrorBody;
    if (!res.ok) throw new HarnessApiError(res.status, safeBody, path);
    return body;
  }

  async function requestVia<T>(candidate: CachedHarnessConnection, path: string, init?: RequestInit, validate?: (body: Record<string, HarnessJson>) => T): Promise<T> {
    const { baseUrl, token } = candidate.connection;
    const timeoutSignal = AbortSignal.timeout(requestTimeoutMs);
    let res: Response;
    try {
      res = await fetchImpl(`${baseUrl}${path}`, {
        ...init,
        signal: init?.signal ? AbortSignal.any([init.signal, timeoutSignal]) : timeoutSignal,
        headers: {
          ...(init?.headers ?? {}),
          authorization: `Bearer ${token}`,
        },
      });
    } catch (error) {
      invalidateConnection(candidate);
      throw error;
    }
    if (res.status === 401) invalidateConnection(candidate);
    let body: Record<string, HarnessJson>;
    try {
      body = await parseJson(path, res);
    } catch (error) {
      if (error instanceof HarnessResponseBodyError) invalidateConnection(candidate);
      throw error;
    }
    return validate ? validate(body) : redact(body) as T;
  }

  async function request<T>(path: string, init?: RequestInit, validate?: (body: Record<string, HarnessJson>) => T): Promise<T> {
    return requestVia(resolveConnection(), path, init, validate);
  }

  async function streamVia(candidate: CachedHarnessConnection, path: string, init?: RequestInit): Promise<Response> {
    const { baseUrl, token } = candidate.connection;
    let res: Response;
    try {
      res = await fetchImpl(`${baseUrl}${path}`, {
        ...init,
        headers: { ...(init?.headers ?? {}), authorization: `Bearer ${token}` },
      });
    } catch (error) {
      invalidateConnection(candidate);
      throw error;
    }
    if (res.status === 401) invalidateConnection(candidate);
    if (!res.ok) {
      try {
        await parseJson(path, res);
      } catch (error) {
        if (error instanceof HarnessResponseBodyError) invalidateConnection(candidate);
        throw error;
      }
    }
    if (!res.headers.get("content-type")?.toLowerCase().includes("text/event-stream")) {
      throw new Error(`invalid harness response for ${path}`);
    }
    return res;
  }

  async function stream(path: string, init?: RequestInit): Promise<Response> {
    return streamVia(resolveConnection(), path, init);
  }

  function requireEvents(body: Record<string, HarnessJson>, path: string): HarnessEventsResponse {
    if (!Array.isArray(body.events) || !isCapabilities(body.capabilities) || typeof body.nextSince !== "string"
      || (body.hasMore !== undefined && typeof body.hasMore !== "boolean")) {
      throw new Error(`invalid harness response for ${path}`);
    }
    for (const event of body.events) {
      if (!isRecord(event) || typeof event.id !== "string" || typeof event.source !== "string" || typeof event.kind !== "string"
        || typeof event.ts !== "string" || !(event.taskId === undefined || event.taskId === null || typeof event.taskId === "string")
        || !(event.attemptId === undefined || event.attemptId === null || typeof event.attemptId === "string") || !isRecord(event.payload)
        || !Object.values(event.payload).every(isHarnessJson)) {
        throw new Error(`invalid harness response for ${path}`);
      }
    }
    return body as unknown as HarnessEventsResponse;
  }

  function requireQueue(body: Record<string, HarnessJson>, path: string): HarnessQueueResponse {
    if (!Array.isArray(body.queue)) throw new Error(`invalid harness response for ${path}`);
    for (const entry of body.queue) {
      if (!isRecord(entry) || typeof entry.id !== "string" || typeof entry.repo !== "string" || typeof entry.slug !== "string"
        || !(entry.preset === null || typeof entry.preset === "string") || !(entry.account === null || typeof entry.account === "string")
        || typeof entry.addedAt !== "string" || typeof entry.status !== "string" || typeof entry.attempt !== "number" || !Number.isFinite(entry.attempt)
        || !(entry.runId === null || typeof entry.runId === "string") || !(entry.window === null || isHarnessJson(entry.window))
        || !(entry.nextEligibleAt === null || typeof entry.nextEligibleAt === "string")
        || !(entry.bypassRequestedAt === null || typeof entry.bypassRequestedAt === "string")
        || !(entry.terminalAt === null || typeof entry.terminalAt === "string") || !(entry.failure === null || isHarnessJson(entry.failure))) {
        throw new Error(`invalid harness response for ${path}`);
      }
    }
    return body as unknown as HarnessQueueResponse;
  }

  function requireConfig(body: Record<string, HarnessJson>, path: string): HarnessRunConfigResponse {
    if (typeof body.revision !== "string" || !isRecord(body.fields)) throw new Error(`invalid harness response for ${path}`);
    for (const field of Object.values(body.fields)) {
      if (!isRecord(field) || !isHarnessJson(field.value) || typeof field.source !== "string" || typeof field.immutable !== "boolean"
        || !["launch", "mid-run", "global"].includes(String(field.mutationClass))) {
        throw new Error(`invalid harness response for ${path}`);
      }
    }
    return body as unknown as HarnessRunConfigResponse;
  }

  function redactConfig(config: HarnessRunConfigResponse): HarnessRunConfigResponse {
    const redactValue = (value: HarnessJson, depth = 0): HarnessJson => depth >= MAX_REDACTION_DEPTH
      ? "[REDACTED]"
      : Array.isArray(value)
      ? value.map((item) => redactValue(item, depth + 1))
      : value && typeof value === "object"
        ? Object.fromEntries(Object.entries(value).map(([key, nested]) => [key, credentialKey.test(key) ? { isSecret: true, redacted: true } : redactValue(nested, depth + 1)]))
        : value;
    return {
      revision: config.revision,
      fields: Object.fromEntries(Object.entries(config.fields).map(([key, field]) => [
        key,
        credentialKey.test(key)
          ? { source: field.source, immutable: field.immutable, mutationClass: field.mutationClass, isSecret: true, redacted: true }
          : { ...field, value: redactValue(field.value!) },
      ])),
    } as HarnessRunConfigResponse;
  }

  function requireSteer(body: Record<string, HarnessJson>, path: string): HarnessSteerResponse {
    if (typeof body.ok !== "boolean" || !(typeof body.id === "string" || body.id === null)
      || typeof body.queued !== "boolean" || typeof body.restart !== "boolean") {
      throw new Error(`invalid harness response for ${path}`);
    }
    return { ok: body.ok, id: body.id, queued: body.queued, restart: body.restart };
  }

  function requirePlan(body: Record<string, HarnessJson>, path: string): HarnessEffectivePlanResponse {
    if (typeof body.runId !== "string" || typeof body.revision !== "string" || typeof body.planHash !== "string"
      || !Array.isArray(body.tasks) || !(body.meta === null || isRecord(body.meta))) throw new Error(`invalid harness response for ${path}`);
    return body as unknown as HarnessEffectivePlanResponse;
  }

  function requireDecisions(body: Record<string, HarnessJson>, path: string): HarnessDecisionsResponse {
    if (!Array.isArray(body.decisions) || !isCapabilities(body.capabilities)) throw new Error(`invalid harness response for ${path}`);
    for (const decision of body.decisions) {
      if (!isRecord(decision) || typeof decision.id !== "string" || !Array.isArray(decision.options)
        || !(decision.task === null || typeof decision.task === "string") || !(decision.summary === null || typeof decision.summary === "string")
        || !(decision.requestedAt === null || typeof decision.requestedAt === "string")
        || (decision.status !== undefined && typeof decision.status !== "string")
        || (decision.category !== undefined && !isHarnessJson(decision.category))
        || (decision.needs !== undefined && !isHarnessJson(decision.needs))
        || (decision.why !== undefined && typeof decision.why !== "string")
        || (decision.blast_radius !== undefined && !isHarnessJson(decision.blast_radius))
        || (decision.choice !== undefined && !isHarnessJson(decision.choice))
        || (decision.answeredAt !== undefined && !isHarnessJson(decision.answeredAt))
        || decision.options.some((option) => typeof option !== "string" && (!isRecord(option) || typeof option.value !== "string"
          || (option.meaning !== undefined && typeof option.meaning !== "string")
          || (option.label !== undefined && typeof option.label !== "string")
          || (option.recommended !== undefined && typeof option.recommended !== "boolean")))) {
        throw new Error(`invalid harness response for ${path}`);
      }
    }
    return body as unknown as HarnessDecisionsResponse;
  }

  function requireAttempts(body: Record<string, HarnessJson>, path: string): HarnessAttemptsResponse {
    if (!Array.isArray(body.attempts)) throw new Error(`invalid harness response for ${path}`);
    for (const attempt of body.attempts) {
      if (!isRecord(attempt) || typeof attempt.attemptId !== "string" || typeof attempt.timeoutSecs !== "number" || !Number.isFinite(attempt.timeoutSecs)
        || (attempt.task !== undefined && typeof attempt.task !== "string")
        || (attempt.liveness !== null && (!isRecord(attempt.liveness) || typeof attempt.liveness.elapsedSecs !== "number" || !Number.isFinite(attempt.liveness.elapsedSecs)
          || typeof attempt.liveness.lastActivity !== "string" || typeof attempt.liveness.lastSignalAt !== "string"))) {
        throw new Error(`invalid harness response for ${path}`);
      }
    }
    return body as unknown as HarnessAttemptsResponse;
  }

  function requireRunDetail(body: Record<string, HarnessJson>, path: string): HarnessRunDetailResponse {
    if (!isRecord(body.run)) throw new Error(`invalid harness response for ${path}`);
    const ratelimits = body.run.ratelimits;
    if (ratelimits !== undefined) {
      if (!isRecord(ratelimits)) throw new Error(`invalid harness response for ${path}`);
      for (const limit of Object.values(ratelimits)) {
        if (!isRecord(limit) || typeof limit.provider !== "string" || typeof limit.state !== "string"
          || (limit.resumeAt !== undefined && limit.resumeAt !== null && typeof limit.resumeAt !== "string")
          || (limit.waitMs !== undefined && limit.waitMs !== null && (typeof limit.waitMs !== "number" || !Number.isFinite(limit.waitMs)))
          || !Object.values(limit).every((value) => value === undefined || isHarnessJson(value))) {
          throw new Error(`invalid harness response for ${path}`);
        }
      }
    }
    return body as unknown as HarnessRunDetailResponse;
  }

  async function answerDecision(
    runId: string,
    decisionId: string,
    choice: string,
  ): Promise<HarnessDecisionAnswerResponse> {
    return request<HarnessDecisionAnswerResponse>(
      `/runs/${encodeURIComponent(runId)}/decisions/${encodeURIComponent(decisionId)}`,
      {
        method: "POST",
        headers: { "content-type": "application/json" },
        body: JSON.stringify({ choice }),
      },
      (body) => redact(body) as HarnessDecisionAnswerResponse,
    );
  }

  async function steerTask(
    runId: string,
    taskId: string,
    body: { text?: string; restart?: boolean },
  ): Promise<HarnessSteerResponse> {
    return request<HarnessSteerResponse>(
      `/runs/${encodeURIComponent(runId)}/tasks/${encodeURIComponent(taskId)}/steer`,
      {
        method: "POST",
        headers: { "content-type": "application/json" },
        body: JSON.stringify(body),
      },
      (response) => requireSteer(response, `/runs/${encodeURIComponent(runId)}/tasks/${encodeURIComponent(taskId)}/steer`),
    );
  }

  async function getRunEvents(runId: string, query: { since?: string; taskId?: string; raw?: string; lastEventId?: string } = {}): Promise<HarnessEventsResponse> {
    const params = new URLSearchParams();
    if (query.since !== undefined) params.set("since", query.since);
    if (query.taskId !== undefined) params.set("task", query.taskId);
    const search = query.raw !== undefined ? query.raw : params.size ? `?${params}` : "";
    const path = `/runs/${encodeURIComponent(runId)}/events${search}`;
    return request(path, { headers: query.lastEventId ? { "last-event-id": query.lastEventId } : undefined }, (body) => redact(requireEvents(body, path) as unknown as HarnessJson) as unknown as HarnessEventsResponse);
  }

  async function openTaskStream(runId: string, taskId: string, query: { since?: string; raw?: string; lastEventId?: string; signal?: AbortSignal } = {}): Promise<Response> {
    const params = new URLSearchParams();
    if (query.since !== undefined) params.set("since", query.since);
    const search = query.raw !== undefined ? query.raw : params.size ? `?${params}` : "";
    const path = `/runs/${encodeURIComponent(runId)}/tasks/${encodeURIComponent(taskId)}/stream${search}`;
    return stream(path, { signal: query.signal, headers: query.lastEventId ? { "last-event-id": query.lastEventId } : undefined });
  }

  async function getRunConfig(runId: string): Promise<HarnessRunConfigResponse> {
    const path = `/runs/${encodeURIComponent(runId)}/config`;
    return request(path, undefined, (body) => redactConfig(requireConfig(body, path)));
  }

  async function getEffectivePlan(runId: string): Promise<HarnessEffectivePlanResponse> {
    const path = `/runs/${encodeURIComponent(runId)}/plan`;
    return request(path, undefined, (body) => redact(requirePlan(body, path) as unknown as HarnessJson) as unknown as HarnessEffectivePlanResponse);
  }

  async function getDecisions(runId: string): Promise<HarnessDecisionsResponse> {
    const path = `/runs/${encodeURIComponent(runId)}/decisions`;
    return request(path, undefined, (body) => redact(requireDecisions(body, path) as unknown as HarnessJson) as unknown as HarnessDecisionsResponse);
  }

  async function getRunDetail(runId: string): Promise<HarnessRunDetailResponse> {
    const path = `/runs/${encodeURIComponent(runId)}`;
    return request(path, undefined, (body) => redact(requireRunDetail(body, path) as unknown as HarnessJson) as unknown as HarnessRunDetailResponse);
  }

  async function controlTask(runId: string, taskId: string, verb: "kill" | "pause" | "resume", body: HarnessTaskControlRequest): Promise<Record<string, HarnessJson>> {
    return request(`/runs/${encodeURIComponent(runId)}/tasks/${encodeURIComponent(taskId)}/${verb}`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) }, (response) => redact(response) as Record<string, HarnessJson>);
  }

  async function controlRun(runId: string, verb: "kill" | "pause" | "resume", body: HarnessRunControlRequest = {}): Promise<Record<string, HarnessJson>> {
    return request(`/runs/${encodeURIComponent(runId)}/${verb}`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) }, (response) => redact(response) as Record<string, HarnessJson>);
  }

  async function patchRunConfig(runId: string, patch: Record<string, HarnessJson>, revision: string): Promise<HarnessRunConfigResponse> {
    const path = `/runs/${encodeURIComponent(runId)}/config`;
    return request(path, { method: "PATCH", headers: { "content-type": "application/json", "if-match": revision }, body: JSON.stringify(patch) }, (body) => redactConfig(requireConfig(body, path)));
  }

  async function poll(): Promise<AdapterResult> {
    const nowMs = now();
    const nowIso = new Date(nowMs).toISOString();
    if (!fixedConnection) connection = null;
    const pollConnection = await healthyConnection();

    const runsResponse = await requestVia<HarnessRunsResponse>(pollConnection, "/runs");
    const runs = runsResponse.runs;
    const queueResponse = await requestVia<HarnessQueueResponse>(pollConnection, "/queue", undefined, (body) => requireQueue(body, "/queue"));
    const items: Item[] = [];
    const plansPanel = buildPlansPanel(runs, queueResponse.queue, nowIso);
    const panels: Panel[] = [plansPanel];

    for (const run of runs) {
      if (HALT_STATUSES.has(run.status)) {
        items.push(buildHaltItem(id, run));
      }
    }

    type EnrichmentResult =
      | { kind: "timeline"; run: HarnessRunSummary; response: HarnessTimelineResponse | null }
      | { kind: "attempts"; run: HarnessRunSummary; response: HarnessAttemptsResponse | "unavailable" }
      | { kind: "events"; run: HarnessRunSummary; response: HarnessEventsResponse | "unavailable" };
    const jobs = runs.flatMap((run) => [
      { kind: "timeline" as const, run },
      ...(isActiveHarnessRun(run)
        ? [{ kind: "attempts" as const, run }, { kind: "events" as const, run }]
        : []),
    ]);
    const livenessByRun = new Map<string, {
      newestJournalMs: number | null;
      attempts: HarnessAttemptEntry[] | null;
    }>();
    const enrichments = await mapWithConcurrency(jobs, requestConcurrency, async (job): Promise<EnrichmentResult> => {
      if (job.kind === "attempts") {
        try {
          const path = `/runs/${encodeURIComponent(job.run.runId)}/attempts`;
          const response = await requestVia<HarnessAttemptsResponse>(pollConnection, path, undefined, (body) => requireAttempts(body, path));
          return { ...job, response };
        } catch {
          return { ...job, response: "unavailable" };
        }
      }
      if (job.kind === "events") {
        try {
          const path = `/runs/${encodeURIComponent(job.run.runId)}/events`;
          const response = await requestVia<HarnessEventsResponse>(pollConnection, path, undefined, (body) => requireEvents(body, path));
          return { ...job, response };
        } catch {
          return { ...job, response: "unavailable" };
        }
      }

      try {
        const response = await requestVia<HarnessTimelineResponse>(pollConnection,
          `/runs/${encodeURIComponent(job.run.runId)}/timeline`,
        );
        return { ...job, response };
      } catch {
        return { ...job, response: null };
      }
    });

    for (const enrichment of enrichments) {
      if (enrichment.kind === "attempts" && enrichment.response === "unavailable") {
        items.push(buildAttemptsUnavailableItem(id, enrichment.run, nowMs));
        const bucket = livenessByRun.get(enrichment.run.runId) ?? { newestJournalMs: null, attempts: null };
        bucket.attempts = null;
        livenessByRun.set(enrichment.run.runId, bucket);
      } else if (enrichment.kind === "attempts" && enrichment.response !== "unavailable") {
        const bucket = livenessByRun.get(enrichment.run.runId) ?? { newestJournalMs: null, attempts: null };
        bucket.attempts = enrichment.response.attempts;
        livenessByRun.set(enrichment.run.runId, bucket);
        const panelRun = plansPanel.data.runs.find((run) => run.runId === enrichment.run.runId);
        for (const attempt of enrichment.response.attempts) {
          if (isAttemptAlarmed(attempt, nowMs)) {
            items.push(buildLivenessAlertItem(id, enrichment.run, attempt, nowMs));
            if (panelRun) {
              panelRun.alarmed = true;
              const task = panelRun.waves.flatMap((wave) => wave.tasks).find((node) => node.id === attempt.task);
              if (task) task.alarmed = true;
            }
          }
        }
      } else if (enrichment.kind === "events" && enrichment.response !== "unavailable") {
        const bucket = livenessByRun.get(enrichment.run.runId) ?? { newestJournalMs: null, attempts: null };
        bucket.newestJournalMs = newestJournalEventMs(enrichment.response.events);
        livenessByRun.set(enrichment.run.runId, bucket);
      } else if (enrichment.kind === "timeline" && enrichment.response) {
        panels.push(buildForensicsPanel(enrichment.run.runId, nowIso, enrichment.response));
      }
    }

    for (const run of runs) {
      if (!isActiveHarnessRun(run)) continue;
      const liveness = livenessByRun.get(run.runId);
      if (!liveness) continue;
      const stalenessMs = computeRunStalenessMs(
        nowMs,
        liveness.newestJournalMs,
        liveness.attempts ? newestAttemptHeartbeatMs(liveness.attempts) : null,
      );
      if (!isRunAlarmedByStaleness(stalenessMs)) continue;
      const panelRun = plansPanel.data.runs.find((candidate) => candidate.runId === run.runId);
      if (panelRun) markRunStalenessAlarm(panelRun);
    }

    return { items, panels };
  }

  return { id, interval, poll, answerDecision, steerTask, getRunEvents, openTaskStream, getRunConfig, getEffectivePlan, getDecisions, getRunDetail, controlTask, controlRun, patchRunConfig };
}