/**
 * Agent transport selection (ADR-061). Adapters call createAgentTransport()
 * where they used to build a LineProcess directly; it returns either:
 *
 *  - legacy pipes (`docker exec -i <cli>`, LineProcess) when
 *    UAI_DURABLE_SESSIONS=0, or
 *  - the durable path: an in-container runner owning the CLI, driven through
 *    inbox/outbox files on the workspace mount (DurableProcess).
 *
 * Durable mode makes host restarts transparent WITHOUT orchestrator
 * changes: the orchestrator's lazy respawn calls factory.create as always,
 * and this helper quietly ATTACHES to a still-live runner (fresh heartbeat +
 * a `running` host_agent_sessions row) at the persisted outbox offset
 * instead of spawning a new CLI — the agent's in-memory context survives.
 * Adapters that re-handshake per process (codex) pass allowAttach:false and
 * get a fresh runner each create — the stale predecessor is asked to stop
 * via its inbox first.
 *
 * The runner script is COPIED into each session dir (not bind-mounted):
 * every existing container already sees the workspace at its identical host
 * path, so durable sessions work for containers created before this
 * feature, and each spawn ships the runner version matching this host.
 */

import { copyFileSync, mkdirSync, statSync, promises as fsp } from "node:fs";
import { join } from "node:path";

import { and, eq } from "drizzle-orm";

import { getDb, schema } from "../db";
import { taskWorkspaceDir } from "../env";
import { DurableProcess, runnerScriptPath } from "./durable-proc";
import { LineProcess, dockerExecArgs, type ExitHandler, type LineHandler } from "./proc";

/** The shared surface adapters program against (LineProcess's shape). */
export interface LineTransport {
  onLine(handler: LineHandler): void;
  onExit(handler: ExitHandler): void;
  writeLine(value: unknown): void;
  readonly stderrTail: string;
  readonly isClosed: boolean;
  close(): Promise<void>;
}

export interface AgentTransportOptions {
  taskId: string;
  agentId: string;
  containerName: string;
  /** CLI + args exactly as they'd follow `docker exec -i <container>`. */
  cli: string;
  cliArgs: string[];
  /** Env names forwarded from the host's own env (`-e NAME`). */
  passEnv?: string[];
  /** Explicit per-exec env values (`-e NAME=value`). */
  explicitEnv?: Record<string, string>;
  /**
   * Whether a live runner from a previous host process may be attached to.
   * True for stateless adapters (claude); false for ones whose host-side
   * protocol state can't outlive the host process (codex handshake).
   */
  allowAttach: boolean;
  kind: string;
  debugLabel?: string;
}

/** Runner heartbeat is 5 s; older than this = not attachable. */
const ATTACH_HEARTBEAT_FRESH_MS = 20_000;

function durableEnabled(): boolean {
  return process.env.UAI_DURABLE_SESSIONS !== "0";
}

export function createAgentTransport(opts: AgentTransportOptions): LineTransport {
  if (!durableEnabled()) {
    const { command, args } = dockerExecArgs(
      opts.containerName,
      opts.cli,
      opts.cliArgs,
      opts.passEnv ?? [],
      opts.explicitEnv ?? {},
    );
    return new LineProcess({ command, args, debugLabel: opts.debugLabel });
  }

  const db = getDb();
  const row = db
    .select()
    .from(schema.hostAgentSessions)
    .where(
      and(
        eq(schema.hostAgentSessions.taskId, opts.taskId),
        eq(schema.hostAgentSessions.agentId, opts.agentId),
      ),
    )
    .get();

  const persistOffset = (offset: number): void => {
    db.update(schema.hostAgentSessions)
      .set({ outboxOffset: offset, updatedAt: Date.now() })
      .where(
        and(
          eq(schema.hostAgentSessions.taskId, opts.taskId),
          eq(schema.hostAgentSessions.agentId, opts.agentId),
        ),
      )
      .run();
  };
  const markClosed = (): void => {
    db.update(schema.hostAgentSessions)
      .set({ status: "closed", updatedAt: Date.now() })
      .where(
        and(
          eq(schema.hostAgentSessions.taskId, opts.taskId),
          eq(schema.hostAgentSessions.agentId, opts.agentId),
        ),
      )
      .run();
  };

  // ---- Attach: a previous host process left this agent's runner alive. ----
  if (opts.allowAttach && row && row.status === "running" && heartbeatFresh(row.sessionDir)) {
    const proc = new DurableProcess({
      hostSessionDir: row.sessionDir,
      initialOutboxOffset: row.outboxOffset,
      onOffsetAdvance: persistOffset,
      onCloseRequested: markClosed,
      debugLabel: opts.debugLabel,
    });
    proc.onExit(markClosed);
    return proc;
  }

  // ---- Spawn a fresh runner, asking any predecessor to stop. --------------
  // Unconditional (not gated on row.status): a runner falsely marked closed
  // by a stale-heartbeat verdict may still be alive, and a stop appended to
  // a dead session's inbox is harmless — this is what guarantees one live
  // CLI per (task, agent).
  if (row) {
    void fsp
      .appendFile(join(row.sessionDir, "inbox.jsonl"), '{"__uai":"stop"}\n')
      .catch(() => {
        // Dir already gone / runner already dead — nothing to stop.
      });
  }

  const sessionDir = join(
    taskWorkspaceDir(opts.taskId),
    ".uai",
    "sessions",
    opts.agentId,
    String(Date.now()),
  );
  mkdirSync(sessionDir, { recursive: true });
  // Ship this host's runner with the session: works in containers created
  // before this feature (the workspace is mounted at its host path) and
  // never skews against the host version.
  const runnerInSession = join(sessionDir, "runner.mjs");
  try {
    copyFileSync(runnerScriptPath(), runnerInSession);
  } catch (err) {
    // Runner asset missing (a packaging miss — the 0.3.0/0.3.1 desktop
    // builds vendored the agent without runner/). Degrade to legacy pipes:
    // sessions work, they just don't survive host restarts.
    console.warn(
      `[transport] runner unavailable (${err instanceof Error ? err.message : err}) — falling back to direct pipes for ${opts.agentId}`,
    );
    const { command, args } = dockerExecArgs(
      opts.containerName,
      opts.cli,
      opts.cliArgs,
      opts.passEnv ?? [],
      opts.explicitEnv ?? {},
    );
    return new LineProcess({ command, args, debugLabel: opts.debugLabel });
  }

  const envArgs: string[] = [];
  for (const name of opts.passEnv ?? []) {
    if (process.env[name]) envArgs.push("-e", name);
  }
  for (const [name, value] of Object.entries(opts.explicitEnv ?? {})) {
    envArgs.push("-e", `${name}=${value}`);
  }

  const proc = new DurableProcess({
    hostSessionDir: sessionDir,
    onOffsetAdvance: persistOffset,
    onCloseRequested: markClosed,
    debugLabel: opts.debugLabel,
    spawnCommand: {
      command: "docker",
      args: [
        "exec",
        "-d",
        ...envArgs,
        opts.containerName,
        "node",
        runnerInSession, // identical path in-container (workspace self-mount)
        sessionDir,
        "--",
        opts.cli,
        ...opts.cliArgs,
      ],
    },
  });

  const now = Date.now();
  db.insert(schema.hostAgentSessions)
    .values({
      taskId: opts.taskId,
      agentId: opts.agentId,
      sessionDir,
      containerName: opts.containerName,
      kind: opts.kind,
      outboxOffset: 0,
      status: "running",
      createdAt: now,
      updatedAt: now,
    })
    .onConflictDoUpdate({
      target: [schema.hostAgentSessions.taskId, schema.hostAgentSessions.agentId],
      set: {
        sessionDir,
        containerName: opts.containerName,
        kind: opts.kind,
        outboxOffset: 0,
        status: "running",
        updatedAt: now,
      },
    })
    .run();

  proc.onExit(markClosed);
  return proc;
}

function heartbeatFresh(sessionDir: string): boolean {
  try {
    return (
      Date.now() - statSync(join(sessionDir, "heartbeat")).mtimeMs <
      ATTACH_HEARTBEAT_FRESH_MS
    );
  } catch {
    return false;
  }
}