import type { WorkbenchCommand } from "../../internal/workbench/contracts.ts";
import { workbenchRuntime } from "../../internal/workbench/runtime.ts";
import { workbenchNativeServiceCommand } from "./workbench-native-service.ts";

export async function runWorkbenchCli(argv: string[], env: Record<string, string | undefined> = process.env) {
  const parsed = parse(argv);
  if (parsed.help || parsed.positionals.length === 0) return help();
  if (parsed.positionals[0] === "service") return workbenchNativeServiceCommand({ service: parsed.positionals[1], action: parsed.positionals[2] ?? "status", cwd: process.cwd(), env });
  if (parsed.positionals[0] === "events" && parsed.positionals[1] === "inspect") return inspectEvents(parsed, env);
  const command = commandFrom(parsed, env);
  if (parsed.overApi) return overApi(command, parsed, env);
  const runtime = workbenchRuntime(env);
  try {
    const result = await runtime.dispatch(command);
    return { ...result, transport: "local", route: "application-dispatcher", identity: identity(result.data) };
  } finally {
    await runtime.close();
  }
}

function help() {
  return {
    ok: true,
    operation: "help",
    mode: "local-dispatcher-default",
    usage: [
      "hwlab-cli workbench health [--over-api]",
      "hwlab-cli workbench session create --actor-id ID [--provider-profile PROFILE] [--over-api]",
      "hwlab-cli workbench turn submit --actor-id ID --session-id ID --message TEXT [--trace-id ID] [--over-api]",
      "hwlab-cli workbench events inspect --session-id ID [--trace-id ID] --over-api [--timeout-ms MS] [--wait-for user,backend,assistant,terminal,final] [--min-events N]",
      "hwlab-cli workbench turn cancel --actor-id ID --trace-id ID [--over-api]",
      "hwlab-cli workbench service api|worker|web start|stop|restart|status|logs"
    ],
    transportContract: "--over-api only changes transport; --overapi is unsupported",
    localConfig: ["WORKBENCH_MODE", "WORKBENCH_NATIVE_STATE_FILE"],
    nativeServiceConfig: ["WORKBENCH_PUBLIC_BASE_URL", "WORKBENCH_API_BIND_HOST", "WORKBENCH_API_PROBE_HOST", "WORKBENCH_API_PORT", "WORKBENCH_API_IDLE_TIMEOUT_SECONDS", "WORKBENCH_WORKER_BIND_HOST", "WORKBENCH_WORKER_PROBE_HOST", "WORKBENCH_WORKER_HEALTH_PORT", "WORKBENCH_WEB_BIND_HOST", "WORKBENCH_WEB_PROBE_HOST", "WORKBENCH_WEB_PORT", "WORKBENCH_NATIVE_API_URL", "WORKBENCH_WEB_RUNTIME_CONFIG"],
    apiConfig: ["WORKBENCH_API_URL", "HWLAB_API_KEY"]
  };
}

async function inspectEvents(parsed: Parsed, env: Record<string, string | undefined>) {
  if (!parsed.overApi) throw codedError("workbench_events_over_api_required", "workbench events inspect requires --over-api so it exercises the product SSE route");
  const baseUrl = parsed.values["api-url"] || env.WORKBENCH_API_URL;
  if (!baseUrl) throw codedError("workbench_api_url_required", "WORKBENCH_API_URL or --api-url is required with --over-api");
  const sessionId = parsed.values["session-id"] ?? "";
  const traceId = parsed.values["trace-id"] ?? "";
  if (!sessionId && !traceId) throw codedError("workbench_realtime_scope_required", "--session-id or --trace-id is required");
  const timeoutMs = positiveInteger(parsed.values["timeout-ms"], 5000);
  const waitFor = semanticWait(parsed.values["wait-for"]);
  const minEvents = parsed.values["min-events"] ? positiveInteger(parsed.values["min-events"], 1) : waitFor.length > 0 ? 0 : 1;
  const params = new URLSearchParams();
  if (sessionId) params.set("sessionId", sessionId);
  if (traceId) params.set("traceId", traceId);
  const path = `/v1/workbench/events?${params.toString()}`;
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), timeoutMs);
  const frames: Array<{ name: string; data: Record<string, any> }> = [];
  let buffer = "";
  try {
    const response = await fetch(`${baseUrl.replace(/\/$/u, "")}${path}`, { signal: controller.signal });
    if (!response.ok || !response.body) throw codedError("workbench_sse_open_failed", `Workbench SSE returned HTTP ${response.status}`);
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    readStream: while (true) {
      const item = await reader.read();
      if (item.done) break;
      buffer += decoder.decode(item.value, { stream: true });
      let boundary = buffer.indexOf("\n\n");
      while (boundary >= 0) {
        const block = buffer.slice(0, boundary);
        buffer = buffer.slice(boundary + 2);
        const frame = parseSseBlock(block);
        if (frame) frames.push(frame);
        if (inspectionSatisfied(businessFrames(frames), minEvents, waitFor)) {
          await reader.cancel().catch(() => undefined);
          break readStream;
        }
        boundary = buffer.indexOf("\n\n");
      }
    }
  } catch (error: any) {
    const business = businessFrames(frames);
    if (error?.name !== "AbortError" && !(error?.code === "ECONNRESET" && inspectionSatisfied(business, minEvents, waitFor))) {
      const observed = semanticSummary(business);
      throw codedError("workbench_sse_read_failed", `Workbench SSE read failed: ${error?.code ?? error?.name ?? "unknown"}; frames=${frames.length}; businessEvents=${business.length}; observed=${observed.observedSemantics.join(",")}; connected=${frames.some((entry) => entry.name === "workbench.connected")}`);
    }
  } finally {
    clearTimeout(timer);
  }
  const business = businessFrames(frames);
  const connected = frames.some((entry) => entry.name === "workbench.connected");
  const semantics = semanticSummary(business);
  const missingSemantics = waitFor.filter((semantic) => !semantics.observedSemantics.includes(semantic));
  const ok = connected && inspectionSatisfied(business, minEvents, waitFor);
  return { ok, operation: "events.inspect", status: ok ? "passed" : "event-timeout", transport: "api-sse", baseUrl, route: `GET ${path}`, identity: { sessionId: sessionId || null, traceId: traceId || null }, connected, eventCount: business.length, eventTypes: business.map(eventType), timeoutMs, minEvents, waitFor, ...semantics, missingSemantics, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: `Workbench SSE did not deliver the required Kafka event conditions; missing=${missingSemantics.join(",") || (business.length < minEvents ? `min-events:${minEvents}` : "connection")}` }, valuesPrinted: false };
}

function commandFrom(parsed: Parsed, env: Record<string, string | undefined>): WorkbenchCommand {
  const [group, action] = parsed.positionals;
  if (group === "health") return { operation: "health" };
  const actor = { id: required(parsed, "actor-id"), role: parsed.values["actor-role"] ?? "user" };
  if (group === "session" && action === "create") return { operation: "session.create", actor, params: compact({ sessionId: parsed.values["session-id"], conversationId: parsed.values["conversation-id"], projectId: parsed.values["project-id"], providerProfile: parsed.values["provider-profile"] }) };
  if (group === "turn" && action === "submit") return { operation: "turn.submit", actor, traceId: parsed.values["trace-id"], params: compact({ sessionId: required(parsed, "session-id"), message: required(parsed, "message"), projectId: parsed.values["project-id"], providerProfile: parsed.values["provider-profile"], shortConnection: true }) };
  if (group === "turn" && action === "cancel") return { operation: "turn.cancel", actor, traceId: required(parsed, "trace-id"), params: compact({ sessionId: parsed.values["session-id"] }) };
  throw codedError("unsupported_workbench_command", `unsupported Workbench command: ${group} ${action ?? ""}`.trim());
}

async function overApi(command: WorkbenchCommand, parsed: Parsed, env: Record<string, string | undefined>) {
  const baseUrl = parsed.values["api-url"] || env.WORKBENCH_API_URL;
  if (!baseUrl) throw codedError("workbench_api_url_required", "WORKBENCH_API_URL or --api-url is required with --over-api");
  const headers: Record<string, string> = { "content-type": "application/json" };
  const apiKey = env.HWLAB_API_KEY;
  if (apiKey) headers.authorization = `Bearer ${apiKey}`;
  const route = "/v1/workbench/commands";
  const response = await fetch(`${baseUrl.replace(/\/$/u, "")}${route}`, { method: "POST", headers, body: JSON.stringify(command) });
  const body = await response.json().catch(() => null) as Record<string, any> | null;
  if (!body) throw codedError("workbench_api_invalid_json", `Workbench API returned invalid JSON (${response.status})`);
  return { ...body, transport: "api", baseUrl, route: `POST ${route}`, httpStatus: response.status, identity: identity(body.data) };
}

type Parsed = { positionals: string[]; values: Record<string, string>; overApi: boolean; help: boolean };
function parse(argv: string[]): Parsed {
  const parsed: Parsed = { positionals: [], values: {}, overApi: false, help: false };
  for (let index = 0; index < argv.length; index += 1) {
    const token = argv[index];
    if (token === "--overapi") throw codedError("unsupported_option", "--overapi was retired; use --over-api");
    if (token === "--over-api") parsed.overApi = true;
    else if (token === "--help" || token === "-h") parsed.help = true;
    else if (token.startsWith("--")) {
      const value = argv[index + 1];
      if (!value || value.startsWith("--")) throw codedError("option_value_required", `${token} requires a value`);
      parsed.values[token.slice(2)] = value;
      index += 1;
    } else parsed.positionals.push(token);
  }
  return parsed;
}
function identity(data: any) { return { sessionId: data?.sessionId ?? data?.session?.sessionId ?? null, traceId: data?.traceId ?? null, workflowId: data?.workflowId ?? null, workflowRunId: data?.workflowRunId ?? null }; }
function record(value: unknown): Record<string, any> | null { return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, any> : null; }
function positiveInteger(value: unknown, fallback: number): number { const parsed = Number.parseInt(String(value ?? ""), 10); return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback; }
const eventSemantics = ["user", "backend", "assistant", "terminal", "final"] as const;
type EventSemantic = typeof eventSemantics[number];
function semanticWait(value: unknown): EventSemantic[] {
  const requested = String(value ?? "").split(",").map((item) => item.trim().toLowerCase()).filter(Boolean);
  const unsupported = requested.filter((item) => !eventSemantics.includes(item as EventSemantic));
  if (unsupported.length > 0) throw codedError("workbench_event_semantic_unsupported", `unsupported --wait-for semantics: ${unsupported.join(",")}; supported=${eventSemantics.join(",")}`);
  return [...new Set(requested)] as EventSemantic[];
}
function businessFrames(frames: Array<{ name: string; data: Record<string, any> }>) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); }
function inspectionSatisfied(business: Array<{ name: string; data: Record<string, any> }>, minEvents: number, waitFor: EventSemantic[]) {
  if (business.length < minEvents) return false;
  const observed = semanticSummary(business).observedSemantics;
  return waitFor.every((semantic) => observed.includes(semantic));
}
function semanticSummary(business: Array<{ name: string; data: Record<string, any> }>) {
  const semanticCounts = Object.fromEntries(eventSemantics.map((semantic) => [semantic, 0])) as Record<EventSemantic, number>;
  const terminalStatuses: string[] = [];
  for (const frame of business) {
    const event = record(frame.data?.event) ?? {};
    for (const semantic of eventSemantics) if (matchesSemantic(event, semantic)) semanticCounts[semantic] += 1;
    if (matchesSemantic(event, "terminal")) {
      const status = String(event.terminalStatus ?? event.status ?? "").trim();
      if (status && !terminalStatuses.includes(status)) terminalStatuses.push(status);
    }
  }
  return { observedSemantics: eventSemantics.filter((semantic) => semanticCounts[semantic] > 0), semanticCounts, terminalStatuses };
}
function matchesSemantic(event: Record<string, any>, semantic: EventSemantic) {
  const type = String(event.eventType ?? event.type ?? "").trim().toLowerCase();
  if (semantic === "user") return type === "user" || type === "user_message";
  if (semantic === "backend") return type === "backend" || type === "backend_status";
  if (semantic === "assistant") return type === "assistant" || type === "assistant_message";
  if (semantic === "terminal") return event.terminal === true || type === "terminal" || type === "terminal_status" || type === "result" || event.agentRunEventType === "terminal_status";
  return event.final === true || event.replyAuthority === true || Boolean(record(event.finalResponse)?.text);
}
function eventType(frame: { data: Record<string, any> }) { return String(frame.data?.event?.eventType ?? frame.data?.event?.type ?? frame.data?.eventType ?? "unknown"); }
function parseSseBlock(block: string): { name: string; data: Record<string, any> } | null {
  if (!block || block.startsWith(":")) return null;
  const lines = block.split(/\r?\n/u);
  const name = lines.find((line) => line.startsWith("event:"))?.slice(6).trim() || "message";
  const dataText = lines.filter((line) => line.startsWith("data:" )).map((line) => line.slice(5).trimStart()).join("\n");
  if (!dataText) return null;
  const data = JSON.parse(dataText);
  return { name, data: record(data) ?? {} };
}
function compact(value: Record<string, unknown>) { return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== undefined)); }
function required(parsed: Parsed, key: string) { const value = parsed.values[key]; if (!value) throw codedError("option_required", `--${key} is required`); return value; }
function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }