import assert from "node:assert/strict";
import { test } from "bun:test";

import { nodeWebObserveRunnerRealtimeSource } from "./hwlab-node-web-observe-runner-realtime-source";

test("realtime fanout keeps the YAML terminal timeout when durationMs is absent", () => {
  const source = nodeWebObserveRunnerRealtimeSource();
  const build = new Function(
    "boundedInteger",
    "sanitize",
    `${source}\nreturn realtimeFanoutEffectiveProfile;`,
  ) as (
    boundedInteger: (value: unknown, fallback: number, min: number, max: number) => number,
    sanitize: <T>(value: T) => T,
  ) => (profile: Record<string, unknown>, durationMs: unknown) => Record<string, unknown>;
  const effectiveProfile = build(
    (value, fallback, min, max) => Math.min(max, Math.max(min, Number.isFinite(Number(value)) ? Number(value) : fallback)),
    (value) => value,
  );
  const profile = {
    subscriberCount: 2,
    debugStreams: ["stdio", "agentrun", "hwlab"],
    fromBeginning: false,
    barrierTimeoutMs: 45_000,
    terminalTimeoutMs: 480_000,
    terminalQuietMs: 2_000,
    reconnectQuietMs: 3_000,
    forbiddenRequestPaths: [],
    requiredEventTypes: { agentrun: ["terminal_status"], hwlab: ["terminal"] },
    expectedKafka: {
      topics: { stdio: "codex-stdio.raw.v1", agentrun: "agentrun.event.v1", hwlab: "hwlab.event.v1" },
      groups: { directPublish: "direct", transactionalProjector: "projector", liveSse: "live" },
      capabilities: { directPublish: true, liveKafkaSse: true },
    },
  };

  assert.equal(effectiveProfile(profile, null).terminalTimeoutMs, 480_000);
  assert.equal(effectiveProfile(profile, undefined).terminalTimeoutMs, 480_000);
  assert.equal(effectiveProfile(profile, 5_000).terminalTimeoutMs, 5_000);
});

test("realtime fanout retries a YAML-bounded transient health navigation", async () => {
  const source = nodeWebObserveRunnerRealtimeSource();
  const build = new Function(
    "sleep",
    "isRetryableNavigationError",
    "baseUrl",
    `${source}\nreturn realtimeGotoHealth;`,
  ) as (
    sleep: (milliseconds: number) => Promise<void>,
    isRetryableNavigationError: (message: string) => boolean,
    baseUrl: string,
  ) => (page: { goto: () => Promise<{ status: () => number }> }, timeoutMs: number, profile: Record<string, number>, label: string) => Promise<Record<string, unknown>>;
  const gotoHealth = build(
    async () => {},
    (message) => /ERR_NETWORK_CHANGED/u.test(message),
    "https://hwlab.example.test",
  );
  let calls = 0;
  const result = await gotoHealth({
    async goto() {
      calls += 1;
      if (calls === 1) throw new Error("page.goto: net::ERR_NETWORK_CHANGED");
      return { status: () => 200 };
    },
  }, 45_000, { navigationMaxAttempts: 4, navigationRetryDelayMs: 1_000 }, "subscriber-a");

  assert.equal(result.ok, true);
  assert.equal(calls, 2);
  assert.equal((result.attempts as unknown[]).length, 2);
});

test("realtime fanout validates stdio against the scoped AgentRun session lineage", () => {
  const source = nodeWebObserveRunnerRealtimeSource();
  const build = new Function(`${source}\nreturn realtimeExpectedStreamSessionId;`) as () => (
    stream: string,
    hwlabSessionId: string,
  ) => string;
  const expectedStreamSessionId = build();
  const hwlabSessionId = "ses_5ec4e141-6abc-466d-9afe-049f7c0ac105";

  assert.equal(expectedStreamSessionId("stdio", hwlabSessionId), "ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105");
  assert.equal(expectedStreamSessionId("agentrun", hwlabSessionId), hwlabSessionId);
  assert.equal(expectedStreamSessionId("hwlab", hwlabSessionId), hwlabSessionId);
});