pikasTech-HWLAB/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts

// SPEC: PJ2026-0104010803 isolated Kafka debug fixture.
// Responsibility: project historical hwlab.event.debug.v1 fixtures for the admin-only isolated debugger; never imported by the product Workbench SSE path.

import { mergeRunnerTrace } from "../composables/workbench-trace-snapshot";
import type { ChatMessage, TraceEvent, WorkbenchTurnTimingProjection } from "../types";
import { firstNonEmptyString } from "../utils";
import { traceAssistantLifecycleEventIsSuperseded } from "../../../../tools/src/hwlab-cli/trace-renderer.ts";

export interface WorkbenchLiveMessageState {
  text: string;
  status: string;
  terminal: boolean;
}

export interface WorkbenchLiveKafkaProjectionInput {
  previous: ChatMessage | null;
  traceId: string;
  sessionId: string;
  event: TraceEvent;
  receivedAt: string;
  title?: string;
}

export interface WorkbenchLiveKafkaUserProjectionInput {
  previous: ChatMessage | null;
  traceId: string;
  sessionId: string;
  event: TraceEvent;
  receivedAt: string;
}

export function reduceWorkbenchLiveKafkaMessageState(previous: WorkbenchLiveMessageState, event: TraceEvent, previousEvents: TraceEvent[] = []): WorkbenchLiveMessageState {
  if (previous.terminal) return previous;
  const terminal = workbenchLiveKafkaEventIsTerminal(event);
  const status = terminal ? firstNonEmptyString(event.status, "completed") ?? "completed" : "running";
  const lifecycleEvents = [...previousEvents, event];
  const assistantText = traceAssistantLifecycleEventIsSuperseded(event, lifecycleEvents, lifecycleEvents.length - 1)
    ? null
    : workbenchLiveKafkaAssistantText(event);
  return {
    text: assistantText ?? (terminal ? workbenchLiveKafkaTerminalText(event, previous.text, status) : previous.text),
    status,
    terminal
  };
}

export function workbenchUserMessageIdForTrace(traceId: string): string {
  return workbenchMessageIdForTrace(traceId, "user");
}

export function workbenchAgentMessageIdForTrace(traceId: string): string {
  return workbenchMessageIdForTrace(traceId, "agent");
}

function workbenchMessageIdForTrace(traceId: string, role: "user" | "agent"): string {
  const suffix = firstNonEmptyString(traceId)
    ?.replace(/^trc_/u, "")
    .replace(/[^A-Za-z0-9_.:-]/gu, "_")
    .slice(0, 48);
  if (!suffix) throw new Error("traceId must produce a stable Workbench message identity");
  return `msg_${suffix}_${role}`;
}

export function workbenchLiveKafkaEnvelope(schema: unknown): boolean {
  return firstNonEmptyString(schema) === "hwlab.event.v1";
}

export function workbenchLiveKafkaEventIsUserMessage(event: TraceEvent | null | undefined): boolean {
  return firstNonEmptyString(event?.type, event?.eventType) === "user";
}

export function workbenchLiveKafkaProjectionTarget(event: TraceEvent | null | undefined): "user" | "agent" {
  return workbenchLiveKafkaEventIsUserMessage(event) ? "user" : "agent";
}

export function projectWorkbenchLiveKafkaUserMessage(input: WorkbenchLiveKafkaUserProjectionInput): ChatMessage | null {
  if (!workbenchLiveKafkaEventIsUserMessage(input.event)) return null;
  const messageId = firstNonEmptyString(input.event.userMessageId, input.event.messageId);
  const text = firstNonEmptyString(input.event.text, input.event.message);
  const receivedAt = timestamp(input.receivedAt);
  const createdAt = timestamp(input.event.createdAt) ?? receivedAt;
  if (!messageId || !text || !receivedAt || !createdAt) return null;
  return {
    ...(input.previous ?? {}),
    id: messageId,
    messageId,
    role: "user",
    title: "用户",
    text,
    status: "sent",
    traceId: input.traceId,
    targetTraceId: firstNonEmptyString(input.event.targetTraceId),
    turnId: input.previous?.turnId ?? input.traceId,
    sessionId: input.sessionId,
    createdAt: input.previous?.createdAt ?? createdAt,
    updatedAt: receivedAt
  } as ChatMessage;
}

export function workbenchLiveKafkaEventIsTerminal(event: TraceEvent | null | undefined): boolean {
  return Boolean(event?.terminal === true || firstNonEmptyString(event?.eventType) === "terminal" || firstNonEmptyString(event?.type) === "result");
}

export function workbenchLiveKafkaAssistantText(event: TraceEvent | null | undefined): string | null {
  if (!event || !["assistant", "assistant_progress"].includes(firstNonEmptyString(event.type, event.eventType) ?? "")) return null;
  const finalResponse = recordValue(event.finalResponse);
  return firstNonEmptyString(finalResponse?.text, event.assistantText, event.text, event.message) ?? null;
}

export function workbenchLiveKafkaTerminalText(event: TraceEvent, previousText: string, status: string): string {
  const finalResponse = recordValue(event.finalResponse);
  const error = recordValue(event.error);
  const failed = ["failed", "blocked", "timeout", "canceled", "cancelled", "interrupted", "expired"].includes(status);
  const text = failed
    ? firstNonEmptyString(finalResponse?.text, event.message, error?.message, event.failureKind, event.errorCode, previousText)
    : firstNonEmptyString(finalResponse?.text, event.assistantText, event.text, previousText, event.message);
  return text ?? `AgentRun ${status}`;
}

export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjectionInput): ChatMessage {
  const receivedAt = timestamp(input.receivedAt);
  if (!receivedAt) throw new Error("receivedAt must be a valid timestamp for live Kafka projection");
  const previous = input.previous;
  const liveState = reduceWorkbenchLiveKafkaMessageState({
    text: previous?.text ?? "",
    status: previous?.status ?? "running",
    terminal: previous?.traceAutoLifecycle === "terminal"
  }, input.event, Array.isArray(previous?.runnerTrace?.events) ? previous.runnerTrace.events : []);
  const startedAt = timestamp(previous?.timing?.startedAt)
    ?? timestamp(previous?.startedAt)
    ?? timestamp(input.event.createdAt)
    ?? receivedAt;
  const previousFinishedAt = timestamp(previous?.timing?.finishedAt) ?? timestamp(previous?.finishedAt);
  const finishedAt = liveState.terminal ? previousFinishedAt ?? receivedAt : null;
  const previousDurationMs = nonNegativeNumber(previous?.timing?.durationMs) ?? nonNegativeNumber(previous?.durationMs);
  const durationMs = liveState.terminal
    ? previousFinishedAt && previousDurationMs !== null
      ? previousDurationMs
      : Math.max(0, Date.parse(finishedAt ?? receivedAt) - Date.parse(startedAt))
    : null;
  const timing: WorkbenchTurnTimingProjection = {
    startedAt,
    lastEventAt: receivedAt,
    finishedAt,
    durationMs,
    observedAt: receivedAt,
    lastEventAgeMs: 0,
    valuesRedacted: true
  };
  const mergedRunnerTrace = mergeRunnerTrace(previous?.runnerTrace ?? null, {
    traceId: input.traceId,
    sessionId: input.sessionId,
    status: liveState.status,
    traceStatus: liveState.status,
    events: [input.event],
    eventCount: 1,
    eventSource: "hwlab-kafka-sse",
    timing,
    startedAt,
    lastEventAt: receivedAt,
    finishedAt,
    durationMs,
    updatedAt: receivedAt
  });
  const runnerTrace = {
    ...mergedRunnerTrace,
    status: liveState.status,
    traceStatus: liveState.status,
    timing,
    startedAt,
    lastEventAt: receivedAt,
    finishedAt,
    durationMs,
    updatedAt: receivedAt
  };
  const messageId = workbenchAgentMessageIdForTrace(input.traceId);
  return {
    ...(previous ?? {}),
    id: messageId,
    messageId,
    role: "agent",
    title: input.title ?? previous?.title ?? "Code Agent",
    text: liveState.text,
    status: liveState.status as ChatMessage["status"],
    traceId: input.traceId,
    turnId: previous?.turnId ?? input.traceId,
    sessionId: input.sessionId,
    createdAt: previous?.createdAt ?? startedAt,
    updatedAt: receivedAt,
    startedAt,
    lastEventAt: receivedAt,
    finishedAt,
    durationMs,
    runnerTrace,
    traceAutoLifecycle: liveState.terminal ? "terminal" : "running",
    timing,
    finalResponse: liveState.terminal && liveState.text
      ? { text: liveState.text, status: liveState.status, traceId: input.traceId }
      : null
  } as ChatMessage;
}

function timestamp(value: unknown): string | null {
  const text = firstNonEmptyString(value);
  return text && Number.isFinite(Date.parse(text)) ? text : null;
}

function nonNegativeNumber(value: unknown): number | null {
  return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : null;
}

function recordValue(value: unknown): Record<string, unknown> | null {
  return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : null;
}