pikasTech-HWLAB/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts

// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; PJ2026-010401080313 Workbench实时权威 draft-2026-07-08-p0-workbench-realtime-authority-v2.
// Responsibility: Workbench SSE transport lifecycle, cursor ownership, turn-style tick state and recovery actions.
// Mechanical source references:
// - OpenCode stream.transport.ts:1-17 transport/turn coordination intent.
// - OpenCode stream.transport.ts:83-130 Wait/State shape.
// - OpenCode stream.transport.ts:229-247 abort/stale wait race.
// - OpenCode stream.transport.ts:391-443 stream acquire/subscribe.
// - OpenCode stream.transport.ts:504-530 reduce output/trace.

import { connectWorkbenchEvents, type WorkbenchEventStream, type WorkbenchRealtimeEvent, type WorkbenchSseIngressFrame } from "@/api/workbench-events";
import type { WorkbenchRealtimeCapabilities } from "@/config/runtime";
import { workbenchRealtimeScopeKey } from "@/utils/workbench-key";

export type { WorkbenchRealtimeEvent };

export interface WorkbenchStreamTransportRestartInput {
  realtimeCapabilities: WorkbenchRealtimeCapabilities;
  sessionId?: string | null;
  traceId?: string | null;
  afterSeq?: number | null;
  forceReconnect?: boolean;
  errorRecoveryMinMs?: number | null;
  flushMaxItemsPerChunk?: number | null;
  flushMaxChunkMs?: number | null;
  flushYieldMs?: number | null;
  onIngress?: (frame: WorkbenchSseIngressFrame) => void;
  onOpen?: () => void;
  onError?: (event: Event) => void;
  onState?: (state: WorkbenchStreamTransportState) => void;
  onRecovery?: (recovery: WorkbenchStreamTransportRecovery) => void;
  onEvent: (event: WorkbenchRealtimeEvent, eventName: string) => void;
}

export interface WorkbenchStreamTransportRestartResult {
  key: string;
  changed: boolean;
  streamStarted: boolean;
}

export type WorkbenchStreamTransportPhase = "idle" | "connecting" | "open" | "event" | "error" | "closed" | "blocked";

export type WorkbenchStreamTransportRecoveryAction = "events-reconnect";

export interface WorkbenchStreamTransportState {
  key: string;
  sessionId: string | null;
  traceId: string | null;
  phase: WorkbenchStreamTransportPhase;
  tick: number;
  live: boolean;
  outboxSeq: number | null;
  traceSeq: number | null;
  reason?: string | null;
  errorName?: string | null;
  diagnostic?: Record<string, unknown> | null;
}

export interface WorkbenchStreamTransportRecovery {
  key: string;
  sessionId: string | null;
  traceId: string | null;
  tick: number;
  reason: string;
  actions: WorkbenchStreamTransportRecoveryAction[];
  outboxSeq: number | null;
  traceSeq: number | null;
  diagnostic: Record<string, unknown>;
}

interface WorkbenchTransportWait {
  tick: number;
  armed: boolean;
  live: boolean;
}

interface WorkbenchTransportCursor {
  outboxSeq: number | null;
  traceSeq: number | null;
}

export function workbenchRealtimeTraceIdForCapabilities(
  capabilities: WorkbenchRealtimeCapabilities,
  ...candidates: Array<string | null | undefined>
): string | null {
  if (capabilities.liveKafkaSse) return null;
  for (const candidate of candidates) {
    const traceId = typeof candidate === "string" ? candidate.trim() : "";
    if (traceId) return traceId;
  }
  return null;
}

export function workbenchRealtimeTransportEnabled(capabilities: WorkbenchRealtimeCapabilities): boolean {
  return capabilities.liveKafkaSse || capabilities.projectionRealtime;
}

export class WorkbenchStreamTransportRuntime {
  private stream: WorkbenchEventStream | null = null;
  private key = "";
  private generation = 0;
  private tick = 0;
  private wait: WorkbenchTransportWait | null = null;
  private readonly cursorByKey = new Map<string, WorkbenchTransportCursor>();
  private readonly lastRecoveryAtByKey = new Map<string, number>();

  restart(input: WorkbenchStreamTransportRestartInput): WorkbenchStreamTransportRestartResult {
    const key = workbenchRealtimeScopeKey(input.sessionId ?? null, input.traceId ?? null);
    if (key === this.key && this.stream && input.forceReconnect !== true) return { key, changed: false, streamStarted: true };
    this.stop();
    this.key = key;
    const generation = this.generation;
    this.armWait(key);
    if (!workbenchRealtimeTransportEnabled(input.realtimeCapabilities)) return { key, changed: true, streamStarted: false };
    if (!input.sessionId && !input.traceId) return { key, changed: true, streamStarted: false };
    this.emitState(input, "connecting", null);
    this.stream = connectWorkbenchEvents({
      realtimeCapabilities: input.realtimeCapabilities,
      sessionId: input.sessionId ?? null,
      traceId: input.traceId ?? null,
      afterSeq: !input.realtimeCapabilities.liveKafkaSse && input.realtimeCapabilities.projectionRealtime ? input.afterSeq ?? this.cursorByKey.get(key)?.outboxSeq ?? null : null,
      flushMaxItemsPerChunk: input.flushMaxItemsPerChunk,
      flushMaxChunkMs: input.flushMaxChunkMs,
      flushYieldMs: input.flushYieldMs,
      onIngress: input.onIngress,
      onOpen: () => {
        if (!this.isCurrentStream(key, generation)) return;
        this.markLive(key);
        this.emitState(input, "open", null);
        input.onOpen?.();
      },
      onError: (event) => {
        if (!this.isCurrentStream(key, generation)) return;
        this.markBlocked(key);
        this.emitState(input, "error", null, event.type);
        input.onError?.(event);
        this.requestRecovery(input, event.type || "eventsource-error");
      },
      onEvent: (event, eventName) => {
        if (!this.isCurrentStream(key, generation)) return;
        if (!input.realtimeCapabilities.liveKafkaSse && input.realtimeCapabilities.projectionRealtime) this.rememberCursor(key, event);
        this.markLive(key);
        this.emitState(input, "event", eventName);
        input.onEvent(event, eventName);
      }
    });
    return { key, changed: true, streamStarted: Boolean(this.stream) };
  }

  stop(): void {
    const previousKey = this.key;
    this.generation += 1;
    this.stream?.close();
    this.stream = null;
    this.key = "";
    if (previousKey) this.wait = null;
  }

  currentKey(): string {
    return this.key;
  }

  currentCursor(key: string = this.key): WorkbenchTransportCursor {
    return this.cursorByKey.get(key) ?? { outboxSeq: null, traceSeq: null };
  }

  private isCurrentStream(key: string, generation: number): boolean {
    return this.stream !== null && this.key === key && this.generation === generation;
  }

  clear(key?: string | null): void {
    const normalized = normalizeKey(key);
    if (normalized) {
      this.cursorByKey.delete(normalized);
      this.lastRecoveryAtByKey.delete(normalized);
      if (this.key === normalized) this.stop();
      return;
    }
    this.stop();
    this.cursorByKey.clear();
    this.lastRecoveryAtByKey.clear();
  }

  private armWait(key: string): void {
    this.tick += 1;
    this.wait = { tick: this.tick, armed: Boolean(key), live: false };
  }

  private markLive(key: string): void {
    if (!this.wait || this.key !== key) return;
    this.wait = { ...this.wait, live: true };
  }

  private markBlocked(key: string): void {
    if (!this.wait || this.key !== key) return;
    this.wait = { ...this.wait, live: false };
  }

  private rememberCursor(key: string, event: WorkbenchRealtimeEvent): void {
    const previous = this.cursorByKey.get(key) ?? { outboxSeq: null, traceSeq: null };
    const outboxSeq = maxCursor(previous.outboxSeq, numericCursor(event.cursor?.outboxSeq ?? event.outboxSeq));
    const traceSeq = maxCursor(previous.traceSeq, numericCursor(event.cursor?.traceSeq ?? event.traceSeq ?? event.event?.seq ?? event.event?.projectedSeq));
    this.cursorByKey.set(key, { outboxSeq, traceSeq });
  }

  private requestRecovery(input: WorkbenchStreamTransportRestartInput, reason: string): void {
    const key = this.key;
    if (!key) return;
    const now = Date.now();
    const minMs = Math.max(0, Math.trunc(input.errorRecoveryMinMs ?? 0));
    const lastAt = this.lastRecoveryAtByKey.get(key) ?? 0;
    if (lastAt > 0 && now - lastAt < minMs) return;
    this.lastRecoveryAtByKey.set(key, now);
    const sessionId = input.sessionId ?? null;
    const traceId = input.traceId ?? null;
    const cursor = this.currentCursor(key);
    const actions: WorkbenchStreamTransportRecoveryAction[] = sessionId || traceId ? ["events-reconnect"] : [];
    const diagnostic = this.diagnosticEnvelope("workbench_sse_recovery", reason, key, actions);
    input.onRecovery?.({ key, sessionId, traceId, tick: this.wait?.tick ?? this.tick, reason, actions, outboxSeq: cursor.outboxSeq, traceSeq: cursor.traceSeq, diagnostic });
  }

  private emitState(input: WorkbenchStreamTransportRestartInput, phase: WorkbenchStreamTransportPhase, reason: string | null, errorName: string | null = null): void {
    if (!input.onState) return;
    const key = this.key || workbenchRealtimeScopeKey(input.sessionId ?? null, input.traceId ?? null);
    const cursor = this.currentCursor(key);
    input.onState({
      key,
      sessionId: input.sessionId ?? null,
      traceId: input.traceId ?? null,
      phase,
      tick: this.wait?.tick ?? this.tick,
      live: this.wait?.live === true,
      outboxSeq: cursor.outboxSeq,
      traceSeq: cursor.traceSeq,
      reason,
      errorName,
      diagnostic: phase === "error" || phase === "blocked" ? this.diagnosticEnvelope("workbench_sse_transport_error", errorName ?? reason ?? phase, key, []) : null
    });
  }

  private diagnosticEnvelope(code: string, reason: string, key: string, actions: WorkbenchStreamTransportRecoveryAction[]): Record<string, unknown> {
    return {
      contractVersion: "hwlab-error-diagnostic-v1",
      code,
      category: "transport",
      source: "workbench-web",
      layer: "workbench-stream-transport",
      rootCause: reason,
      recoveryAction: actions.join(",") || "observe-transport-state",
      transportState: this.wait?.live === true ? "live" : "degraded",
      scopedKey: key,
      tick: this.wait?.tick ?? this.tick,
      retryable: true,
      valuesPrinted: false,
      valuesRedacted: true
    };
  }
}

export function createWorkbenchStreamTransportRuntime(): WorkbenchStreamTransportRuntime {
  return new WorkbenchStreamTransportRuntime();
}

function normalizeKey(value: string | null | undefined): string | null {
  const text = String(value ?? "").trim();
  return text || null;
}

function numericCursor(value: unknown): number | null {
  const parsed = Number(value);
  return Number.isFinite(parsed) && parsed >= 0 ? Math.trunc(parsed) : null;
}

function maxCursor(previous: number | null, next: number | null): number | null {
  if (next === null) return previous;
  return previous === null ? next : Math.max(previous, next);
}