pikasTech-HWLAB/web/hwlab-cloud-web/src/stores/agent-observer.ts

import { computed, ref } from "vue";
import { defineStore } from "pinia";
import { openAgentObserverStream } from "@/api/agentObserver";
import { agentObserverRuntimeConfig, type AgentObserverRuntimeConfig } from "@/config/agent-observer";
import { asRecord, nonEmptyString } from "@/utils";

export type AgentObserverView = "table" | "cards" | "tree";
export type AgentObserverDensity = "compact" | "standard";
export type AgentObserverPhase = "idle" | "replay" | "live" | "reconnecting" | "stale" | "error";

export interface AgentObserverEvent {
  id: string;
  sourceEventId: string | null;
  runId: string;
  commandId: string | null;
  sessionId: string | null;
  traceId: string | null;
  sourceSeq: number | null;
  createdAt: string | null;
  type: string;
  status: string;
  label: string;
  message: string;
  terminal: boolean;
  raw: Record<string, unknown>;
}

export interface AgentObserverRun {
  id: string;
  taskId: string | null;
  commandId: string | null;
  attemptId: string | null;
  runnerId: string | null;
  sessionId: string | null;
  traceId: string | null;
  projectId: string | null;
  workspace: string | null;
  backend: string | null;
  providerId: string | null;
  runStatus: string;
  commandState: string | null;
  commandType: string | null;
  phase: string;
  result: string | null;
  errorCode: string | null;
  lastMessage: string | null;
  startedAt: string | null;
  updatedAt: string | null;
  lastSourceSeq: number | null;
  eventIds: string[];
  changedAtMs: number;
}

export type AgentObserverRelationKind = "scope" | "session" | "task" | "run" | "attempt" | "command";

export interface AgentObserverRelationRow {
  id: string;
  kind: AgentObserverRelationKind;
  label: string;
  detail: string | null;
  parentId: string | null;
  runId: string | null;
  status: string | null;
  depth: number;
  childCount: number;
  expanded: boolean;
  position: number;
  setSize: number;
}

interface AgentObserverRelationNode {
  id: string;
  kind: AgentObserverRelationKind;
  label: string;
  detail: string | null;
  parentId: string | null;
  runId: string | null;
  status: string | null;
  children: string[];
}

export const useAgentObserverStore = defineStore("agentObserver", () => {
  const entities = ref<Record<string, AgentObserverRun>>({});
  const orderedIds = ref<string[]>([]);
  const events = ref<AgentObserverEvent[]>([]);
  const selectedRunId = ref<string | null>(null);
  const view = ref<AgentObserverView>("table");
  const density = ref<AgentObserverDensity>("compact");
  const search = ref("");
  const statusFilter = ref("all");
  const projectFilter = ref("all");
  const agentFilter = ref("all");
  const timeWindow = ref("all");
  const sort = ref("updated-desc");
  const phase = ref<AgentObserverPhase>("idle");
  const lastEventAt = ref<string | null>(null);
  const connectedAt = ref<string | null>(null);
  const error = ref<string | null>(null);
  const nowMs = ref(Date.now());
  const config = ref<AgentObserverRuntimeConfig | null>(null);
  const expandedRelationIds = ref<string[]>([]);
  const collapsedRelationIds = ref<string[]>([]);
  const seenEventIds = new Set<string>();
  const queue: unknown[] = [];
  let source: EventSource | null = null;
  let reconnectTimer: number | null = null;
  let staleTimer: number | null = null;
  let flushScheduled = false;
  let manuallyClosed = false;

  const runs = computed(() => orderedIds.value.map((id) => entities.value[id]).filter((item): item is AgentObserverRun => Boolean(item)));
  const selectedRun = computed(() => selectedRunId.value ? entities.value[selectedRunId.value] ?? null : null);
  const visibleRuns = computed(() => {
    const needle = search.value.trim().toLowerCase();
    const result = runs.value.filter((run) => {
      if (statusFilter.value !== "all" && statusGroup(run.runStatus) !== statusFilter.value) return false;
      if (projectFilter.value !== "all" && !matchesScopeFilter(run, projectFilter.value)) return false;
      if (agentFilter.value !== "all" && (run.backend ?? run.providerId) !== agentFilter.value) return false;
      if (!insideTimeWindow(run.updatedAt ?? run.startedAt, timeWindow.value)) return false;
      if (!needle) return true;
      return [run.id, run.taskId, run.commandId, run.sessionId, run.traceId, run.projectId, run.workspace, run.backend, run.providerId, run.errorCode, run.lastMessage]
        .some((value) => value?.toLowerCase().includes(needle));
    });
    return result.sort((left, right) => compareRuns(left, right, sort.value));
  });
  const projects = computed(() => unique(runs.value.map((run) => run.projectId)));
  const scopeOptions = computed(() => [
    ...unique(runs.value.map((run) => run.projectId)).map((value) => ({ value: `project:${value}`, label: `项目 · ${value}` })),
    ...unique(runs.value.map((run) => run.workspace)).map((value) => ({ value: `workspace:${value}`, label: `Workspace · ${value}` }))
  ]);
  const agents = computed(() => unique(runs.value.map((run) => run.backend ?? run.providerId)));
  const selectedEvents = computed(() => {
    const run = selectedRun.value;
    if (!run || !config.value) return [];
    const ids = new Set(run.eventIds.slice(-config.value.inspectorEventTailLimit));
    return events.value.filter((event) => ids.has(event.id));
  });
  const stats = computed(() => {
    const values = visibleRuns.value;
    return {
      total: values.length,
      running: values.filter((run) => statusGroup(run.runStatus) === "running").length,
      waiting: values.filter((run) => statusGroup(run.runStatus) === "waiting").length,
      failed: values.filter((run) => statusGroup(run.runStatus) === "failed").length,
      terminal: values.filter((run) => statusGroup(run.runStatus) === "terminal").length,
      runners: new Set(values.map((run) => run.runnerId).filter(Boolean)).size
    };
  });
  const dataAgeMs = computed(() => lastEventAt.value ? Math.max(0, nowMs.value - Date.parse(lastEventAt.value)) : null);
  const relationRows = computed(() => buildRelationRows(
    visibleRuns.value,
    selectedRunId.value,
    new Set(expandedRelationIds.value),
    new Set(collapsedRelationIds.value)
  ));

  function connect(): void {
    disconnect(false);
    config.value = agentObserverRuntimeConfig();
    if (!config.value) {
      phase.value = "error";
      error.value = "Agent observer runtime config 未由 owning YAML 渲染。";
      return;
    }
    manuallyClosed = false;
    phase.value = entities.value && orderedIds.value.length > 0 ? "reconnecting" : "replay";
    error.value = null;
    source = openAgentObserverStream({
      onEnvelope: enqueueEnvelope,
      onConnected: () => {
        connectedAt.value = new Date().toISOString();
        phase.value = "live";
        error.value = null;
      },
      onHeartbeat: () => {
        if (phase.value === "stale") phase.value = "live";
      },
      onStreamError: (payload) => {
        const record = asRecord(payload);
        error.value = nonEmptyString(record?.message) ?? "Kafka observer stream failed.";
        phase.value = "error";
      },
      onTransportError: scheduleReconnect
    });
    if (staleTimer === null) {
      staleTimer = window.setInterval(() => {
        nowMs.value = Date.now();
        if (phase.value === "live" && dataAgeMs.value !== null && dataAgeMs.value > (config.value?.staleAfterMs ?? Number.MAX_SAFE_INTEGER)) phase.value = "stale";
      }, 1000);
    }
  }

  function disconnect(manual = true): void {
    manuallyClosed = manual;
    source?.close();
    source = null;
    if (reconnectTimer !== null) window.clearTimeout(reconnectTimer);
    reconnectTimer = null;
    if (manual && staleTimer !== null) window.clearInterval(staleTimer);
    if (manual) staleTimer = null;
  }

  function reconnect(): void {
    manuallyClosed = false;
    connect();
  }

  function scheduleReconnect(): void {
    source?.close();
    source = null;
    if (manuallyClosed || reconnectTimer !== null || !config.value) return;
    phase.value = "reconnecting";
    reconnectTimer = window.setTimeout(() => {
      reconnectTimer = null;
      connect();
    }, config.value.reconnectDelayMs);
  }

  function enqueueEnvelope(payload: unknown): void {
    if (!config.value) return;
    if (queue.length >= config.value.liveBufferLimit) {
      phase.value = "error";
      error.value = "客户端 Kafka 事件缓冲达到 YAML 上限。";
      source?.close();
      return;
    }
    queue.push(payload);
    if (flushScheduled) return;
    flushScheduled = true;
    queueMicrotask(flushQueue);
  }

  function flushQueue(): void {
    flushScheduled = false;
    if (!config.value) return;
    const batch = queue.splice(0, config.value.batchMaxEvents);
    for (const payload of batch) reduceEnvelope(payload);
    if (queue.length > 0) {
      flushScheduled = true;
      queueMicrotask(flushQueue);
    }
  }

  function reduceEnvelope(payload: unknown): void {
    if (!config.value) return;
    const envelope = asRecord(payload);
    const body = asRecord(envelope?.event);
    const context = asRecord(envelope?.context);
    const runId = nonEmptyString(envelope?.runId) ?? nonEmptyString(body?.runId);
    const eventId = nonEmptyString(envelope?.eventId) ?? nonEmptyString(envelope?.sourceEventId);
    if (!runId || !eventId || seenEventIds.has(eventId)) return;
    seenEventIds.add(eventId);
    const createdAt = nonEmptyString(body?.createdAt) ?? nonEmptyString(envelope?.producedAt);
    const sourceSeq = integer(body?.sourceSeq) ?? integer(context?.sourceSeq);
    const event: AgentObserverEvent = {
      id: eventId,
      sourceEventId: nonEmptyString(envelope?.sourceEventId),
      runId,
      commandId: nonEmptyString(envelope?.commandId) ?? nonEmptyString(body?.commandId),
      sessionId: nonEmptyString(envelope?.sessionId) ?? nonEmptyString(envelope?.hwlabSessionId),
      traceId: nonEmptyString(envelope?.traceId),
      sourceSeq,
      createdAt,
      type: nonEmptyString(body?.type) ?? "event",
      status: nonEmptyString(body?.status) ?? "running",
      label: nonEmptyString(body?.label) ?? "agentrun:event",
      message: nonEmptyString(body?.message) ?? nonEmptyString(body?.text) ?? "",
      terminal: body?.terminal === true,
      raw: body ?? {}
    };
    events.value.push(event);
    while (events.value.length > config.value.reducerEventLimit) {
      events.value.shift();
    }

    const previous = entities.value[runId];
    const sequenceIsCurrent = !previous || previous.lastSourceSeq === null || (sourceSeq !== null && sourceSeq >= previous.lastSourceSeq);
    lastEventAt.value = latestTimestamp(lastEventAt.value, createdAt) ?? new Date().toISOString();
    nowMs.value = Date.now();
    if (previous && !sequenceIsCurrent) return;
    const terminalStatus = nonEmptyString(body?.terminalStatus);
    const runStatus = normalizeRunStatus(nonEmptyString(body?.runStatus), terminalStatus, event.status, event.terminal, previous?.runStatus);
    const next: AgentObserverRun = {
      id: runId,
      taskId: nonEmptyString(body?.taskId) ?? previous?.taskId ?? null,
      commandId: event.commandId ?? previous?.commandId ?? null,
      attemptId: nonEmptyString(body?.attemptId) ?? previous?.attemptId ?? null,
      runnerId: nonEmptyString(body?.runnerId) ?? previous?.runnerId ?? null,
      sessionId: event.sessionId ?? previous?.sessionId ?? null,
      traceId: event.traceId ?? previous?.traceId ?? null,
      projectId: nonEmptyString(body?.projectId) ?? nonEmptyString(context?.projectId) ?? previous?.projectId ?? null,
      workspace: nonEmptyString(body?.workspace) ?? previous?.workspace ?? null,
      backend: nonEmptyString(body?.backend) ?? previous?.backend ?? null,
      providerId: nonEmptyString(body?.providerId) ?? previous?.providerId ?? null,
      runStatus,
      commandState: nonEmptyString(body?.commandState) ?? previous?.commandState ?? null,
      commandType: nonEmptyString(body?.commandType) ?? previous?.commandType ?? null,
      phase: event.label.replace(/^agentrun:/u, ""),
      result: event.terminal ? (event.message || terminalStatus) : (previous?.result ?? null),
      errorCode: nonEmptyString(body?.errorCode) ?? nonEmptyString(body?.failureKind) ?? previous?.errorCode ?? null,
      lastMessage: event.message || (previous?.lastMessage ?? null),
      startedAt: previous?.startedAt ?? createdAt,
      updatedAt: createdAt ?? previous?.updatedAt ?? null,
      lastSourceSeq: sourceSeq ?? previous?.lastSourceSeq ?? null,
      eventIds: [...(previous?.eventIds ?? []), eventId].slice(-config.value.inspectorEventTailLimit),
      changedAtMs: Date.now()
    };
    entities.value = { ...entities.value, [runId]: next };
    orderedIds.value = [runId, ...orderedIds.value.filter((id) => id !== runId)];
    trimEntities();
  }

  function trimEntities(): void {
    if (!config.value || orderedIds.value.length <= config.value.reducerEntityLimit) return;
    const keep = orderedIds.value.slice(0, config.value.reducerEntityLimit);
    if (selectedRunId.value && !keep.includes(selectedRunId.value) && entities.value[selectedRunId.value]) keep[keep.length - 1] = selectedRunId.value;
    const next: Record<string, AgentObserverRun> = {};
    for (const id of keep) if (entities.value[id]) next[id] = entities.value[id];
    entities.value = next;
    orderedIds.value = keep;
  }

  function selectRun(runId: string | null): void {
    selectedRunId.value = runId;
  }

  function toggleRelation(relationId: string, expanded: boolean): void {
    if (expanded) {
      collapsedRelationIds.value = collapsedRelationIds.value.filter((id) => id !== relationId);
      if (!expandedRelationIds.value.includes(relationId)) expandedRelationIds.value = [...expandedRelationIds.value, relationId];
      return;
    }
    expandedRelationIds.value = expandedRelationIds.value.filter((id) => id !== relationId);
    if (!collapsedRelationIds.value.includes(relationId)) collapsedRelationIds.value = [...collapsedRelationIds.value, relationId];
  }

  return {
    entities, orderedIds, events, runs, visibleRuns, selectedRun, selectedEvents, selectedRunId,
    view, density, search, statusFilter, projectFilter, agentFilter, timeWindow, sort, phase, lastEventAt,
    connectedAt, error, config, projects, scopeOptions, agents, stats, dataAgeMs, relationRows,
    connect, disconnect, reconnect, selectRun, toggleRelation
  };
});

function buildRelationRows(
  runs: AgentObserverRun[],
  selectedRunId: string | null,
  expandedIds: Set<string>,
  collapsedIds: Set<string>
): AgentObserverRelationRow[] {
  const nodes = new Map<string, AgentObserverRelationNode>();
  const rootIds: string[] = [];
  const selectedPath = new Set<string>();
  const activePath = new Set<string>();
  const unassociatedId = "scope:unassociated";

  const ensureNode = (node: Omit<AgentObserverRelationNode, "children">): AgentObserverRelationNode => {
    const existing = nodes.get(node.id);
    if (existing) return existing;
    const created = { ...node, children: [] };
    nodes.set(node.id, created);
    if (node.parentId) {
      const parent = nodes.get(node.parentId);
      if (parent && !parent.children.includes(node.id)) parent.children.push(node.id);
    } else if (!rootIds.includes(node.id)) rootIds.push(node.id);
    return created;
  };

  const ensureUnassociated = (): AgentObserverRelationNode => ensureNode({
    id: unassociatedId,
    kind: "scope",
    label: "未关联",
    detail: "Kafka 事件未提供完整 Project / Workspace → Session → Task lineage",
    parentId: null,
    runId: null,
    status: null
  });

  for (const run of runs) {
    const hasCompleteContainment = Boolean((run.projectId || run.workspace) && run.sessionId && run.taskId);
    let parent = hasCompleteContainment
      ? ensureNode({
        id: relationId("scope", `${run.projectId ?? ""}\u0000${run.workspace ?? ""}`),
        kind: "scope",
        label: `${run.projectId ?? "项目未提供"} / ${run.workspace ?? "Workspace 未提供"}`,
        detail: null,
        parentId: null,
        runId: null,
        status: null
      })
      : ensureUnassociated();
    const path = [parent.id];

    if (hasCompleteContainment && run.sessionId && run.taskId) {
      parent = ensureNode({
        id: relationId("session", `${parent.id}\u0000${run.sessionId}`),
        kind: "session",
        label: `Session ${run.sessionId}`,
        detail: run.traceId ? `Trace ${run.traceId}` : null,
        parentId: parent.id,
        runId: null,
        status: null
      });
      path.push(parent.id);
      parent = ensureNode({
        id: relationId("task", `${parent.id}\u0000${run.taskId}`),
        kind: "task",
        label: `Task ${run.taskId}`,
        detail: null,
        parentId: parent.id,
        runId: null,
        status: null
      });
      path.push(parent.id);
    }

    parent = ensureNode({
      id: relationId("run", `${parent.id}\u0000${run.id}`),
      kind: "run",
      label: `Run ${run.id}`,
      detail: run.phase,
      parentId: parent.id,
      runId: run.id,
      status: run.runStatus
    });
    path.push(parent.id);

    if (run.attemptId) {
      parent = ensureNode({
        id: relationId("attempt", `${parent.id}\u0000${run.attemptId}`),
        kind: "attempt",
        label: `Attempt ${run.attemptId}`,
        detail: null,
        parentId: parent.id,
        runId: run.id,
        status: run.runStatus
      });
      path.push(parent.id);
    }

    if (run.commandId) {
      parent = ensureNode({
        id: relationId("command", `${parent.id}\u0000${run.commandId}`),
        kind: "command",
        label: `Command ${run.commandId}`,
        detail: [run.commandType, run.commandState].filter(Boolean).join(" · ") || null,
        parentId: parent.id,
        runId: run.id,
        status: run.commandState ?? run.runStatus
      });
      path.push(parent.id);
    }

    if (run.id === selectedRunId) for (const id of path) selectedPath.add(id);
    if (["running", "waiting"].includes(statusGroup(run.runStatus))) for (const id of path) activePath.add(id);
  }

  const rows: AgentObserverRelationRow[] = [];
  const append = (id: string, depth: number, position: number, setSize: number): void => {
    const node = nodes.get(id);
    if (!node) return;
    const defaultExpanded = depth === 0 || selectedPath.has(id) || activePath.has(id);
    const expanded = node.children.length > 0 && !collapsedIds.has(id) && (expandedIds.has(id) || defaultExpanded);
    rows.push({
      id: node.id,
      kind: node.kind,
      label: node.label,
      detail: node.detail,
      parentId: node.parentId,
      runId: node.runId,
      status: node.status,
      depth,
      childCount: node.children.length,
      expanded,
      position,
      setSize
    });
    if (expanded) node.children.forEach((childId, index) => append(childId, depth + 1, index + 1, node.children.length));
  };
  rootIds.forEach((id, index) => append(id, 0, index + 1, rootIds.length));
  return rows;
}

function relationId(kind: AgentObserverRelationKind, value: string): string {
  return `${kind}:${encodeURIComponent(value)}`;
}

function latestTimestamp(previous: string | null, next: string | null): string | null {
  if (!previous) return next;
  if (!next) return previous;
  const previousMs = Date.parse(previous);
  const nextMs = Date.parse(next);
  if (!Number.isFinite(previousMs)) return next;
  if (!Number.isFinite(nextMs)) return previous;
  return nextMs >= previousMs ? next : previous;
}

function normalizeRunStatus(runStatus: string | null, terminalStatus: string | null, eventStatus: string, terminal: boolean, previous = "unknown"): string {
  if (terminalStatus) return terminalStatus === "cancelled" ? "canceled" : terminalStatus;
  if (terminal) return eventStatus;
  if (["completed", "failed", "canceled", "cancelled"].includes(previous)) return previous;
  return runStatus ?? eventStatus ?? previous;
}

function statusGroup(status: string): string {
  const value = status.toLowerCase();
  if (["failed", "error", "rejected"].includes(value)) return "failed";
  if (["completed", "succeeded", "canceled", "cancelled"].includes(value)) return "terminal";
  if (["queued", "pending", "waiting", "admitted"].includes(value)) return "waiting";
  return "running";
}

function compareRuns(left: AgentObserverRun, right: AgentObserverRun, sort: string): number {
  if (sort === "id-asc") return left.id.localeCompare(right.id);
  if (sort === "status-asc") return left.runStatus.localeCompare(right.runStatus) || left.id.localeCompare(right.id);
  const leftTime = Date.parse(left.updatedAt ?? left.startedAt ?? "") || 0;
  const rightTime = Date.parse(right.updatedAt ?? right.startedAt ?? "") || 0;
  return sort === "updated-asc" ? leftTime - rightTime : rightTime - leftTime;
}

function unique(values: Array<string | null>): string[] {
  return [...new Set(values.filter((value): value is string => Boolean(value)))].sort();
}

function matchesScopeFilter(run: AgentObserverRun, filter: string): boolean {
  if (filter.startsWith("project:")) return run.projectId === filter.slice("project:".length);
  if (filter.startsWith("workspace:")) return run.workspace === filter.slice("workspace:".length);
  return run.projectId === filter;
}

function insideTimeWindow(value: string | null, window: string): boolean {
  if (window === "all") return true;
  const duration = window === "15m" ? 15 * 60_000 : window === "1h" ? 60 * 60_000 : window === "6h" ? 6 * 60 * 60_000 : 0;
  if (!duration || !value) return false;
  const timestamp = Date.parse(value);
  return Number.isFinite(timestamp) && timestamp >= Date.now() - duration;
}

function integer(value: unknown): number | null {
  const number = Number(value);
  return Number.isInteger(number) ? number : null;
}