#!/usr/bin/env bun
import { errMessage } from "agent-relay-sdk";
import { fireAndForget, guardedInterval, installUnhandledRejectionGuard } from "./async-guard";
import { loadConfig, initConfigFile } from "./config";
import { createRelayClient } from "./relay";
import type { ManagedSessionExitDiagnostics } from "./relay";
import { createRegistrationDriver } from "./registration";
import { createControlHandler } from "./control";
import { diagnoseSessionExit, hydrateTerminalGuests, managedSessionLivenessDetailed, reapTerminalGuests, refreshManagedAgentReport } from "./spawn";
import { startApiServer } from "./api";
import { recoverManagedAgents } from "./recovery";
import { ProviderProbeCache } from "./provider-probe";
import { sweepEmptyWorkspaceContainers, workspacesRoot } from "./workspace-probe";
import { startOrchestratorMaintenanceScheduler } from "./maintenance";
import { OrchestratorQuotaPoller, resolveLiveCodexResetConsumeTransport, type CodexResetFiredEvent } from "./quota-poller";
import { SharedCallmuxSupervisor } from "./shared-callmux";
import { createCommandPoller } from "./command-poller";
import { defaultLocalHealthProbe, localReadinessUrl, resolveRuntimePath, selfHealLingeringUpgradeGuard } from "./self-upgrade";
import { detectSelfSupervision } from "./self-supervision";

const args = process.argv.slice(2);

if (args[0] === "init") {
  const path = initConfigFile({});
  console.error(`[orchestrator] Config initialized at ${path}`);
  console.error("[orchestrator] Edit it to set your relay URL, base directory, and secrets.");
  process.exit(0);
}

if (args[0] === "--version" || args[0] === "-v") {
  const { VERSION } = await import("./version");
  console.log(VERSION);
  process.exit(0);
}

if (args[0] === "--help" || args[0] === "-h") {
  console.log(`agent-relay-orchestrator — manage agent lifecycle across hosts

Usage:
  agent-relay-orchestrator              Start the orchestrator daemon
  agent-relay-orchestrator init         Create default config file
  agent-relay-orchestrator --version    Print version and exit

Environment:
  AGENT_RELAY_URL                   Relay server URL (default: http://localhost:4850)
  AGENT_RELAY_TOKEN                 Authentication token
  AGENT_RELAY_ORCHESTRATOR_ID       Orchestrator ID (default: hostname)
  AGENT_RELAY_ORCHESTRATOR_BASE_DIR Base directory for agent CWDs
  AGENT_RELAY_ORCHESTRATOR_API_PORT API server port (default: 4860)
  AGENT_RELAY_ORCHESTRATOR_CONFIG   Path to config file

Config file: ~/.agent-relay/orchestrator.json
`);
  process.exit(0);
}

// #1676 — last-resort backstop, installed before anything can schedule work. An unhandled
// rejection in the orchestrator is NOT fatal: every realistic source is a recoverable I/O
// transient against the relay or a peer host, and the process that supervises other hosts'
// agents makes the fleet less recoverable by dying. `uncaughtException` deliberately stays
// at the runtime default (fatal) — see the SDK function for the full reasoning. If this
// handler ever fires, a guarded site was missed; it logs at the volume a crash would have.
installUnhandledRejectionGuard();

const config = loadConfig();
const probeCache = new ProviderProbeCache(config);
const relay = createRelayClient(config, probeCache);
const control = createControlHandler(config, relay);
// #1478 — on a relay RECONNECT (e.g. after a relay restart), re-report the live managed-agent roster
// so the relay's orchestrator↔agent control association is rebuilt for still-running agents. Without
// this, register() re-establishes messaging but never re-sends the roster, and the relay never
// reconstructs it — so relay_shutdown_agent stays -32004 and the dashboard web-terminal button stays
// missing for reconnecting agents until the next spawn/exit change happens to trigger a re-report.
relay.setOnReconnected(() => relay.updateManagedAgents(control.getManagedAgents()));
// #1425 finding 1 — the live banked-reset consume transport is wired ONLY for a genuine production
// run (explicit AGENT_RELAY_CODEX_RESET_CONSUME_LIVE=1 opt-in) and NEVER for the dev-service profile
// (AGENT_RELAY_DEV_PROFILE=1) / a test / CI. In every other case the resolver returns undefined, so
// the poller's class-default THROWING stub stands and the destructive RPC is UNREACHABLE by
// construction — a dev/test/CI invocation of the manual consume errors instead of burning a scarce
// reset. The env is the merged process + orchestrator-config env (dev sets its flag in config.env).
// Auto-fire also gets the onResetFired hook here (finding 7) so an autonomous fire emits the SAME
// `provider-quota.reset-fired` relay event as the manual path — the count refresh already flows via
// reportProviderQuota inside the poller. Best-effort + auto-only: the manual path emits its own event
// relay-side, so we never double-emit for manual.
const liveCodexResetConsume = resolveLiveCodexResetConsumeTransport({ ...process.env, ...config.env });
const quotaPoller = new OrchestratorQuotaPoller(config, relay, {
  ...(liveCodexResetConsume ? { codexResetCreditConsume: liveCodexResetConsume } : {}),
  onResetFired: (event: CodexResetFiredEvent) => {
    if (event.trigger !== "auto") return;
    void fireAndForget("codex reset-fired event report", relay.reportCodexResetFired(event));
  },
});
const sharedCallmux = new SharedCallmuxSupervisor(config);

// #1676 — registration retry: exponential backoff + a surfaced degraded state, replacing
// the fixed 5s `for(;;)` spin that was the second half of the crash loop.
const registration = createRegistrationDriver({ register: () => relay.register() });

const POLL_INTERVAL_MS = 3_000;
const GUEST_REAP_INTERVAL_MS = 60_000;
let commandPoller: ReturnType<typeof createCommandPoller> | null = null;
// #1760 — health reported before the poll loop is up (or after shutdown): an idle, non-stalled loop.
const nullCommandLoopHealth = { getHealth: () => ({ inFlight: false, phase: "idle" as const, activeForMs: 0, stalled: false }) };
let healthCheckTimer: Timer | null = null;
let guestReaperTimer: Timer | null = null;
let apiServer: { stop(): void; url: string } | null = null;

async function startup(): Promise<void> {
  console.error(`[orchestrator] Starting orchestrator: ${config.id}`);
  console.error(`[orchestrator]   relay: ${config.relayUrl}`);
  console.error(`[orchestrator]   baseDir: ${config.baseDir}`);
  console.error(`[orchestrator]   providers: ${config.providers.join(", ")}`);
  console.error(`[orchestrator]   env keys: ${Object.keys(config.env).length}`);

  // Start API server before registration so we can advertise the URL
  // #1760 — the poller is created later in startPolling(); pass a lazy accessor so /api/health reads
  // whichever poller is live at request time (null before startPolling, giving no commandLoop field).
  apiServer = startApiServer(config, probeCache, relay, quotaPoller, registration, {
    getHealth: () => (commandPoller ?? nullCommandLoopHealth).getHealth(),
  });
  console.error(`[orchestrator]   apiUrl: ${apiServer.url}`);

  relay.setApiUrl(apiServer.url);

  // #1509 r7 §13 — COSMETIC self-heal against a lingering OWN-unit self-upgrade restart guard,
  // kicked off BEFORE registration and NOT awaited. Under the r7 one-shot model (RunAtLoad,
  // KeepAlive=false) a guard that fails to boot itself out sits loaded-but-idle at PID `-`: it can
  // NEITHER loop (KeepAlive=false ⇒ never relaunches) NOR wedge a later dispatch (the dispatch
  // precondition blocks only on a RUNNING guard, and this reap is not on the critical path), so
  // this once-at-startup unawaited call can no longer leave a wedge — it merely tidies the launchd
  // listing. It also cleans up any PRE-r7 KeepAlive guard left over across the upgrade transition;
  // for those it gates on a LOCAL /api/health signal (Relay-independent) and reaps only when proven
  // healthy. It removes each stale generation by its exact unique `.s<seq>.<uuid>` label; a NEW
  // upgrade's guard is RUNNING under a different uuid-unique label, so it is never mistaken for a
  // stale one nor removed by a concurrent reap. Scoped to `${selfUnit}.upgrade-guard`.
  const selfHealSupervision = detectSelfSupervision();
  void fireAndForget("Upgrade-guard self-heal", selfHealLingeringUpgradeGuard(selfHealSupervision.selfUnit, {
    probe: defaultLocalHealthProbe(localReadinessUrl(config.apiPort)),
    runtimePath: resolveRuntimePath(selfHealSupervision.runtimePrefix),
  }).then((reapedGuards) => {
    if (reapedGuards.length > 0) {
      console.error(`[orchestrator] Reaped ${reapedGuards.length} lingering own-unit self-upgrade guard job(s): ${reapedGuards.join(", ")}`);
    }
  }));

  // Register with relay. The server and orchestrator are often restarted
  // together, so startup must tolerate the server not listening yet.
  await registration.registerUntilConnected();
  relay.startHeartbeatLoop();

  // Recover existing tmux sessions
  await recoverManagedAgents(config, control, relay);

  // Host-local maintenance must run where the agent processes and tmux sockets live.
  startOrchestratorMaintenanceScheduler(config);
  quotaPoller.start();
  try {
    sharedCallmux.start();
  } catch (err) {
    console.error(`[orchestrator] Shared callmux supervisor failed to start: ${errMessage(err)}`);
  }

  // Sweep empty workspace container dirs left behind by prior cleanups (#280).
  const swept = sweepEmptyWorkspaceContainers(workspacesRoot(config.baseDir));
  if (swept.length > 0) console.error(`[orchestrator] Swept ${swept.length} empty workspace container(s)`);

  // Restore guest-terminal TTLs persisted before the last restart, then reap any
  // that expired (or were orphaned) while the orchestrator was down (#144).
  hydrateTerminalGuests();
  const reaped = reapTerminalGuests(config);
  if (reaped.length > 0) console.error(`[orchestrator] Reaped ${reaped.length} expired guest terminal(s)`);

  // Start polling for command requests
  startPolling();

  // Periodic health check — remove dead sessions.
  // #1676 — this was `setInterval(healthCheck, 60_000)`, and it is the most likely site
  // of the incident's UNCAUGHT AbortError: healthCheck awaits relay.updateManagedAgents(),
  // whose fetch is timeout-aborted, and an async timer callback's rejection has no handler
  // — in Bun that exits the process. guardedInterval consumes it and logs.
  healthCheckTimer = guardedInterval("Managed-agent health check", 60_000, healthCheck);

  // Periodic guest-terminal reaper — enforces guest TTL without requiring a new
  // guest creation to trigger cleanup (#144).
  guestReaperTimer = guardedInterval("Guest terminal reap", GUEST_REAP_INTERVAL_MS, () => {
    reapTerminalGuests(config);
  });

  console.error("[orchestrator] Ready. Polling for command requests...");
}

function startPolling(): void {
  commandPoller = createCommandPoller({ relay, control, intervalMs: POLL_INTERVAL_MS, errorBackoffMs: POLL_INTERVAL_MS });
  commandPoller.start();
}

async function healthCheck(): Promise<void> {
  const agents = control.getManagedAgents();
  let changed = false;
  const exitedAgents: ManagedSessionExitDiagnostics[] = [];
  let remainingAgents = agents;
  for (const agent of agents) {
    const refreshed = refreshManagedAgentReport(agent);
    if (JSON.stringify(refreshed) !== JSON.stringify(agent)) {
      remainingAgents = remainingAgents.map((item) => item.agentId === agent.agentId ? refreshed : item);
      control.setManagedAgents(remainingAgents);
      changed = true;
    }
    const sessionName = refreshed.sessionName ?? refreshed.tmuxSession;
    // Capture the systemd diagnostics read at the exact moment liveness is first
    // observed dead, and reuse it below instead of querying again — a `--collect`ed
    // transient unit can be garbage-collected between two separate reads, which is
    // what let #1317 launder a real status=1 death into a reported "success exit=0/0".
    const { liveness, systemd: capturedSystemd } = managedSessionLivenessDetailed(sessionName);
    if (liveness === "unknown") {
      console.error(`[orchestrator] Session liveness unknown: ${sessionName}; preserving and retrying next health check`);
      continue;
    }
    if (liveness === "dead") {
      // #1746 fwd — resolve THIS specific dead agent by its UNIQUE keys only. Since #1746 fwd made
      // selectSessionRecord "try every key" (a MISS falls through to the next, for the kill-switch),
      // passing policyName here would let a full-key-set miss (this agent's own record already removed)
      // fall through to the policyName branch and misattribute a live sibling policy session's exit —
      // flipping the sibling offline while it's still running and dropping this agent with no exit record.
      const diagnostics = diagnoseSessionExit({
        agentId: refreshed.agentId,
        spawnRequestId: refreshed.spawnRequestId,
        tmuxSession: sessionName,
        systemdOverride: capturedSystemd,
      }) ?? {
        agentId: refreshed.agentId,
        provider: refreshed.provider,
        sessionName,
        tmuxSession: refreshed.tmuxSession,
        cwd: refreshed.cwd,
        label: refreshed.label,
        policyName: refreshed.policyName,
        spawnRequestId: refreshed.spawnRequestId,
        supervisor: refreshed.supervisor ?? "unknown",
        systemdUnit: refreshed.systemdUnit,
        terminalSession: refreshed.terminalSession,
        terminalAvailable: refreshed.terminalAvailable,
        pid: refreshed.pid,
        startedAt: refreshed.startedAt,
        detectedAt: Date.now(),
        runtimeMs: Math.max(0, Date.now() - refreshed.startedAt),
        runnerInfoPresent: false,
        unavailable: ["orchestrator session record unavailable", "stdout/stderr log unavailable"],
        lastError: `managed ${refreshed.provider} session ${sessionName} disappeared; orchestrator session record unavailable`,
      };
      exitedAgents.push(diagnostics);
      console.error(`[orchestrator] Session dead: ${sessionName} — ${diagnostics.lastError}`);
      if (diagnostics.logFile) {
        console.error(`[orchestrator]   log: ${diagnostics.logFile} (${diagnostics.logBytes ?? "unknown"} bytes${diagnostics.logEmpty ? ", empty" : ""})`);
      }
      if (diagnostics.logTail?.length) {
        console.error(`[orchestrator]   last log line: ${diagnostics.logTail[diagnostics.logTail.length - 1]}`);
      }
      remainingAgents = remainingAgents.filter((a) => (a.sessionName ?? a.tmuxSession) !== sessionName);
      control.setManagedAgents(remainingAgents);
      changed = true;
    }
  }
  if (changed) {
    await relay.updateManagedAgents(control.getManagedAgents(), exitedAgents);
  }
}

async function shutdown(): Promise<void> {
  console.error("[orchestrator] Shutting down...");
  commandPoller?.stop();
  if (healthCheckTimer) clearInterval(healthCheckTimer);
  if (guestReaperTimer) clearInterval(guestReaperTimer);
  if (apiServer) apiServer.stop();
  quotaPoller.stop();
  sharedCallmux.stop();
  relay.stopHeartbeatLoop();
  process.exit(0);
}

process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);

startup().catch((err) => {
  console.error(`[orchestrator] Fatal: ${err}`);
  process.exit(1);
});