/**
 * 制御盤（ラズパイ）とブラウザの中継（指示書AY 段階6。D2〜D4・D14・D15・D19・D20・D25）。
 * 期待値の正は指示書AY §3 表5（`__tests__/relay.test.ts`）。
 *
 * ★メモリ内だけで動く。サーバーのディスクに書かない★（一時ファイル・OS の一時領域も禁止。D3）。
 *   チャンクはメモリで受けて、待っているブラウザの応答（`ReadableStream`）へ流すだけ。
 * ★1プロセス前提★（`globalThis.__dwatchDsAgentRelay` のシングルトン。D4）。`next start` を複数立てる・cluster にすると壊れる。
 *   ★再起動で待ち行列と進行中のエクスポートは失われる★。
 * ★リクエストを待たせる時間（一覧15秒・ポーリング25秒・最初のチャンク30秒・チャンクの間隔と保留45秒・準備50秒）は
 *   どれも nginx の `proxy_read_timeout`（既定60秒）より短い★（設定値は `lib/config/ds-agent.ts`）。
 * ★時刻とタイマーは注入できる★（`RelayDeps`。テストは偽の時計を渡す）。既定のタイマーは `unref()` する。
 * ★`lib/logger` を import しない★。ログは `onEvent` で外に渡す（段階10）。
 *
 * 期限の境界: ★一覧の待ちだけは「ちょうどで時間切れ」★（§3 R2。依頼から15秒ちょうどで timeout）。
 * それ以外（未接続 R4・準備 R11・受け取り R14・最初のチャンク R18・保留 R31・間隔 R32・失効 R40）は
 * ★ちょうどはまだ有効、1ms 超えたら期限切れ★（タイマーを +1ms にする）。
 *
 * エクスポートの状態: 準備中 → 準備完了（受け取り待ち）→ 送信待ち → 流れ中 → 完了 / 失敗 / 取り消し / 失効。
 * ★準備中〜流れ中を「進行中」と数える★（D25 ⑤）。チャンク・complete・送信中の fail は ★export の指示の id★ に来る（D25 ①）。
 */
import { createHash, randomUUID, type Hash } from "node:crypto";
import {
  AGENT_OFFLINE_AFTER_SEC,
  CHUNK_MAX_BYTES,
  EXPORT_CLAIM_TIMEOUT_SEC,
  EXPORT_CONCURRENCY_PER_DEVICE,
  EXPORT_CONCURRENCY_TOTAL,
  EXPORT_FIRST_CHUNK_TIMEOUT_SEC,
  EXPORT_IDLE_TIMEOUT_SEC,
  EXPORT_MAX_BYTES,
  EXPORT_PREPARE_TIMEOUT_SEC,
  LIST_TIMEOUT_SEC,
  PENDING_COMMANDS_PER_DEVICE,
} from "@/lib/config/ds-agent";
import { toJstIso } from "@/lib/format/date";
import type {
  DsAgentCommand,
  DsAgentListResult,
  DsAgentListResultData,
  DsAgentPrepared,
  DsAgentStatus,
} from "@/lib/types";

export interface RelayLimits {
  offlineAfterSec: number;
  listTimeoutSec: number;
  exportPrepareTimeoutSec: number;
  exportClaimTimeoutSec: number;
  exportFirstChunkTimeoutSec: number;
  exportIdleTimeoutSec: number;
  chunkMaxBytes: number;
  exportMaxBytes: number;
  exportConcurrencyPerDevice: number;
  exportConcurrencyTotal: number;
  pendingCommandsPerDevice: number;
}

export type ExportFailReason =
  | "prepare_timeout"
  | "first_chunk_timeout"
  | "idle_timeout"
  | "flow_timeout"
  | "agent_error"
  | "too_large"
  | "chunk_invalid"
  | "chunk_out_of_order"
  | "complete_invalid"
  | "cancelled"
  | "expired";

/** 中継の出来事（段階10 でログへ写す）。★ファイル名・file_id・トークンを載せない★ */
export type RelayEvent =
  | { event: "ds_agent.connected" | "ds_agent.disconnected"; serial_number: string; agent_version: string | null }
  | {
      event: "ds_logs.export_failed";
      serial_number: string;
      stage: "prepare" | "send";
      reason: ExportFailReason;
      bytes_sent: number;
      agent_code?: string;
    }
  | {
      event: "ds_logs.export_completed";
      serial_number: string;
      file_count: number;
      total_bytes: number;
      /** 準備の依頼（createExport）から complete まで */
      duration_ms: number;
    };

export interface RelayDeps {
  /** ms */
  now: () => number;
  /** 既定は `setTimeout`（★`unref()` する★） */
  setTimer: (fn: () => void, ms: number) => unknown;
  clearTimer: (handle: unknown) => void;
  /** 既定は `lib/config/ds-agent.ts` */
  limits: RelayLimits;
  /** ログの受け口（段階10）。relay は logger を import しない。既定は `setRelayEventSink()` で登録した関数 */
  onEvent?: (e: RelayEvent) => void;
}

export type ListRequestResult =
  | { kind: "ok"; result: DsAgentListResultData }
  | { kind: "offline" | "timeout" | "busy" }
  | { kind: "agent_error"; code: string };

export type CreateExportResult =
  | { kind: "prepared"; exportId: string; totalBytes: number; fileCount: number; expiresInSec: number }
  | { kind: "offline" | "timeout" }
  | { kind: "busy"; code: "export_busy" | "agent_busy" }
  | { kind: "agent_error"; code: string };

export type ClaimExportResult =
  | { kind: "stream"; body: ReadableStream<Uint8Array>; totalBytes: number; fileCount: number }
  | { kind: "not_found" | "timeout" }
  | { kind: "agent_error"; code: string };

export type SubmitResult = "ok" | "not_found" | "gone";
export type PushChunkResult = "ok" | "invalid" | "out_of_order" | "cancelled" | "gone" | "not_found";
export type CompleteResult = "ok" | "invalid" | "cancelled" | "not_found";

export interface Relay {
  // ── 制御盤側 ──
  /** 認証の通ったすべての制御盤のリクエストで呼ぶ（D19）。`agentVersion` は接続の記録に使う（形の合わない値は呼び出し側で null） */
  touch(serial: string, agentVersion: string | null): void;
  poll(serial: string, agentVersion: string | null, waitSec: number, signal: AbortSignal): Promise<DsAgentCommand | null>;
  submitListResult(serial: string, commandId: string, result: DsAgentListResult): SubmitResult;
  /** 一覧の本体の形が不正なとき、ブラウザの一覧を `result_invalid` で終わらせる（指示書BJ J2。list だけ） */
  rejectListResult(serial: string, commandId: string): SubmitResult;
  /** ★`total_bytes` が上限を超えても "ok"★（エクスポート側を失敗にする。R13） */
  submitPrepared(serial: string, commandId: string, prepared: DsAgentPrepared): SubmitResult;
  failCommand(serial: string, commandId: string, code: string, message: string): SubmitResult;
  pushChunk(serial: string, exportCommandId: string, seq: number, bytes: Uint8Array, sha256Hex: string): Promise<PushChunkResult>;
  completeExport(serial: string, exportCommandId: string, totalBytes: number, sha256Hex: string): CompleteResult;
  // ── ブラウザ側 ──
  agentStatus(serial: string): DsAgentStatus;
  requestList(serial: string): Promise<ListRequestResult>;
  createExport(serial: string, fileIds: readonly string[], signal: AbortSignal): Promise<CreateExportResult>;
  claimExport(serial: string, exportId: string, signal: AbortSignal): Promise<ClaimExportResult>;
}

export const DEFAULT_RELAY_LIMITS: RelayLimits = {
  offlineAfterSec: AGENT_OFFLINE_AFTER_SEC,
  listTimeoutSec: LIST_TIMEOUT_SEC,
  exportPrepareTimeoutSec: EXPORT_PREPARE_TIMEOUT_SEC,
  exportClaimTimeoutSec: EXPORT_CLAIM_TIMEOUT_SEC,
  exportFirstChunkTimeoutSec: EXPORT_FIRST_CHUNK_TIMEOUT_SEC,
  exportIdleTimeoutSec: EXPORT_IDLE_TIMEOUT_SEC,
  chunkMaxBytes: CHUNK_MAX_BYTES,
  exportMaxBytes: EXPORT_MAX_BYTES,
  exportConcurrencyPerDevice: EXPORT_CONCURRENCY_PER_DEVICE,
  exportConcurrencyTotal: EXPORT_CONCURRENCY_TOTAL,
  pendingCommandsPerDevice: PENDING_COMMANDS_PER_DEVICE,
};

function defaultSetTimer(fn: () => void, ms: number): unknown {
  const handle = setTimeout(fn, ms);
  // ★テストとプロセスの終了を止めない★
  handle.unref?.();
  return handle;
}

function defaultClearTimer(handle: unknown): void {
  clearTimeout(handle as ReturnType<typeof setTimeout>);
}

// ── 内部の状態 ──

interface WaitingPoll {
  resolve: (command: DsAgentCommand | null) => void;
}

interface Device {
  serial: string;
  lastSeenMs: number | null;
  agentVersion: string | null;
  /** 最後に onEvent で知らせた接続状態（変化のときだけ知らせる。R39） */
  announcedConnected: boolean;
  offlineTimer: unknown;
  waitingPoll: WaitingPoll | null;
  /** まだ制御盤に渡していない指示 */
  queue: DsAgentCommand[];
}

interface ListRecord {
  kind: "list";
  id: string;
  serial: string;
  state: "pending" | "expired";
  resolve: (r: ListRequestResult) => void;
  timer: unknown;
}

type ExportState =
  | "preparing"
  | "prepared"
  | "sending"
  | "streaming"
  | "done"
  | "failed"
  | "cancelled"
  | "expired"
  | "prepare_timeout";

const IN_PROGRESS: ReadonlySet<ExportState> = new Set(["preparing", "prepared", "sending", "streaming"]);

interface ExportRecord {
  kind: "export";
  id: string;
  exportId: string;
  serial: string;
  fileCount: number;
  createdAtMs: number;
  state: ExportState;
  resolveCreate: ((r: CreateExportResult) => void) | null;
  resolveClaim: ((r: ClaimExportResult) => void) | null;
  /** 準備・受け取り・最初のチャンク・間隔の待ちのどれか1つ */
  timer: unknown;
  sendId: string | null;
  totalBytes: number;
  sha256: string;
  received: number;
  nextSeq: number;
  hasher: Hash | null;
  stream: ReadableStream<Uint8Array> | null;
  controller: ReadableStreamDefaultController<Uint8Array> | null;
  /** 流量制御で保留中のチャンクの POST */
  pendingPush: ((r: PushChunkResult) => void) | null;
  flowTimer: unknown;
}

interface SendRecord {
  kind: "send";
  id: string;
  serial: string;
}

type CommandRecord = ListRecord | ExportRecord | SendRecord;

export function createRelay(partial: Partial<RelayDeps> = {}): Relay {
  const now = partial.now ?? (() => Date.now());
  const setTimer = partial.setTimer ?? defaultSetTimer;
  const clearTimer = partial.clearTimer ?? defaultClearTimer;
  const limits = partial.limits ?? DEFAULT_RELAY_LIMITS;
  const onEvent = partial.onEvent ?? ((e: RelayEvent) => globalThis.__dwatchDsAgentRelayEventSink?.(e));

  const devices = new Map<string, Device>();
  const commands = new Map<string, CommandRecord>();
  const exportIds = new Map<string, string>();

  /** 終わった指示を消すまでの時間（その間は 410 を返せる墓標として残す） */
  const tombstoneMs = () => limits.exportIdleTimeoutSec * 1000;
  /** ★ちょうどはまだ有効、1ms 超えたら期限切れ★ */
  const exclusiveMs = (sec: number) => sec * 1000 + 1;

  function emit(e: RelayEvent): void {
    onEvent?.(e);
  }

  function clear(handle: unknown): void {
    if (handle !== null && handle !== undefined) clearTimer(handle);
  }

  // ── 接続 ──

  function device(serial: string): Device {
    let d = devices.get(serial);
    if (!d) {
      d = {
        serial,
        lastSeenMs: null,
        agentVersion: null,
        announcedConnected: false,
        offlineTimer: null,
        waitingPoll: null,
        queue: [],
      };
      devices.set(serial, d);
    }
    return d;
  }

  function isConnected(d: Device | undefined): boolean {
    if (!d) return false;
    if (d.waitingPoll) return true;
    return d.lastSeenMs !== null && now() - d.lastSeenMs <= limits.offlineAfterSec * 1000;
  }

  function scheduleOfflineCheck(d: Device): void {
    clear(d.offlineTimer);
    d.offlineTimer = null;
    if (d.lastSeenMs === null) return;
    const delay = Math.max(0, d.lastSeenMs + exclusiveMs(limits.offlineAfterSec) - now());
    d.offlineTimer = setTimer(() => {
      d.offlineTimer = null;
      if (isConnected(d)) {
        // ポーリングの最中は、終わったときに見直す
        if (!d.waitingPoll) scheduleOfflineCheck(d);
        return;
      }
      if (d.announcedConnected) {
        d.announcedConnected = false;
        emit({ event: "ds_agent.disconnected", serial_number: d.serial, agent_version: d.agentVersion });
      }
    }, delay);
  }

  function markSeen(d: Device, agentVersion: string | null): void {
    d.lastSeenMs = now();
    if (agentVersion !== null) d.agentVersion = agentVersion;
    if (!d.announcedConnected) {
      d.announcedConnected = true;
      emit({ event: "ds_agent.connected", serial_number: d.serial, agent_version: d.agentVersion });
    }
    scheduleOfflineCheck(d);
  }

  /** 制御盤に指示を渡す（ポーリングが待っていればすぐ、いなければ待ち行列へ） */
  function deliver(d: Device, command: DsAgentCommand): void {
    if (d.waitingPoll) {
      d.waitingPoll.resolve(command);
    } else {
      d.queue.push(command);
    }
  }

  function removeQueued(serial: string, commandId: string | null): void {
    if (commandId === null) return;
    const d = devices.get(serial);
    if (!d) return;
    d.queue = d.queue.filter((c) => c.command_id !== commandId);
  }

  function lookup(serial: string, commandId: string): CommandRecord | undefined {
    const rec = commands.get(commandId);
    // ★別の制御盤の指示は「無い」と同じ★（403 にしない。R35）。照合は `===`（D-4）
    return rec && rec.serial === serial ? rec : undefined;
  }

  function scheduleRemoval(rec: ExportRecord | ListRecord): void {
    setTimer(() => {
      commands.delete(rec.id);
      if (rec.kind === "export") {
        exportIds.delete(rec.exportId);
        if (rec.sendId) commands.delete(rec.sendId);
      }
    }, tombstoneMs());
  }

  // ── エクスポートの終わらせ方 ──

  function countInProgress(serial: string | null): number {
    let n = 0;
    for (const rec of commands.values()) {
      if (rec.kind === "export" && IN_PROGRESS.has(rec.state) && (serial === null || rec.serial === serial)) n++;
    }
    return n;
  }

  function endExport(
    rec: ExportRecord,
    state: "failed" | "cancelled" | "expired" | "prepare_timeout",
    reason: ExportFailReason,
    opts: { createResult?: CreateExportResult; claimResult?: ClaimExportResult; agentCode?: string } = {}
  ): void {
    if (!IN_PROGRESS.has(rec.state)) return;
    const previous = rec.state;
    rec.state = state;
    clear(rec.timer);
    rec.timer = null;
    clear(rec.flowTimer);
    rec.flowTimer = null;
    removeQueued(rec.serial, rec.id);
    removeQueued(rec.serial, rec.sendId);

    if (previous === "preparing") {
      rec.resolveCreate?.(opts.createResult ?? { kind: "agent_error", code: opts.agentCode ?? reason });
    } else if (previous === "sending") {
      rec.resolveClaim?.(opts.claimResult ?? { kind: "agent_error", code: opts.agentCode ?? reason });
    } else if (previous === "streaming") {
      try {
        rec.controller?.error(new Error(`export ${reason}`));
      } catch {
        // 既にブラウザが取り消したストリーム
      }
    }
    rec.resolveCreate = null;
    rec.resolveClaim = null;
    rec.pendingPush?.("cancelled");
    rec.pendingPush = null;

    emit({
      event: "ds_logs.export_failed",
      serial_number: rec.serial,
      stage: previous === "preparing" ? "prepare" : "send",
      reason,
      bytes_sent: rec.received,
      ...(opts.agentCode !== undefined ? { agent_code: opts.agentCode } : {}),
    });
    scheduleRemoval(rec);
  }

  /** 流れ中: 次のチャンクを待つ（間隔の上限。R32） */
  function startIdleTimer(rec: ExportRecord): void {
    clear(rec.timer);
    rec.timer = setTimer(() => {
      rec.timer = null;
      endExport(rec, "failed", "idle_timeout");
    }, exclusiveMs(limits.exportIdleTimeoutSec));
  }

  function onPull(rec: ExportRecord): void {
    if (!rec.pendingPush) return;
    const resolve = rec.pendingPush;
    rec.pendingPush = null;
    clear(rec.flowTimer);
    rec.flowTimer = null;
    resolve("ok");
    if (rec.state === "streaming") startIdleTimer(rec);
  }

  // ── Relay ──

  const relay: Relay = {
    touch(serial, agentVersion) {
      markSeen(device(serial), agentVersion);
    },

    poll(serial, agentVersion, waitSec, signal) {
      const d = device(serial);
      markSeen(d, agentVersion);
      // ★同じ制御盤の2本目は古い方を 204★（R9）
      d.waitingPoll?.resolve(null);

      const queued = d.queue.shift();
      if (queued) return Promise.resolve(queued);

      return new Promise<DsAgentCommand | null>((resolve) => {
        let settled = false;
        const timer = setTimer(() => finish(null), waitSec * 1000);
        const onAbort = () => finish(null);
        const waiting: WaitingPoll = { resolve: (command) => finish(command) };
        function finish(command: DsAgentCommand | null): void {
          if (settled) return;
          settled = true;
          clear(timer);
          signal.removeEventListener("abort", onAbort);
          if (d.waitingPoll === waiting) d.waitingPoll = null;
          markSeen(d, null);
          resolve(command);
        }
        d.waitingPoll = waiting;
        if (signal.aborted) finish(null);
        else signal.addEventListener("abort", onAbort, { once: true });
      });
    },

    submitListResult(serial, commandId, result) {
      const rec = lookup(serial, commandId);
      if (!rec || rec.kind !== "list") return "not_found";
      if (rec.state === "expired") return "gone";
      clear(rec.timer);
      commands.delete(rec.id);
      rec.resolve("error" in result ? { kind: "agent_error", code: result.error.code } : { kind: "ok", result });
      return "ok";
    },

    rejectListResult(serial, commandId) {
      const rec = lookup(serial, commandId);
      if (!rec || rec.kind !== "list") return "not_found";
      if (rec.state === "expired") return "gone";
      clear(rec.timer);
      commands.delete(rec.id);
      rec.resolve({ kind: "agent_error", code: "result_invalid" });
      return "ok";
    },

    submitPrepared(serial, commandId, prepared) {
      const rec = lookup(serial, commandId);
      if (!rec || rec.kind !== "export") return "not_found";
      if (rec.state !== "preparing") return "gone";
      if (prepared.total_bytes > limits.exportMaxBytes) {
        // ★zip の大きさで断る（D15）。prepared の応答は 204 のまま★
        endExport(rec, "failed", "too_large", { agentCode: "too_large" });
        return "ok";
      }
      clear(rec.timer);
      rec.state = "prepared";
      rec.totalBytes = prepared.total_bytes;
      rec.sha256 = prepared.sha256;
      rec.fileCount = prepared.file_count;
      rec.timer = setTimer(() => {
        rec.timer = null;
        endExport(rec, "expired", "expired");
      }, exclusiveMs(limits.exportClaimTimeoutSec));
      const resolve = rec.resolveCreate;
      rec.resolveCreate = null;
      resolve?.({
        kind: "prepared",
        exportId: rec.exportId,
        totalBytes: rec.totalBytes,
        fileCount: rec.fileCount,
        expiresInSec: limits.exportClaimTimeoutSec,
      });
      return "ok";
    },

    failCommand(serial, commandId, code) {
      const rec = lookup(serial, commandId);
      if (!rec || rec.kind === "send") return "not_found";
      if (rec.kind === "list") {
        if (rec.state === "expired") return "gone";
        clear(rec.timer);
        commands.delete(rec.id);
        rec.resolve({ kind: "agent_error", code });
        return "ok";
      }
      if (!IN_PROGRESS.has(rec.state)) return "gone";
      endExport(rec, "failed", "agent_error", { agentCode: code });
      return "ok";
    },

    async pushChunk(serial, exportCommandId, seq, bytes, sha256Hex) {
      const rec = lookup(serial, exportCommandId);
      // ★send の指示の id には送られてこない約束（D25 ①）★
      if (!rec || rec.kind !== "export") return "not_found";
      if (rec.state === "expired") return "gone";
      if (!IN_PROGRESS.has(rec.state)) return "cancelled";
      if (rec.state === "preparing" || rec.state === "prepared") return "out_of_order";

      // ★前のチャンクの 204 を待たずに次を送った（指示書BJ J8）★
      if (rec.pendingPush !== null) {
        endExport(rec, "failed", "chunk_out_of_order", { agentCode: "chunk_out_of_order" });
        return "out_of_order";
      }

      if (seq !== rec.nextSeq) {
        endExport(rec, "failed", "chunk_out_of_order", { agentCode: "chunk_out_of_order" });
        return "out_of_order";
      }
      const actual = createHash("sha256").update(bytes).digest("hex");
      if (
        bytes.length === 0 ||
        bytes.length > limits.chunkMaxBytes ||
        actual !== sha256Hex ||
        rec.received + bytes.length > rec.totalBytes
      ) {
        endExport(rec, "failed", "chunk_invalid", { agentCode: "chunk_invalid" });
        return "invalid";
      }

      clear(rec.timer);
      rec.timer = null;
      rec.hasher ??= createHash("sha256");
      rec.hasher.update(bytes);
      rec.received += bytes.length;
      rec.nextSeq += 1;

      if (rec.state === "sending") {
        rec.state = "streaming";
        const resolve = rec.resolveClaim;
        rec.resolveClaim = null;
        resolve?.({ kind: "stream", body: rec.stream!, totalBytes: rec.totalBytes, fileCount: rec.fileCount });
      }

      const controller = rec.controller!;
      controller.enqueue(bytes);
      if ((controller.desiredSize ?? 0) > 0) {
        startIdleTimer(rec);
        return "ok";
      }
      // ★流量制御: ブラウザが読む（pull が呼ばれる）まで 204 を返さない（最大45秒。D20）★
      return new Promise<PushChunkResult>((resolve) => {
        rec.pendingPush = resolve;
        rec.flowTimer = setTimer(() => {
          rec.flowTimer = null;
          endExport(rec, "failed", "flow_timeout");
        }, exclusiveMs(limits.exportIdleTimeoutSec));
      });
    },

    completeExport(serial, exportCommandId, totalBytes, sha256Hex) {
      const rec = lookup(serial, exportCommandId);
      if (!rec || rec.kind !== "export") return "not_found";
      if (!IN_PROGRESS.has(rec.state)) return "cancelled";
      // ★受け取った合計 ＝ prepared の total_bytes ＝ complete の total_bytes、かつ SHA-256 も3つとも一致（D15）★
      const receivedSha = rec.hasher ? rec.hasher.digest("hex") : null;
      rec.hasher = null;
      if (
        rec.state !== "streaming" ||
        rec.received !== rec.totalBytes ||
        totalBytes !== rec.totalBytes ||
        sha256Hex !== rec.sha256 ||
        receivedSha !== rec.sha256
      ) {
        endExport(rec, "failed", "complete_invalid", { agentCode: "complete_invalid" });
        return "invalid";
      }
      rec.state = "done";
      clear(rec.timer);
      rec.timer = null;
      try {
        rec.controller?.close();
      } catch {
        // 既にブラウザが取り消したストリーム
      }
      emit({
        event: "ds_logs.export_completed",
        serial_number: rec.serial,
        file_count: rec.fileCount,
        total_bytes: rec.totalBytes,
        duration_ms: now() - rec.createdAtMs,
      });
      scheduleRemoval(rec);
      return "ok";
    },

    agentStatus(serial) {
      const d = devices.get(serial);
      return {
        connected: isConnected(d),
        last_seen_at: d && d.lastSeenMs !== null ? toJstIso(new Date(d.lastSeenMs)) : null,
        agent_version: d ? d.agentVersion : null,
      };
    },

    requestList(serial) {
      const d = devices.get(serial);
      if (!d || !isConnected(d)) return Promise.resolve({ kind: "offline" });
      if (d.queue.length >= limits.pendingCommandsPerDevice) return Promise.resolve({ kind: "busy" });

      return new Promise<ListRequestResult>((resolve) => {
        const id = randomUUID();
        const rec: ListRecord = { kind: "list", id, serial, state: "pending", resolve, timer: null };
        // ★一覧の待ちだけは「ちょうどで時間切れ」（§3 R2）★
        rec.timer = setTimer(() => {
          rec.timer = null;
          rec.state = "expired";
          removeQueued(serial, id);
          resolve({ kind: "timeout" });
          scheduleRemoval(rec);
        }, limits.listTimeoutSec * 1000);
        commands.set(id, rec);
        // ★エクスポートの進行中でも断らない（D25 ③）★
        deliver(d, { command_id: id, kind: "list", params: {} });
      });
    },

    createExport(serial, fileIds, signal) {
      const d = devices.get(serial);
      if (!d || !isConnected(d)) return Promise.resolve({ kind: "offline" });
      if (
        countInProgress(serial) >= limits.exportConcurrencyPerDevice ||
        countInProgress(null) >= limits.exportConcurrencyTotal
      ) {
        return Promise.resolve({ kind: "busy", code: "export_busy" });
      }
      if (d.queue.length >= limits.pendingCommandsPerDevice) {
        return Promise.resolve({ kind: "busy", code: "agent_busy" });
      }

      return new Promise<CreateExportResult>((resolve) => {
        const id = randomUUID();
        const rec: ExportRecord = {
          kind: "export",
          id,
          exportId: randomUUID(),
          serial,
          fileCount: fileIds.length,
          createdAtMs: now(),
          state: "preparing",
          resolveCreate: resolve,
          resolveClaim: null,
          timer: null,
          sendId: null,
          totalBytes: 0,
          sha256: "",
          received: 0,
          nextSeq: 0,
          hasher: null,
          stream: null,
          controller: null,
          pendingPush: null,
          flowTimer: null,
        };
        rec.timer = setTimer(() => {
          rec.timer = null;
          endExport(rec, "prepare_timeout", "prepare_timeout", { createResult: { kind: "timeout" } });
        }, exclusiveMs(limits.exportPrepareTimeoutSec));
        commands.set(id, rec);
        exportIds.set(rec.exportId, id);
        // ★準備の POST が切れたらエクスポートを取り消す（4-9）★
        const onAbort = () => {
          if (rec.state === "preparing") endExport(rec, "cancelled", "cancelled");
        };
        if (signal.aborted) onAbort();
        else signal.addEventListener("abort", onAbort, { once: true });
        if (rec.state !== "preparing") return;
        deliver(d, {
          command_id: id,
          kind: "export",
          params: { file_ids: [...fileIds], max_bytes: limits.exportMaxBytes, chunk_bytes: limits.chunkMaxBytes },
        });
      });
    },

    claimExport(serial, exportId, signal) {
      const id = exportIds.get(exportId);
      const rec = id === undefined ? undefined : lookup(serial, id);
      // ★1回だけ・失効後・シリアル違いは 404★（R14〜R16）
      if (!rec || rec.kind !== "export" || rec.state !== "prepared") return Promise.resolve({ kind: "not_found" });

      return new Promise<ClaimExportResult>((resolve) => {
        clear(rec.timer);
        rec.state = "sending";
        rec.resolveClaim = resolve;
        rec.stream = new ReadableStream<Uint8Array>(
          {
            start(controller) {
              rec.controller = controller;
            },
            pull() {
              onPull(rec);
            },
            cancel() {
              endExport(rec, "cancelled", "cancelled");
            },
          },
          { highWaterMark: 1 }
        );
        rec.timer = setTimer(() => {
          rec.timer = null;
          endExport(rec, "failed", "first_chunk_timeout", { claimResult: { kind: "timeout" } });
        }, exclusiveMs(limits.exportFirstChunkTimeoutSec));

        const sendId = randomUUID();
        rec.sendId = sendId;
        commands.set(sendId, { kind: "send", id: sendId, serial });
        // ★ブラウザの切断で取り消す（`cancel()` と `request.signal` の両方。4-9）★
        const onAbort = () => endExport(rec, "cancelled", "cancelled", { claimResult: { kind: "not_found" } });
        if (signal.aborted) {
          onAbort();
          return;
        }
        signal.addEventListener("abort", onAbort, { once: true });
        deliver(device(serial), { command_id: sendId, kind: "send", params: { export_command_id: rec.id } });
      });
    },
  };

  return relay;
}

declare global {
  /** 中継のシングルトン（★1プロセス前提。D4★）。開発時の HMR で二重にしないため `globalThis` に置く */
  var __dwatchDsAgentRelay: Relay | undefined;
  /** 中継の出来事の受け口（`setRelayEventSink()`）。ルートのバンドルが別でも同じ関数を見るよう `globalThis` に置く */
  var __dwatchDsAgentRelayEventSink: ((e: RelayEvent) => void) | undefined;
}

/**
 * `onEvent` を渡さずに作った relay（`getRelay()`）の出来事の受け口を登録する。
 * ★relay は logger を import しない★ので、ログへ写すのはルート側（`lib/ds-agent/relay-log.ts`）の仕事。
 */
export function setRelayEventSink(sink: ((e: RelayEvent) => void) | undefined): void {
  globalThis.__dwatchDsAgentRelayEventSink = sink;
}

export function getRelay(): Relay {
  globalThis.__dwatchDsAgentRelay ??= createRelay();
  return globalThis.__dwatchDsAgentRelay;
}

/** ★テスト専用★。`null` で次の `getRelay()` が作り直す */
export function __setRelayForTesting(relay: Relay | null): void {
  globalThis.__dwatchDsAgentRelay = relay ?? undefined;
}