pikasTech-HWLAB/tools/src/workbench-cli.ts

import type { WorkbenchCommand } from "../../internal/workbench/contracts.ts";
import { workbenchRuntime } from "../../internal/workbench/runtime.ts";
import { workbenchNativeServiceCommand, workbenchNativeServicesStatus } 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" && parsed.positionals[1] === "status") {
    if (parsed.positionals.length !== 2) throw codedError("invalid_arguments", "workbench service status does not accept positional arguments");
    return workbenchNativeServicesStatus({ cwd: process.cwd(), env });
  }
  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] [--timing-detail summary|full]",
      "hwlab-cli workbench turn cancel --actor-id ID --trace-id ID [--over-api]",
      "hwlab-cli workbench service status",
      "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 timingDetail = timingDetailValue(parsed.values["timing-detail"]);
  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 headers: Record<string, string> = {};
  const apiKey = env.HWLAB_API_KEY;
  if (apiKey) headers.authorization = `Bearer ${apiKey}`;
  const frames: SseFrame[] = [];
  let buffer = "";
  try {
    const response = await fetch(`${baseUrl.replace(/\/$/u, "")}${path}`, { signal: controller.signal, headers });
    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 (framesInspectionSatisfied(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" && framesInspectionSatisfied(frames, minEvents, waitFor))) {
      const observed = semanticSummary(business);
      throw codedError("workbench_sse_read_failed", `Workbench SSE read failed: ${error?.message ?? 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 connectedFrame = frames.find((entry) => entry.name === "workbench.connected") ?? null;
  const connected = connectedFrame !== null;
  const connectedContract = connectedFrame ? workbenchConnectedContract(connectedFrame.data) : null;
  const semantics = semanticSummary(business);
  const missingSemantics = waitFor.filter((semantic) => !semantics.observedSemantics.includes(semantic));
  const ok = connected && inspectionSatisfied(business, minEvents, waitFor);
  const allEventTypes = business.map(eventType);
  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, connectedContract, eventCount: business.length, eventTypes: timingDetail === "full" ? allEventTypes : [...new Set(allEventTypes)], eventTypeCounts: countValues(allEventTypes), timeoutMs, minEvents, waitFor, ...semantics, timing: timingSummary(business, timingDetail), 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[];
}
type SseFrame = { name: string; data: Record<string, any>; observedAt: string };
function workbenchConnectedContract(data: Record<string, any>) {
  const refreshReplay = record(data.refreshReplay);
  const counts = record(refreshReplay?.counts);
  const query = record(refreshReplay?.query);
  const queryTiming = record(query?.timing);
  const queryIndex = record(query?.index);
  return {
    deliverySemantics: String(data.deliverySemantics ?? "").trim() || null,
    realtimeSource: String(data.realtimeSource ?? "").trim() || null,
    replay: data.replay === true,
    liveOnly: data.liveOnly === true,
    lossPossible: data.lossPossible === true,
    refreshReplay: refreshReplay ? {
      phase: String(refreshReplay.phase ?? "").trim() || null,
      topic: String(refreshReplay.topic ?? "").trim() || null,
      barrier: Array.isArray(refreshReplay.barrier) ? refreshReplay.barrier : [],
      counts: counts ? {
        matched: Number(counts.matched ?? 0),
        replayed: Number(counts.replayed ?? 0),
        buffered: Number(counts.buffered ?? 0),
        bufferedDelivered: Number(counts.bufferedDelivered ?? 0),
        liveDelivered: Number(counts.liveDelivered ?? 0),
        deduplicated: Number(counts.deduplicated ?? 0)
      } : null,
      query: query ? {
        completionReason: String(query.completionReason ?? "").trim() || null,
        complete: query.complete === true,
        scannedCount: nullableNumber(query.scannedCount),
        parsedCount: nullableNumber(query.parsedCount),
        matchedCount: nullableNumber(query.matchedCount),
        filterRejectedCount: nullableNumber(query.filterRejectedCount),
        keyRejectedCount: nullableNumber(query.keyRejectedCount),
        partitionKeyScoped: query.partitionKeyScoped === true,
        targetPartition: nullableNumber(query.targetPartition),
        index: queryIndex ? {
          hit: queryIndex.hit === true,
          source: String(queryIndex.source ?? "").trim() || null,
          fallbackReason: String(queryIndex.fallbackReason ?? "").trim() || null,
          indexedEventCount: nullableNumber(queryIndex.indexedEventCount),
          indexedSessionCount: nullableNumber(queryIndex.indexedSessionCount),
          incompleteSessionCount: nullableNumber(queryIndex.incompleteSessionCount),
          blocking: false,
          valuesPrinted: false
        } : null,
        timing: queryTiming ? {
          endOffsetSnapshotMs: nullableNumber(queryTiming.endOffsetSnapshotMs),
          groupOffsetsMs: nullableNumber(queryTiming.groupOffsetsMs),
          consumerConnectMs: nullableNumber(queryTiming.consumerConnectMs),
          consumerSubscribeMs: nullableNumber(queryTiming.consumerSubscribeMs),
          scanMs: nullableNumber(queryTiming.scanMs),
          cleanupMs: nullableNumber(queryTiming.cleanupMs),
          indexLookupMs: nullableNumber(queryTiming.indexLookupMs),
          totalMs: nullableNumber(queryTiming.totalMs),
          valuesPrinted: false
        } : null,
        valuesPrinted: false
      } : null
    } : null,
    valuesPrinted: false
  };
}
function businessFrames(frames: SseFrame[]) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); }
function countValues(values: string[]) {
  const counts: Record<string, number> = {};
  for (const value of values) counts[value] = (counts[value] ?? 0) + 1;
  return counts;
}
function nullableNumber(value: unknown) {
  const parsed = Number(value);
  return Number.isFinite(parsed) && parsed >= 0 ? parsed : null;
}
function framesInspectionSatisfied(frames: SseFrame[], minEvents: number, waitFor: EventSemantic[]) {
  return frames.some((entry) => entry.name === "workbench.connected")
    && inspectionSatisfied(businessFrames(frames), minEvents, waitFor);
}
function inspectionSatisfied(business: SseFrame[], 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: SseFrame[]) {
  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 timingSummary(business: SseFrame[], detail: "summary" | "full") {
  const rawTimeline = business.map((frame, index) => {
    const event = record(frame.data?.event) ?? {};
    const sourceAt = timestamp(event.createdAt);
    const producedAt = timestamp(frame.data?.producedAt);
    const observedAt = timestamp(frame.observedAt);
    const semantics = eventSemantics.filter((semantic) => matchesSemantic(event, semantic));
    return {
      index,
      eventType: eventType(frame),
      label: String(event.label ?? "").trim() || null,
      semantics,
      sourceAt,
      producedAt,
      observedAt,
      projectionLagMs: elapsedMs(sourceAt, producedAt),
      deliveryLagMs: elapsedMs(producedAt, observedAt)
    };
  });
  const seen = new Set<string>();
  const timeline = rawTimeline.filter((entry) => {
    const key = [entry.eventType, entry.label, entry.sourceAt].join("\u0000");
    if (seen.has(key)) return false;
    seen.add(key);
    return true;
  }).sort((left, right) => Date.parse(left.sourceAt ?? left.observedAt ?? "") - Date.parse(right.sourceAt ?? right.observedAt ?? ""))
    .map((entry, index) => ({ ...entry, index }));
  const sourceTimeline = timeline.filter((entry) => entry.sourceAt !== null);
  const firstSourceAt = sourceTimeline[0]?.sourceAt ?? null;
  const withDeltas = timeline.map((entry, index) => ({
    ...entry,
    sinceFirstMs: elapsedMs(firstSourceAt, entry.sourceAt),
    sincePreviousMs: elapsedMs(timeline[index - 1]?.sourceAt ?? null, entry.sourceAt)
  }));
  const firstBackendAt = firstSemanticAt(withDeltas, "backend");
  const firstAssistantAt = firstSemanticAt(withDeltas, "assistant");
  const terminalAt = firstSemanticAt(withDeltas, "terminal");
  const finalAt = firstSemanticAt(withDeltas, "final");
  const runnerDispatchedAt = firstLabelAt(withDeltas, "agentrun:backend:runner-dispatch-completed");
  const runnerReadyAt = firstLabelAt(withDeltas, "agentrun:backend:run-claimed");
  const providerStartedAt = firstLabelAt(withDeltas, "agentrun:backend:turn/started");
  const slowestGap = withDeltas
    .filter((entry) => entry.sincePreviousMs !== null)
    .sort((left, right) => (right.sincePreviousMs ?? 0) - (left.sincePreviousMs ?? 0))[0] ?? null;
  return {
    clock: "agentrun-source-event-created-at",
    rawEventCount: rawTimeline.length,
    analyzedEventCount: timeline.length,
    duplicateEventCount: rawTimeline.length - timeline.length,
    milestones: { firstEventAt: firstSourceAt, firstBackendAt, firstAssistantAt, terminalAt, finalAt },
    phasesMs: {
      firstEventToBackend: elapsedMs(firstSourceAt, firstBackendAt),
      admissionAndDispatch: elapsedMs(firstSourceAt, runnerDispatchedAt),
      runnerProvisioning: elapsedMs(runnerDispatchedAt, runnerReadyAt),
      runtimePreparation: elapsedMs(runnerReadyAt, providerStartedAt),
      providerExecution: elapsedMs(providerStartedAt, firstAssistantAt),
      backendToAssistant: elapsedMs(firstBackendAt, firstAssistantAt),
      assistantToTerminal: elapsedMs(firstAssistantAt, terminalAt),
      firstEventToTerminal: elapsedMs(firstSourceAt, terminalAt),
      firstEventToFinal: elapsedMs(firstSourceAt, finalAt)
    },
    slowestGap: slowestGap ? {
      durationMs: slowestGap.sincePreviousMs,
      toIndex: slowestGap.index,
      toEventType: slowestGap.eventType,
      toLabel: slowestGap.label
    } : null,
    ...(detail === "full" ? { timeline: withDeltas } : {}),
    valuesPrinted: false
  };
}
function timingDetailValue(value: unknown): "summary" | "full" {
  const detail = String(value ?? "summary").trim().toLowerCase();
  if (detail === "summary" || detail === "full") return detail;
  throw codedError("workbench_timing_detail_unsupported", "--timing-detail must be summary or full");
}
function firstSemanticAt(timeline: Array<{ semantics: EventSemantic[]; sourceAt: string | null }>, semantic: EventSemantic) {
  return timeline.find((entry) => entry.semantics.includes(semantic) && entry.sourceAt)?.sourceAt ?? null;
}
function firstLabelAt(timeline: Array<{ label: string | null; sourceAt: string | null }>, label: string) {
  return timeline.find((entry) => entry.label === label && entry.sourceAt)?.sourceAt ?? null;
}
function timestamp(value: unknown): string | null {
  const text = String(value ?? "").trim();
  return text && Number.isFinite(Date.parse(text)) ? new Date(Date.parse(text)).toISOString() : null;
}
function elapsedMs(start: string | null, end: string | null): number | null {
  if (!start || !end) return null;
  return Math.max(0, Date.parse(end) - Date.parse(start));
}
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): SseFrame | 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) ?? {}, observedAt: new Date().toISOString() };
}
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 }); }