import { describe, expect, test } from "bun:test";

import type { WorkbenchApplication, WorkbenchTemporalGateway } from "./contracts.ts";
import { createWorkbenchActivities } from "./activities.ts";
import { createWorkbenchDispatcher } from "./dispatcher.ts";
import { createCloudWorkbenchApplication } from "./cloud-application.ts";
import { createWorkbenchHttpApp } from "./http.ts";
import { workbenchRuntime } from "./runtime.ts";
import { runWorkbenchCli } from "../../tools/src/workbench-cli.ts";

test("temporal API runtime does not require worker Kafka publisher configuration", async () => {
  const runtime = workbenchRuntime({
    WORKBENCH_MODE: "temporal",
    WORKBENCH_CLOUD_API_URL: "http://cloud.internal",
    WORKBENCH_CLOUD_API_AUTHORIZATION: "internal-key",
    WORKBENCH_TEMPORAL_ADDRESS: "127.0.0.1:7233",
    WORKBENCH_TEMPORAL_NAMESPACE: "unidesk",
    WORKBENCH_TEMPORAL_TASK_QUEUE: "hwlab-v03-workbench",
    WORKBENCH_ACTIVITY_START_TO_CLOSE_TIMEOUT_MS: "30000",
    WORKBENCH_ACTIVITY_RETRY_INITIAL_INTERVAL_MS: "1000",
    WORKBENCH_ACTIVITY_RETRY_MAXIMUM_INTERVAL_MS: "30000",
    WORKBENCH_ACTIVITY_RETRY_MAXIMUM_ATTEMPTS: "5"
  }, { eventPublisher: null });
  expect(runtime.eventPublisher).toBeNull();
  await runtime.close();
});

describe("Workbench dispatcher", () => {
  test("submits through Temporal without synthesizing terminal state", async () => {
    const calls: string[] = [];
    const application: WorkbenchApplication = {
      async health() { return { status: "ready" }; },
      async createSession() { return { session: { sessionId: "ses_test" } }; },
      async admitTurn(input) { calls.push("admit"); return input; },
      async dispatchTurn() { throw new Error("API dispatcher must not dispatch directly"); },
      async cancelTurn() { throw new Error("API dispatcher must not cancel directly"); }
    };
    const temporal: WorkbenchTemporalGateway = {
      async submit(input) { calls.push("temporal"); return { workflowId: `workbench-turn-${input.traceId}`, workflowRunId: "run-1", reused: false }; },
      async cancel() { return { workflowId: "cancel", workflowRunId: "run-2", reused: false }; },
      async close() {}
    };
    const dispatch = createWorkbenchDispatcher({ application, temporal, mode: "temporal" });
    const result = await dispatch({ operation: "turn.submit", actor: { id: "usr_test" }, traceId: "trc_test", params: { sessionId: "ses_test", message: "hi" } });
    expect(calls).toEqual(["admit", "temporal"]);
    expect(result).toMatchObject({ ok: true, mode: "temporal", data: { status: "running", traceId: "trc_test", terminalAuthority: "hwlab.event.v1", resultSynthesized: false } });
  });

  test("cancel is a durable Temporal command", async () => {
    const application = stubApplication();
    const temporal: WorkbenchTemporalGateway = {
      async submit() { throw new Error("not used"); },
      async cancel(input) { return { workflowId: `workbench-cancel-${input.traceId}`, workflowRunId: "run-cancel", reused: false }; },
      async close() {}
    };
    const result = await createWorkbenchDispatcher({ application, temporal, mode: "temporal" })({ operation: "turn.cancel", actor: { id: "usr_test" }, traceId: "trc_test" });
    expect(result).toMatchObject({ ok: true, data: { status: "cancel_requested", resultSynthesized: false } });
  });
});

describe("Workbench Cloud application adapter", () => {
  test("persists admission before dispatch through distinct internal routes", async () => {
    const calls: Array<{ path: string; authorization: string | null }> = [];
    const application = createCloudWorkbenchApplication({
      baseUrl: "http://cloud.internal",
      authorization: "unified-internal-key",
      fetchImpl: (async (url: string | URL | Request, init?: RequestInit) => {
        calls.push({ path: new URL(String(url)).pathname, authorization: new Headers(init?.headers).get("authorization") });
        return Response.json({ ok: true, accepted: true, status: "admitted" });
      }) as typeof fetch
    });
    const input = { actor: { id: "usr_test" }, traceId: "trc_test", sessionId: "ses_test", params: { sessionId: "ses_test", message: "hi" } };
    await application.admitTurn(input);
    await application.dispatchTurn(input);
    expect(calls).toEqual([
      { path: "/v1/internal/workbench/admit", authorization: "Bearer unified-internal-key" },
      { path: "/v1/agent/chat", authorization: "Bearer unified-internal-key" }
    ]);
  });

  test("preserves an existing Bearer authorization scheme", async () => {
    let authorization: string | null = null;
    const application = createCloudWorkbenchApplication({
      baseUrl: "http://cloud.internal",
      authorization: "Bearer unified-internal-key",
      fetchImpl: (async (_url: string | URL | Request, init?: RequestInit) => {
        authorization = new Headers(init?.headers).get("authorization");
        return Response.json({ ok: true, session: { sessionId: "ses_test" } });
      }) as typeof fetch
    });
    await application.createSession({ actor: { id: "usr_test" }, params: {} });
    expect(authorization).toBe("Bearer unified-internal-key");
  });

  test("preserves Cloud API retryability and billing failure semantics", async () => {
    const application = createCloudWorkbenchApplication({
      baseUrl: "http://cloud.internal",
      authorization: "unified-internal-key",
      fetchImpl: (async () => Response.json({
        ok: false,
        error: { code: "insufficient_credits", layer: "billing", retryable: false, message: "credits exhausted", userMessage: "余额不足" }
      }, { status: 402 })) as typeof fetch
    });
    const input = { actor: { id: "usr_test" }, traceId: "trc_billing", sessionId: "ses_billing", params: { message: "hi" } };
    await expect(application.dispatchTurn(input)).rejects.toMatchObject({
      code: "insufficient_credits",
      layer: "billing",
      retryable: false,
      status: 402,
      userMessage: "余额不足"
    });
  });
});

describe("Workbench Temporal activities", () => {
  const activity = { startToCloseTimeoutMs: 1000, retryInitialIntervalMs: 100, retryMaximumIntervalMs: 500, retryMaximumAttempts: 3 };
  const input = { actor: { id: "usr_test" }, traceId: "trc_activity", sessionId: "ses_activity", params: { message: "hi" } };

  test("publishes semantic Kafka failure and stops retrying non-retryable billing errors", async () => {
    const published: any[] = [];
    const application = stubApplication();
    application.dispatchTurn = async () => { throw Object.assign(new Error("credits exhausted"), { code: "insufficient_credits", layer: "billing", retryable: false, userMessage: "余额不足" }); };
    const activities = createWorkbenchActivities(application, {
      activity,
      attempt: () => 1,
      eventPublisher: { async publish(event) { published.push(event); return { published: true }; } }
    });
    await expect(activities.dispatchWorkbenchTurn(input)).rejects.toMatchObject({ type: "insufficient_credits", nonRetryable: true });
    expect(published).toHaveLength(2);
    expect(published[0]).toMatchObject({ event: { type: "error", failureDomain: "billing", code: "insufficient_credits", retryable: false, willRetry: false, attempt: 1, maxAttempts: 3, terminal: false } });
    expect(published[1]).toMatchObject({ event: { type: "terminal_status", terminalStatus: "failed", errorCode: "insufficient_credits", message: "余额不足", terminal: true } });
  });

  test("publishes the YAML-controlled exponential retry without sealing the turn early", async () => {
    const published: any[] = [];
    const application = stubApplication();
    application.dispatchTurn = async () => { throw Object.assign(new Error("billing unavailable"), { code: "user_billing_preflight_failed", layer: "billing", retryable: true }); };
    const activities = createWorkbenchActivities(application, {
      activity,
      attempt: () => 2,
      eventPublisher: { async publish(event) { published.push(event); return { published: true }; } }
    });
    await expect(activities.dispatchWorkbenchTurn(input)).rejects.toMatchObject({ type: "user_billing_preflight_failed", nonRetryable: false });
    expect(published).toHaveLength(1);
    expect(published[0]).toMatchObject({ event: { type: "backend_status", retryPhase: "retryScheduled", attempt: 2, maxAttempts: 3, backoffMs: 200, terminal: false } });
  });
});

describe("Workbench native HTTP adapter", () => {
  test("native session list exposes the first user preview for summary-only rail loading", async () => {
    const app = createWorkbenchHttpApp({
      async dispatch() { return { ok: true }; },
      async snapshot() {
        return {
          sessions: {
            ses_title: {
              sessionId: "ses_title",
              updatedAt: "2026-07-20T01:00:00.000Z",
              messages: [{ role: "user", content: "persisted native title", createdAt: "2026-07-20T00:59:00.000Z" }]
            },
            ses_empty: { sessionId: "ses_empty", updatedAt: "2026-07-20T00:00:00.000Z", messages: [] }
          },
          turns: {}
        };
      }
    });

    const response = await app.fetch(new Request("http://native.test/v1/workbench/sessions"));
    expect(response.status).toBe(200);
    expect(await response.json()).toMatchObject({
      sessions: [
        { sessionId: "ses_title", firstUserMessagePreview: "persisted native title" },
        { sessionId: "ses_empty", firstUserMessagePreview: null }
      ]
    });
  });

  test("AgentRun native session list uses Kafka index summaries instead of admission messages", async () => {
    const app = createWorkbenchHttpApp({
      mode: "agentrun-native",
      kafkaEventBridge: {
        workbenchSessionSummaries() {
          return { ok: true, sessions: [{ sessionId: "ses_title", firstUserMessagePreview: "Kafka title", lastUserMessageAt: "2026-07-20T01:00:00.000Z" }] };
        }
      },
      async dispatch() { return { ok: true }; },
      async snapshot() {
        return {
          sessions: {
            ses_title: {
              sessionId: "ses_title",
              updatedAt: "2026-07-20T00:00:00.000Z",
              messages: [{ role: "user", content: "admission snapshot title" }]
            }
          },
          turns: {}
        };
      }
    });

    const response = await app.fetch(new Request("http://native.test/v1/workbench/sessions"));
    expect(await response.json()).toMatchObject({
      sessions: [{ sessionId: "ses_title", firstUserMessagePreview: "Kafka title", lastUserMessageAt: "2026-07-20T01:00:00.000Z" }]
    });
  });

  test("authorizes only the Workbench navigation entry", async () => {
    const app = createWorkbenchHttpApp({
      async dispatch() { return { ok: true }; },
      async snapshot() { return { sessions: {}, turns: {} }; }
    });

    for (const path of ["/auth/session", "/auth/bootstrap"]) {
      const response = await app.fetch(new Request(`http://native.test${path}`));
      expect(response.status).toBe(200);
      expect(await response.json()).toMatchObject({
        authenticated: true,
        access: { nav: { profileId: "workbench-native", allowedIds: ["workbench.code"] } }
      });
    }
  });

  test("serves read-only empty states for Workbench auxiliary panels", async () => {
    const app = createWorkbenchHttpApp({
      async dispatch() { return { ok: true }; },
      async snapshot() { return { sessions: {}, turns: {} }; }
    });

    for (const path of ["/v1/caserun/cases", "/v1/hwpod/specs?probe=1", "/v1/hwpod-node-ops"]) {
      const response = await app.fetch(new Request(`http://native.test${path}`));
      expect(response.status).toBe(200);
      expect(await response.json()).toMatchObject({ ok: true, status: "unavailable", mode: "native-test" });
    }
    const performance = await app.fetch(new Request("http://native.test/v1/web-performance", { method: "POST", body: "{}" }));
    expect(performance.status).toBe(202);
    expect(await performance.json()).toMatchObject({ ok: true, status: "accepted", persisted: false, mode: "native-test" });
  });

  test("L0 CLI observes AgentRun-derived events only through the Kafka SSE route", async () => {
    const subscribers = new Set<(envelope: any) => void>();
    const bridge = {
      capabilities: { liveKafkaSse: true },
      ready: Promise.resolve(),
      subscribeLiveHwlabEvents(listener: (envelope: any) => void) { subscribers.add(listener); return () => subscribers.delete(listener); }
    };
    const app = createWorkbenchHttpApp({
      mode: "agentrun-native",
      kafkaEventBridge: bridge,
      async dispatch() { return { ok: true }; },
      async snapshot() { return { sessions: {}, turns: {} }; }
    });
    let authorization: string | null = null;
    const server = Bun.serve({
      port: 0,
      fetch(request) {
        authorization = request.headers.get("authorization");
        return app.fetch(request);
      }
    });
    try {
      const pending = runWorkbenchCli([
        "events", "inspect",
        "--session-id", "ses_l0_sse",
        "--trace-id", "trc_l0_sse",
        "--over-api",
        "--api-url", `http://127.0.0.1:${server.port}`,
        "--timeout-ms", "1000"
      ], { HWLAB_API_KEY: "l0-sse-api-key" });
      for (let index = 0; index < 20 && subscribers.size === 0; index += 1) await Bun.sleep(5);
      for (const listener of subscribers) listener({ schema: "hwlab.event.v1", sessionId: "ses_l0_sse", traceId: "trc_l0_sse", event: { type: "backend", eventType: "backend", label: "agentrun:backend:command-created" } });
      expect(await pending).toMatchObject({
        ok: true,
        operation: "events.inspect",
        connected: true,
        connectedContract: {
          deliverySemantics: "live-only",
          realtimeSource: "hwlab.event.v1",
          replay: false,
          liveOnly: true,
          refreshReplay: null
        },
        eventCount: 1,
        eventTypes: ["backend"]
      });
      expect(authorization).toBe("Bearer l0-sse-api-key");
    } finally {
      server.stop(true);
      await app.close();
    }
  });

  test("L0 CLI reports the Workbench SSE HTTP status when authentication fails", async () => {
    const server = Bun.serve({ port: 0, fetch: () => new Response("unauthorized", { status: 401 }) });
    try {
      await expect(runWorkbenchCli([
        "events", "inspect",
        "--trace-id", "trc_l0_auth_failure",
        "--over-api",
        "--api-url", `http://127.0.0.1:${server.port}`,
        "--timeout-ms", "1000"
      ], {})).rejects.toMatchObject({
        code: "workbench_sse_read_failed",
        message: expect.stringContaining("Workbench SSE returned HTTP 401")
      });
    } finally {
      server.stop(true);
    }
  });

  test("L0 CLI replays retained Kafka events before the same live SSE stream", async () => {
    const sessionId = "ses_l0_refresh";
    const traceId = "trc_l0_refresh";
    const bridge = {
      capabilities: { liveKafkaSse: true, kafkaRefreshReplay: true },
      refreshReplay: { groupIdPrefix: "hwlab-l0-refresh", timeoutMs: 1000, scanLimit: 100, matchedEventLimit: 20, liveBufferLimit: 20 },
      ready: Promise.resolve(),
      subscribeLiveHwlabEvents() { return () => {}; },
      async queryHwlabEventRetention(options: any) {
        expect(options).toMatchObject({ sessionId, traceId, partitionKey: sessionId, fromBeginning: true });
        return {
          topic: "hwlab.event.v1",
          events: Array.from({ length: 100 }, (_, index) => ({
            topic: "hwlab.event.v1",
            partition: 0,
            offset: String(index),
            value: {
              schema: "hwlab.event.v1",
              eventType: "hwlab.trace.event.projected",
              eventId: `hwlab:evt_l0_refresh_${index}`,
              sourceEventId: `evt_l0_refresh_${index}`,
              sessionId,
              hwlabSessionId: sessionId,
              traceId,
              event: { type: "backend", eventType: "backend", sessionId, traceId, sourceEventId: `evt_l0_refresh_${index}` }
            }
          })),
          completionReason: "end-offset",
          completion: { reason: "end-offset", complete: true, barrierReached: true, retentionStartVerified: true },
          reachedEndOffsets: true,
          endOffsetsAvailable: true,
          endOffsets: [{ partition: 0, startOffset: "0", endOffset: "100" }],
          scannedCount: 100,
          parsedCount: 100,
          matchedCount: 100,
          filterRejectedCount: 0,
          keyRejectedCount: 0,
          partitionKeyScoped: true,
          targetPartition: 0,
          timing: { endOffsetSnapshotMs: 0, groupOffsetsMs: 0, consumerConnectMs: 0, consumerSubscribeMs: 0, scanMs: 0, cleanupMs: 0, indexLookupMs: 1, totalMs: 1 },
          index: { hit: true, source: "process-session-index", indexedEventCount: 300, indexedSessionCount: 3, incompleteSessionCount: 0, valuesRedacted: true }
        };
      }
    };
    const app = createWorkbenchHttpApp({
      mode: "agentrun-native",
      kafkaEventBridge: bridge,
      async dispatch() { return { ok: true }; },
      async snapshot() { return { sessions: {}, turns: {} }; }
    });
    const server = Bun.serve({ port: 0, fetch: (request) => app.fetch(request) });
    try {
      const result = await runWorkbenchCli([
        "events", "inspect",
        "--session-id", sessionId,
        "--trace-id", traceId,
        "--over-api",
        "--api-url", `http://127.0.0.1:${server.port}`,
        "--timeout-ms", "1000"
      ], {});
      expect(result).toMatchObject({
        ok: true,
        connected: true,
        connectedContract: {
          deliverySemantics: "kafka-retention-then-live",
          realtimeSource: "hwlab.event.v1",
          replay: true,
          liveOnly: false,
          refreshReplay: {
            phase: "live",
            counts: { replayed: 100 },
            query: {
              complete: true,
              scannedCount: 100,
              parsedCount: 100,
              matchedCount: 100,
              filterRejectedCount: 0,
              keyRejectedCount: 0,
              partitionKeyScoped: true,
              targetPartition: 0,
              index: { hit: true, source: "process-session-index", indexedEventCount: 300, indexedSessionCount: 3, incompleteSessionCount: 0, blocking: false },
              timing: { scanMs: 0, indexLookupMs: 1, totalMs: 1 }
            }
          }
        },
        eventCount: 100,
        eventTypes: ["backend"],
        eventTypeCounts: { backend: 100 }
      });
      expect(JSON.stringify(result).length).toBeLessThan(10_000);
    } finally {
      server.stop(true);
      await app.close();
    }
  });

  test("L0 CLI waits for explicit Kafka event semantics in one SSE window", async () => {
    const subscribers = new Set<(envelope: any) => void>();
    const bridge = {
      capabilities: { liveKafkaSse: true },
      ready: Promise.resolve(),
      subscribeLiveHwlabEvents(listener: (envelope: any) => void) { subscribers.add(listener); return () => subscribers.delete(listener); }
    };
    const app = createWorkbenchHttpApp({
      mode: "agentrun-native",
      kafkaEventBridge: bridge,
      async dispatch() { return { ok: true }; },
      async snapshot() { return { sessions: {}, turns: {} }; }
    });
    const server = Bun.serve({ port: 0, fetch: (request) => app.fetch(request) });
    try {
      const pending = runWorkbenchCli([
        "events", "inspect",
        "--session-id", "ses_l0_semantics",
        "--trace-id", "trc_l0_semantics",
        "--over-api",
        "--api-url", `http://127.0.0.1:${server.port}`,
        "--timeout-ms", "1000",
        "--wait-for", "user,backend,assistant,terminal,final"
      ], {});
      for (let index = 0; index < 20 && subscribers.size === 0; index += 1) await Bun.sleep(5);
      const events = [
        { type: "user", eventType: "user", createdAt: "2026-07-19T16:00:00.000Z" },
        { type: "backend", eventType: "backend", createdAt: "2026-07-19T16:00:02.000Z" },
        { type: "assistant", eventType: "assistant", createdAt: "2026-07-19T16:01:02.000Z", final: true, replyAuthority: true, finalResponse: { text: "done" } },
        { type: "result", eventType: "terminal", createdAt: "2026-07-19T16:01:03.000Z", terminal: true, terminalStatus: "completed", status: "completed" }
      ];
      for (const event of events) {
        for (const listener of subscribers) listener({ schema: "hwlab.event.v1", sessionId: "ses_l0_semantics", traceId: "trc_l0_semantics", event });
      }
      expect(await pending).toMatchObject({
        ok: true,
        eventCount: 4,
        eventTypes: ["user", "backend", "assistant", "terminal"],
        waitFor: ["user", "backend", "assistant", "terminal", "final"],
        observedSemantics: ["user", "backend", "assistant", "terminal", "final"],
        missingSemantics: [],
        semanticCounts: { user: 1, backend: 1, assistant: 1, terminal: 1, final: 1 },
        terminalStatuses: ["completed"],
        timing: {
          clock: "agentrun-source-event-created-at",
          phasesMs: {
            firstEventToBackend: 2000,
            admissionAndDispatch: null,
            runnerProvisioning: null,
            runtimePreparation: null,
            providerExecution: null,
            backendToAssistant: 60000,
            assistantToTerminal: 1000,
            firstEventToTerminal: 63000,
            firstEventToFinal: 62000
          },
          slowestGap: {
            durationMs: 60000,
            toIndex: 2,
            toEventType: "assistant"
          },
          valuesPrinted: false
        }
      });
    } finally {
      server.stop(true);
      await app.close();
    }
  });

  test("L0 CLI reports missing Kafka event semantics on timeout", async () => {
    const subscribers = new Set<(envelope: any) => void>();
    const app = createWorkbenchHttpApp({
      mode: "agentrun-native",
      kafkaEventBridge: {
        capabilities: { liveKafkaSse: true },
        ready: Promise.resolve(),
        subscribeLiveHwlabEvents(listener: (envelope: any) => void) { subscribers.add(listener); return () => subscribers.delete(listener); }
      },
      async dispatch() { return { ok: true }; },
      async snapshot() { return { sessions: {}, turns: {} }; }
    });
    const server = Bun.serve({ port: 0, fetch: (request) => app.fetch(request) });
    try {
      const pending = runWorkbenchCli([
        "events", "inspect",
        "--trace-id", "trc_l0_missing_terminal",
        "--over-api",
        "--api-url", `http://127.0.0.1:${server.port}`,
        "--timeout-ms", "250",
        "--wait-for", "assistant,terminal"
      ], {});
      for (let index = 0; index < 20 && subscribers.size === 0; index += 1) await Bun.sleep(2);
      for (const listener of subscribers) listener({ schema: "hwlab.event.v1", traceId: "trc_l0_missing_terminal", event: { type: "assistant", eventType: "assistant" } });
      expect(await pending).toMatchObject({
        ok: false,
        status: "event-timeout",
        observedSemantics: ["assistant"],
        missingSemantics: ["terminal"],
        terminalStatuses: [],
        error: { code: "workbench_kafka_sse_event_missing" }
      });
    } finally {
      server.stop(true);
      await app.close();
    }
  });

  test("L0 SSE exposes Kafka startup failure without waiting for a read model", async () => {
    const startupError = Object.assign(new Error("Kafka broker unavailable"), { code: "kafka_broker_unavailable" });
    const response = await createWorkbenchHttpApp({
      mode: "agentrun-native",
      kafkaEventBridge: {
        capabilities: { liveKafkaSse: true },
        liveReady: Promise.reject(startupError),
        subscribeLiveHwlabEvents() { return () => {}; }
      },
      async dispatch() { return { ok: true }; },
      async snapshot() { return { sessions: {}, turns: {} }; }
    }).fetch(new Request("http://localhost/v1/workbench/events?traceId=trc_l0_kafka_error"));
    expect(response.status).toBe(200);
    const body = await response.text();
    expect(body).toContain("event: workbench.connected");
    expect(body).toContain("event: workbench.error");
    expect(body).toContain("kafka_broker_unavailable");
  });
});

function stubApplication(): WorkbenchApplication {
  return {
    async health() { return {}; },
    async createSession() { return {}; },
    async admitTurn(input) { return input; },
    async dispatchTurn() { return {}; },
    async cancelTurn() { return {}; }
  };
}