import type { OrchestratorConfig } from "./config";
import { describeFailure, guardedInterval, isAbortError } from "./async-guard";
import type { ProviderProbeCache } from "./provider-probe";
import { PROCESS_ARTIFACT_SHA, detectSelfSupervision } from "./self-supervision";
import { GIT_SHA, ORCHESTRATOR_PROTOCOL_VERSION, VERSION, runtimeMetadata } from "./version";
import type { AgentLifecycle, ProviderQuotaConfigMap, ProviderQuotaLeaseAcquireInput, ProviderQuotaLeaseAcquireResult, ProviderQuotaUpdateInput, QueuedSpawn, WorkspaceMetadata, WorkspaceMode, ManagedSessionExitDiagnostics as SdkManagedSessionExitDiagnostics, SpawnProvider } from "agent-relay-sdk";
import { ReconnectionManager, RelayHttpClient } from "agent-relay-sdk";

// #1452 round-13 — result of an atomic command claim. `token` is the per-execution fence token minted
// for a review publish (the sole credential authorizing its later terminal settlement); absent otherwise.
export interface ClaimCommandResult {
  claimed: boolean;
  token?: string;
}

export interface RelayClient {
  register(): Promise<void>;
  heartbeat(): Promise<void>;
  // #1478 — returns the roster PATCH's ok status so the reconnect resync (see setOnReconnected) can
  // retry on a non-2xx instead of treating a 401/500 as a completed re-report. `void` is tolerated for
  // callers/mocks that don't report status (treated as success).
  // #1513 — `queuedSpawns` (optional) reports spawns held back by the host headroom admission gate so
  // the relay/dashboard can show a "queued: waiting for host memory" row. When provided it REPLACES the
  // relay's held queued-spawn snapshot (which every roster PATCH — ordinary or reconnect resync — then
  // includes); omitting it leaves the current snapshot untouched, so ordinary roster pushes don't drop a
  // still-queued spawn. Pass `[]` to clear it once a spawn is admitted or times out.
  updateManagedAgents(agents: ManagedAgentReport[], exitedAgents?: ManagedSessionExitDiagnostics[], queuedSpawns?: QueuedSpawn[]): Promise<boolean | void>;
  pollCommands(signal?: AbortSignal): Promise<RelayCommand[]>;
  getCommand(commandId: string): Promise<RelayCommand | null>;
  updateCommand(commandId: string, status: string, result?: Record<string, unknown>, error?: string): Promise<boolean>;
  // #1452 round-11 HIGH#1 — atomically claim a command (CAS pending→accepted). Returns true iff THIS
  // host won the claim; the caller MUST NOT execute an origin-mutating action (the review publish push)
  // when it returns false (the command was canceled by the reconciler, or already claimed).
  // #1452 round-13 — claiming a review PUBLISH returns a per-execution FENCE TOKEN: the ONLY credential
  // that authorizes the command's later terminal settlement (settleReviewPublish). `{claimed:false}` on a
  // lost claim (canceled/reaped). No token for non-publish claims.
  claimCommand(commandId: string, expectedType?: string): Promise<ClaimCommandResult>;
  // #1452 round-12 HIGH#2 — re-fence a claimed publish command IMMEDIATELY before its origin push (a CAS
  // accepted/running → running that refreshes the lease). Returns true iff still live; false means it was
  // swept to timed_out (or terminalized) after the claim, so the host MUST skip the push (fail closed).
  fencePublishPush(commandId: string, expectedType?: string): Promise<boolean>;
  // #1452 round-13 HIGH#2 — settle a claimed review publish via the token-gated route. The fence token
  // proves executor ownership, so this is the ONLY external path allowed to terminalize the command
  // (generic PATCH + WS bus are refused). Returns false on any transport/auth failure.
  settleReviewPublish(commandId: string, token: string, status: "succeeded" | "failed", result?: Record<string, unknown>, error?: string): Promise<boolean>;
  acquireProviderQuotaLease(orchestratorId: string, input: ProviderQuotaLeaseAcquireInput): Promise<ProviderQuotaLeaseAcquireResult>;
  releaseProviderQuotaLease(orchestratorId: string, input: ProviderQuotaLeaseAcquireInput & { leaseToken: string }): Promise<{ released: boolean }>;
  reportProviderQuota(input: ProviderQuotaUpdateInput): Promise<unknown>;
  // #1425 finding 7 — report an autonomous auto-fire so the relay emits the same
  // `provider-quota.reset-fired` observability event the manual path emits.
  reportCodexResetFired(event: CodexResetFiredEventReport): Promise<unknown>;
  getProviderQuotaConfig(): Promise<ProviderQuotaConfigMap>;
  setApiUrl(url: string): void;
  // #1478 — register a handler fired on every relay RECONNECT (not the first connect). A relay
  // restart lets the orchestrator's messaging channel reconnect while the relay's managed-agent
  // control association is NOT rebuilt (register() re-sends registration only, never the roster, and
  // the relay never reconstructs it). The handler re-reports the live roster so shutdown routing and
  // the dashboard terminal button recover for still-running agents without waiting for a spawn/exit.
  setOnReconnected(handler: () => void | boolean | Promise<void | boolean>): void;
  startHeartbeatLoop(): void;
  stopHeartbeatLoop(): void;
  remintRunnerToken(currentToken: string): Promise<RunnerTokenRemint | null>;
  // #1201 F2 — re-evaluate the self-restart safe-point against CURRENT relay state, called
  // immediately before killing the predecessor in a self-restart teardown. Fail-closed: any
  // transport/parse failure resolves to a violation so the caller aborts the kill.
  recheckSelfRestartSafePoint(agentId: string): Promise<{ ok: boolean; reason?: string }>;
  // #1676 — observable heartbeat health. A heartbeat that times out marks the peer
  // DEGRADED and schedules a retry; it never terminates the process. Surfaced on the
  // orchestrator's own /api/health so a degraded peer is visible without log archaeology.
  getHealth(): RelayHealth;
  connected: boolean;
}

// #1676 — heartbeat health as seen from this orchestrator. `connected` is what we believe
// about our registration; `degraded` is the narrower "the last heartbeat did not land",
// which trips on the FIRST failure (well before `connected` drops) so a flaky peer is
// visible immediately rather than only once we give up on it.
export interface RelayHealth {
  connected: boolean;
  degraded: boolean;
  consecutiveHeartbeatFailures: number;
  lastHeartbeatError?: string;
  lastHeartbeatOkAt?: number;
  degradedSince?: number;
}

export class CommandUpdateError extends Error {
  constructor(
    readonly commandId: string,
    readonly status: string,
    readonly responseStatus: number,
    readonly responseBody: string,
  ) {
    super(`relay command update ${commandId} → ${status} failed: HTTP ${responseStatus}${responseBody ? `: ${responseBody}` : ""}`);
  }
}

const COMMAND_UPDATE_MAX_ATTEMPTS = 3;
const COMMAND_UPDATE_RETRY_DELAY_MS = 100;

function retryableCommandUpdateStatus(status: number): boolean {
  return status === 408 || status === 429 || status >= 500;
}

function commandUpdateRetryDelay(attempt: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, COMMAND_UPDATE_RETRY_DELAY_MS * attempt));
}

// #1425 finding 7 — the auto-fire observability payload the orchestrator reports to the relay.
export interface CodexResetFiredEventReport {
  provider: string;
  accountKey: string;
  trigger: "auto" | "manual";
  outcome: string;
  burned: boolean;
  idempotencyKey: string;
  availableCount?: number;
}

interface RunnerTokenRemint {
  token: string;
  record: { jti: string; profileId?: string; expiresAt?: number };
}

export interface ManagedAgentReport {
  agentId: string;
  provider: SpawnProvider;
  model?: string;
  effort?: string;
  profile?: string;
  workspaceMode?: WorkspaceMode;
  lifecycle?: AgentLifecycle;
  workspace?: WorkspaceMetadata;
  sessionName?: string;
  supervisor?: "process" | "systemd" | "launchd" | "unknown";
  systemdUnit?: string;
  terminalSession?: string;
  terminalAvailable?: boolean;
  tmuxSession: string;
  cwd: string;
  label?: string;
  approvalMode: string;
  policyName?: string;
  spawnRequestId?: string;
  automationRunId?: string;
  pid?: number;
  startedAt: number;
}

// Canonical shape lives in agent-relay-sdk — alias and re-export, never re-declare.
export type ManagedSessionExitDiagnostics = SdkManagedSessionExitDiagnostics;

export interface RelayCommand {
  id: string;
  type: string;
  target: string;
  params: Record<string, unknown>;
  status: string;
}

// Reconnect backoff: 30s → 1m → 2m → … capped at 1h, with jitter. Replaces the
// former fixed BACKOFF_SCHEDULE_MS staircase — see SDK ReconnectionManager.
const RECONNECT_INITIAL_MS = 30_000;
const RECONNECT_MAX_MS = 3_600_000; // 1 hour
const RECONNECT_JITTER_MS = 1_000;
const TRANSIENT_ABORT_RECONNECT_MS = 250;
const ABORT_FAILURES_BEFORE_DISCONNECT = 3;

export function buildRegistrationMeta(
  config: Pick<OrchestratorConfig, "tmuxPrefix">,
  runtime: ReturnType<typeof runtimeMetadata>,
  now = Date.now,
  pid = process.pid,
): Record<string, unknown> {
  const supervision = detectSelfSupervision();
  return {
    ...runtime,
    pid,
    tmuxPrefix: config.tmuxPrefix,
    startedAt: now(),
    version: VERSION,
    protocolVersion: ORCHESTRATOR_PROTOCOL_VERSION,
    gitSha: reportedGitSha(),
    supervisor: supervision.supervisor,
    ...(supervision.selfUnit ? { selfUnit: supervision.selfUnit } : {}),
    ...(supervision.runtimePrefix ? { runtimePrefix: supervision.runtimePrefix } : {}),
  };
}

// #1316: attest the PROCESS, not a marker file re-read through the mutable
// runtime symlink. PROCESS_ARTIFACT_SHA is frozen at startup from this process's
// own module ancestry, so a symlink swap without a restart cannot change it.
function reportedGitSha(): string | undefined {
  return PROCESS_ARTIFACT_SHA ?? GIT_SHA;
}

export function createRelayClient(config: OrchestratorConfig, probeCache: ProviderProbeCache): RelayClient {
  const agentId = `orchestrator-${config.id}`;
  let heartbeatTimer: Timer | null = null;
  let connected = false;
  // #1478 — reconnect / roster-resync bookkeeping. `everConnected` distinguishes the FIRST connect
  // (startup, where recoverManagedAgents already reports the roster) from a later reconnect.
  // `rosterResyncPending` means we owe the relay a full managed-agent re-report — a reconnect may have
  // reached a relay that lost the orchestrator↔agent association. It is retained until a write that
  // reflects the CURRENT roster actually lands (see patchRoster), so a partial reconnect (register OK
  // but cursor bootstrap or the PATCH failed) is retried on the next successful heartbeat, never lost.
  let everConnected = false;
  let rosterResyncPending = false;
  let rosterSyncInFlight = false;
  let onReconnected: (() => void | boolean | Promise<void | boolean>) | undefined;
  // #1478 (round-3, sol lost-update HIGH) — ALL roster PATCHes (ordinary updateManagedAgents from
  // spawn/mutation/health AND the reconnect resync) flow through the one serialized `patchRoster`
  // queue below, so two roster writes never overlap and a delayed stale write can never clobber a
  // newer one. `rosterGeneration` is a monotonic counter bumped per write REQUEST; a write clears the
  // resync obligation only if it is still the newest generation when it lands (compare-and-clear), so
  // the last write to land always reflects the newest roster and pending is never cleared against a
  // superseded snapshot.
  // #1478 (round-4, sol epoch HIGH) — `rosterResyncEpoch` advances every time a resync obligation is
  // marked (see markRosterResyncPending). The generation check alone is NOT enough: an ordinary
  // pre-restart PATCH still in flight when a reconnect marks pending can 2xx-complete while its
  // generation is STILL the newest (no newer write was requested yet), and would wrongly clear the NEW
  // obligation — leaving the restarted relay without the control association. Each write captures the
  // epoch at REQUEST time and clears the obligation only if that epoch still equals the current one, so
  // a write requested BEFORE the obligation can never satisfy the clear.
  let rosterGeneration = 0;
  let rosterResyncEpoch = 0;
  // #1513 — the current set of spawns held back by the host headroom admission gate. Held here (not
  // recomputed per write) so EVERY roster PATCH — ordinary updateManagedAgents AND the reconnect
  // resync — carries it, and a spawn that is still queued isn't dropped from the dashboard by an
  // unrelated roster push. Replaced only when a caller passes an explicit `queuedSpawns` array.
  let queuedSpawnsSnapshot: QueuedSpawn[] = [];
  let rosterWriteChain: Promise<unknown> = Promise.resolve();
  let abortFailures = 0;
  // #1676 — peer-degraded bookkeeping. Any failed heartbeat (abort or not) marks the peer
  // degraded until one lands; `connected` still flips only on the slower disconnect rule.
  let consecutiveHeartbeatFailures = 0;
  let degraded = false;
  let degradedSince: number | undefined;
  let lastHeartbeatError: string | undefined;
  let lastHeartbeatOkAt: number | undefined;
  const reconnectMgr = new ReconnectionManager({ initialMs: RECONNECT_INITIAL_MS, maxMs: RECONNECT_MAX_MS, jitterMs: RECONNECT_JITTER_MS });
  let cursorFloor = 0;
  let apiUrl: string | undefined;
  // Shared transport: auth-header injection + timeout + typed errors. The
  // orchestrator-control endpoints have no typed method, so route them through
  // the generic request() escape hatch. setToken() swaps in the runtime token
  // the relay mints at registration.
  const http = new RelayHttpClient({ baseUrl: config.relayUrl, token: config.token });

  async function apiCall(method: string, path: string, body?: unknown): Promise<Response> {
    return http.request(method, `/api${path}`, body);
  }

  async function register(): Promise<void> {
    const runtime = runtimeMetadata();
    const providerSnapshot = await probeCache.getSnapshot(true);
    const res = await apiCall("POST", "/orchestrators", {
      id: config.id,
      hostname: config.hostname,
      providers: providerSnapshot.providers,
      providerStatus: providerSnapshot.providerStatus,
      providerCatalog: providerSnapshot.providerCatalog,
      baseDir: config.baseDir,
      apiUrl,
      envKeys: Object.keys(config.env),
      package: runtime.package,
      contracts: runtime.contracts,
      capabilities: runtime.capabilities,
      version: VERSION,
      protocolVersion: ORCHESTRATOR_PROTOCOL_VERSION,
      gitSha: reportedGitSha(),
      meta: buildRegistrationMeta(config, runtime),
    });
    if (!res.ok) {
      const err = await res.text();
      throw new Error(`Failed to register orchestrator: ${res.status} ${err}`);
    }
    const registered = await res.json().catch(() => null) as { runtimeToken?: { token?: string } } | null;
    if (registered?.runtimeToken?.token) http.setToken(registered.runtimeToken.token);
    // #1478 — ANY re-registration after the first connect may have reached a relay that lost our
    // managed-agent association (a restart with a fresh/empty roster, or a timeout-masked restart that
    // never dropped `connected` — see the transient-abort path below). Mark a roster resync owed NOW,
    // BEFORE the cursor bootstrap: if that step throws, register() rejects and reconnect() swallows it
    // while `connected` stays true (no further reconnect fires), so the pending flag is the only thing
    // that keeps the replay alive — the next successful heartbeat flushes it. The first connect is
    // exempt: startup's recoverManagedAgents already reports the roster.
    if (everConnected) markRosterResyncPending();
    connected = true;
    everConnected = true;
    abortFailures = 0;
    reconnectMgr.reset();

    // Bootstrap message cursor
    const cursor = await apiCall("GET", "/messages/cursor");
    if (cursor.ok) {
      const data = await cursor.json() as { latestId: number };
      cursorFloor = data.latestId;
    }

    console.error(`[orchestrator] Registered as ${config.id} (agent: ${agentId})`);
    await flushRosterResync();
  }

  // #1478 (round-4, sol epoch HIGH) — mark a roster resync owed AND advance the resync epoch. The epoch
  // advances on EVERY obligation so a write requested BEFORE this obligation can never clear it (see the
  // rosterResyncEpoch declaration): patchRoster clears the obligation only for a write whose captured
  // epoch still equals the current one. The classic race this closes: an ordinary pre-restart PATCH is
  // in flight when a reconnect marks pending; the old PATCH 2xx-completes while its generation is still
  // the newest, and — without the epoch — would clear the NEW obligation, so the reconnect's resync
  // never fires and the restarted relay is left without the control association.
  function markRosterResyncPending(): void {
    rosterResyncPending = true;
    rosterResyncEpoch += 1;
  }

  // #1478 — trigger an owed roster resync. The handler re-reports the CURRENT roster through the same
  // serialized `patchRoster` queue as every other write, which does the compare-and-clear: a rejected /
  // timed-out / non-2xx / superseded write leaves `rosterResyncPending` set so the next successful
  // heartbeat retries. `rosterSyncInFlight` only avoids stacking duplicate resync triggers per beat;
  // it does NOT gate ordinary writes (those serialize on the queue). Never throws into heartbeat.
  async function flushRosterResync(): Promise<void> {
    if (!rosterResyncPending || !onReconnected || rosterSyncInFlight) return;
    rosterSyncInFlight = true;
    try {
      await onReconnected();
    } catch (err) {
      console.error(`[orchestrator] Managed-agent roster resync after reconnect failed (will retry): ${err}`);
    } finally {
      rosterSyncInFlight = false;
    }
  }

  // #1676 — a heartbeat that fails marks the peer degraded and keeps beating. Logged once
  // per degradation (not once per beat) so a long outage doesn't drown the log.
  function markPeerDegraded(err: unknown): void {
    consecutiveHeartbeatFailures += 1;
    lastHeartbeatError = describeFailure(err);
    if (!degraded) {
      degraded = true;
      degradedSince = Date.now();
      console.error(`[orchestrator] Relay peer degraded: heartbeat ${lastHeartbeatError}; retrying`);
    }
  }

  function markPeerHealthy(): void {
    consecutiveHeartbeatFailures = 0;
    lastHeartbeatOkAt = Date.now();
    lastHeartbeatError = undefined;
    if (degraded) {
      const forMs = degradedSince ? Date.now() - degradedSince : 0;
      console.error(`[orchestrator] Relay peer recovered after ${Math.round(forMs / 1000)}s degraded`);
      degraded = false;
      degradedSince = undefined;
    }
  }

  // #1676 — NEVER rejects. It is driven by a timer, and a rejection reaching the runtime
  // exits the Bun process (that is the incident: a heartbeat AbortError against a peer
  // host killed the orchestrator, and systemd restarted it into a register-retry loop).
  // The recovery work in the catch block is itself wrapped, so even a failure while
  // handling a failure degrades to a log line and the next beat.
  async function heartbeat(): Promise<void> {
    try {
      const runtime = runtimeMetadata();
      const providerSnapshot = await probeCache.getSnapshot();
      const res = await apiCall("POST", `/orchestrators/${config.id}/heartbeat`, {
        package: runtime.package,
        contracts: runtime.contracts,
        capabilities: runtime.capabilities,
        version: VERSION,
        protocolVersion: ORCHESTRATOR_PROTOCOL_VERSION,
        gitSha: reportedGitSha(),
        providers: providerSnapshot.providers,
        providerStatus: providerSnapshot.providerStatus,
        providerCatalog: providerSnapshot.providerCatalog,
      });
      if (!res.ok) throw new Error(`heartbeat failed: ${res.status}`);
      if (!connected) {
        console.error("[orchestrator] Reconnected to relay");
        connected = true;
        everConnected = true;
        reconnectMgr.reset();
        // #1478 — reconnected via a bare heartbeat (register() didn't re-run, e.g. it threw earlier
        // and the relay recovered on the next beat), so we owe a roster resync.
        markRosterResyncPending();
      }
      abortFailures = 0;
      markPeerHealthy();
      // #1478 — reliable driver: replay an owed roster resync on every healthy heartbeat. Covers a
      // reconnect whose register() cursor-bootstrap threw, a timeout-masked restart, and retries a
      // previously-failed PATCH. No-ops when nothing is pending.
      await flushRosterResync();
    } catch (err) {
      // #1676 — the whole recovery path is inside its own guard: `reconnect()` sleeps and
      // re-registers, and neither may turn a transient peer failure into process death.
      try {
        markPeerDegraded(err);
        if (isAbortError(err)) {
          abortFailures += 1;
          if (connected && abortFailures === 1) {
            console.error(`[orchestrator] Relay heartbeat timed out: ${err}; retrying quickly`);
          }
          if (abortFailures < ABORT_FAILURES_BEFORE_DISCONNECT) {
            await reconnect(TRANSIENT_ABORT_RECONNECT_MS);
            return;
          }
        } else {
          abortFailures = 0;
        }
        if (connected) {
          console.error(`[orchestrator] Lost connection to relay: ${err}`);
          connected = false;
        }
        await reconnect();
      } catch (recoveryErr) {
        console.error(`[orchestrator] Relay heartbeat recovery failed (will retry next beat): ${describeFailure(recoveryErr)}`);
      }
    }
  }

  async function reconnect(delayOverrideMs?: number): Promise<void> {
    const delayMs = delayOverrideMs ?? reconnectMgr.nextDelay();
    const label = delayMs < 1000 ? `${delayMs}ms` : `${Math.round(delayMs / 1000)}s`;
    console.error(`[orchestrator] Reconnecting in ${label}...`);
    await new Promise((resolve) => setTimeout(resolve, delayMs));
    try {
      await register();
    } catch {
      // Will retry on next heartbeat
    }
  }

  // #1478 (round-3) — the single serialized roster-write queue. Every roster PATCH (ordinary and
  // reconnect resync) enqueues here, so writes never overlap and land in request order; the last write
  // to land therefore reflects the newest roster. On success it clears `rosterResyncPending` ONLY when
  // BOTH hold: (a) its own `generation` is still the latest requested — a superseded write (a newer
  // request arrived while this one was queued/in-flight) leaves the obligation pending so the next
  // heartbeat re-syncs; and (b) #1478 round-4 — its captured `requestedEpoch` still equals the current
  // `rosterResyncEpoch`, i.e. no NEW obligation was marked after this write was REQUESTED. Without (b),
  // an ordinary write requested before a reconnect's obligation could 2xx-complete while still the
  // newest generation and wrongly clear that newer obligation.
  function patchRoster(agents: ManagedAgentReport[], exitedAgents: ManagedSessionExitDiagnostics[], generation: number, requestedEpoch: number, queuedSpawns: QueuedSpawn[]): Promise<boolean> {
    const run = rosterWriteChain.then(async () => {
      const res = await apiCall("PATCH", `/orchestrators/${config.id}/agents`, {
        agents,
        ...(exitedAgents.length ? { exitedAgents } : {}),
        // #1513 — always include the queued-spawn snapshot (captured at REQUEST time alongside `agents`,
        // not re-read here) so a full-replace roster write can't silently drop a still-queued spawn, and
        // a later clear that overtakes this write in the chain can't retroactively blank it. Empty array
        // = explicit clear server-side.
        queuedSpawns,
      });
      if (res.ok && generation === rosterGeneration && requestedEpoch === rosterResyncEpoch) rosterResyncPending = false;
      return res.ok;
    });
    // Keep the chain alive regardless of this write's outcome so one failure can't wedge the queue.
    rosterWriteChain = run.then(() => undefined, () => undefined);
    return run;
  }

  async function updateManagedAgents(agents: ManagedAgentReport[], exitedAgents: ManagedSessionExitDiagnostics[] = [], queuedSpawns?: QueuedSpawn[]): Promise<boolean> {
    // #1513 — an explicit queuedSpawns array replaces the held snapshot every roster PATCH carries;
    // omitting it (the common case) leaves the snapshot untouched so an ordinary roster push doesn't
    // drop a still-queued spawn. Capture the value to SEND synchronously here (request time), so a later
    // updateManagedAgents that mutates the snapshot before this write runs can't change what this PATCH
    // sends — the same request-time capture guarantee `agents` already has.
    if (queuedSpawns !== undefined) queuedSpawnsSnapshot = queuedSpawns;
    const queuedSpawnsToSend = queuedSpawnsSnapshot;
    // Claim a generation AND capture the resync epoch synchronously (before any await) so request order
    // — and thus queue order — is deterministic relative to concurrent callers, and the epoch reflects
    // the obligations that exist AT REQUEST time. #1478 — surface non-2xx so a resync retries.
    const generation = ++rosterGeneration;
    const requestedEpoch = rosterResyncEpoch;
    return patchRoster(agents, exitedAgents, generation, requestedEpoch, queuedSpawnsToSend);
  }

  async function pollCommands(signal?: AbortSignal): Promise<RelayCommand[]> {
    const url = `/commands?target=${encodeURIComponent(agentId)}&status=pending&limit=50`;
    const res = await http.request("GET", `/api${url}`, undefined, signal ? { signal } : undefined);
    if (!res.ok) return [];
    return await res.json() as RelayCommand[];
  }

  async function getCommand(commandId: string): Promise<RelayCommand | null> {
    const res = await apiCall("GET", `/commands/${encodeURIComponent(commandId)}`);
    if (!res.ok) return null;
    return await res.json() as RelayCommand;
  }

  async function updateCommand(commandId: string, status: string, result?: Record<string, unknown>, error?: string): Promise<boolean> {
    let lastError: unknown;
    for (let attempt = 1; attempt <= COMMAND_UPDATE_MAX_ATTEMPTS; attempt += 1) {
      try {
        const res = await apiCall("PATCH", `/commands/${encodeURIComponent(commandId)}`, {
          status,
          ...(result ? { result } : {}),
          ...(error ? { error } : {}),
        });
        if (res.ok) return true;
        const responseBody = await res.text().catch(() => "");
        const updateError = new CommandUpdateError(commandId, status, res.status, responseBody);
        if (!retryableCommandUpdateStatus(res.status)) throw updateError;
        lastError = updateError;
      } catch (updateError) {
        if (updateError instanceof CommandUpdateError && !retryableCommandUpdateStatus(updateError.responseStatus)) throw updateError;
        lastError = updateError;
      }
      if (attempt < COMMAND_UPDATE_MAX_ATTEMPTS) await commandUpdateRetryDelay(attempt);
    }
    throw lastError;
  }

  async function claimCommand(commandId: string, expectedType?: string): Promise<ClaimCommandResult> {
    // Atomic CAS on the relay: only ONE of {this claim, the reconciler's cancel} wins. On any transport
    // failure we return {claimed:false} (fail closed) — better to skip a publish than to push a canceled
    // command. A won publish claim carries a fence token that authorizes the later terminal settlement.
    const res = await apiCall("POST", `/commands/${encodeURIComponent(commandId)}/claim`, expectedType ? { type: expectedType } : {});
    if (!res.ok) return { claimed: false };
    const body = await res.json().catch(() => null) as { claimed?: boolean; token?: string } | null;
    return { claimed: Boolean(body?.claimed), ...(body?.token ? { token: body.token } : {}) };
  }

  async function fencePublishPush(commandId: string, expectedType?: string): Promise<boolean> {
    // #1452 round-12 HIGH#2 — re-fence the claimed publish command right before the push. On ANY transport
    // failure return false (fail closed): skipping a publish is always safe; pushing a possibly-timed-out
    // command is not. A won fence refreshes the lease so the push completes before the next sweep.
    const res = await apiCall("POST", `/commands/${encodeURIComponent(commandId)}/fence-push`, expectedType ? { type: expectedType } : {});
    if (!res.ok) return false;
    const body = await res.json().catch(() => null) as { fenced?: boolean } | null;
    return Boolean(body?.fenced);
  }

  async function settleReviewPublish(commandId: string, token: string, status: "succeeded" | "failed", result?: Record<string, unknown>, error?: string): Promise<boolean> {
    // #1452 round-13 HIGH#2 — settle the claimed publish through the token-gated route (the only external
    // path allowed to terminalize it). Fail-open reporting is fine here: an unsettled command ages out via
    // the TTL sweep, and the reconciler fail-closed cleans the pinned ref — it never orphans.
    const res = await apiCall("POST", `/commands/${encodeURIComponent(commandId)}/settle-review-publish`, {
      token,
      status,
      ...(result ? { result } : {}),
      ...(error ? { error } : {}),
    });
    return res.ok;
  }

  function startHeartbeatLoop(): void {
    if (heartbeatTimer) return;
    // #1676 — guardedInterval, never bare `setInterval(asyncFn)`: an async timer callback's
    // rejection has no handler and exits the process. heartbeat() is already written not to
    // reject; this is the structural belt so a future edit inside it cannot re-arm the bug.
    heartbeatTimer = guardedInterval("Relay heartbeat", config.heartbeatIntervalMs, heartbeat);
  }

  function stopHeartbeatLoop(): void {
    if (heartbeatTimer) {
      clearInterval(heartbeatTimer);
      heartbeatTimer = null;
    }
  }

  return {
    register,
    heartbeat,
    updateManagedAgents,
    pollCommands,
    getCommand,
    updateCommand,
    claimCommand,
    fencePublishPush,
    settleReviewPublish,
    setApiUrl(url: string) { apiUrl = url; },
    setOnReconnected(handler: () => void | boolean | Promise<void | boolean>) { onReconnected = handler; },
    startHeartbeatLoop,
    stopHeartbeatLoop,
    // Proxy a runner's expired token to the relay using our (long-lived)
    // orchestrator credential, returning a freshly minted runner token. Lets a
    // live runner self-heal an expired token instead of stranding the session.
    async remintRunnerToken(currentToken: string): Promise<RunnerTokenRemint | null> {
      try {
        const res = await apiCall("POST", `/orchestrators/${config.id}/runner-token`, { token: currentToken });
        if (!res.ok) return null;
        return await res.json() as RunnerTokenRemint;
      } catch {
        return null;
      }
    },
    async recheckSelfRestartSafePoint(agentId: string): Promise<{ ok: boolean; reason?: string }> {
      try {
        const res = await apiCall("POST", `/agents/${encodeURIComponent(agentId)}/self-restart-recheck`, {});
        if (!res.ok) return { ok: false, reason: `self-restart safe-point re-check returned ${res.status}` };
        const body = await res.json().catch(() => null) as { ok?: boolean; reason?: string } | null;
        if (!body || typeof body.ok !== "boolean") return { ok: false, reason: "self-restart safe-point re-check returned an invalid response" };
        return { ok: body.ok, reason: body.reason };
      } catch (error) {
        return { ok: false, reason: `self-restart safe-point re-check unavailable: ${isAbortError(error) ? "timed out" : String(error)}` };
      }
    },
    acquireProviderQuotaLease(orchestratorId: string, input: ProviderQuotaLeaseAcquireInput): Promise<ProviderQuotaLeaseAcquireResult> {
      return http.acquireProviderQuotaLease(orchestratorId, input);
    },
    releaseProviderQuotaLease(orchestratorId: string, input: ProviderQuotaLeaseAcquireInput & { leaseToken: string }): Promise<{ released: boolean }> {
      return http.releaseProviderQuotaLease(orchestratorId, input);
    },
    reportProviderQuota(input: ProviderQuotaUpdateInput): Promise<unknown> {
      return http.upsertProviderQuota(input);
    },
    reportCodexResetFired(event: CodexResetFiredEventReport): Promise<unknown> {
      return http.reportCodexResetFired(event);
    },
    getProviderQuotaConfig(): Promise<ProviderQuotaConfigMap> {
      return http.getProviderQuotaConfig();
    },
    getHealth(): RelayHealth {
      return {
        connected,
        degraded,
        consecutiveHeartbeatFailures,
        ...(lastHeartbeatError ? { lastHeartbeatError } : {}),
        ...(lastHeartbeatOkAt ? { lastHeartbeatOkAt } : {}),
        ...(degradedSince ? { degradedSince } : {}),
      };
    },
    get connected() { return connected; },
  };
}