pikasTech-HWLAB/internal/cloud/workbench-facts-store.ts

/*
 * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; draft-2026-06-20-p1-zero-split-durable-realtime; PJ2026-010403 API契约 draft-2026-06-18-r1.
 * 职责: Workbench durable facts store adapter。统一读取 session/message/turn/trace projection facts，不在 GET 中推进 AgentRun facts。
 */
import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import { safeConversationId, safeSessionId, safeTraceId } from "./server-http-utils.ts";
import { durableTraceStatus, normalizeWorkbenchStatus, TERMINAL_STATUSES, traceTerminalEvidence } from "./workbench-turn-projection.ts";

const TERMINAL_DURABLE_TRACE_CACHE_MAX = 256;
const terminalDurableTraceCache = new Map();

export function createWorkbenchFactsStore(options = {}, actor = null) {
  const accessStore = options.accessController?.store ?? options.accessController ?? null;
  const traceSessionCache = options.codeAgentTraceSessionCache instanceof Map ? options.codeAgentTraceSessionCache : null;
  const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
  const runtimeStore = options.runtimeStore ?? null;
  const results = options.codeAgentChatResults ?? null;

  async function queryFacts(params = {}) {
    if (typeof runtimeStore?.queryWorkbenchFacts === "function") {
      try {
        const result = await runtimeStore.queryWorkbenchFacts(params);
        return {
          facts: normalizeWorkbenchFactsResult(result?.facts),
          count: Number.isFinite(Number(result?.count)) ? Number(result.count) : 0,
          durable: true,
          persistence: result?.persistence ?? null
        };
      } catch (error) {
        logWorkbenchFactsQueryFailure(options.logger ?? console, params, error);
        return {
          facts: emptyWorkbenchFacts(),
          count: 0,
          durable: true,
          error,
          projection: projectionStoreUnavailableTrace(safeTraceId(params.traceId) ?? "trc_unassigned", error)
        };
      }
    }
    return {
      facts: emptyWorkbenchFacts(),
      count: 0,
      durable: false,
      persistence: null
    };
  }

  async function getSessionById(sessionId) {
    const safeId = safeSessionId(sessionId);
    if (!safeId) return null;
    const session = await accessStore?.getAgentSession?.(safeId) ?? null;
    return canActorReadSession(session, actor) ? session : null;
  }

  async function getSessionByRouteId(routeId) {
    const sessionId = safeSessionId(routeId);
    if (sessionId) return getSessionById(sessionId);
    const conversationId = safeConversationId(routeId);
    if (!conversationId) return null;
    const listed = await accessStore?.listAgentSessionsForUser?.({
      ownerUserId: actor?.id,
      actorRole: actor?.role,
      ownerScoped: true,
      conversationId,
      limit: 1,
      includeArchived: false
    }) ?? [];
    const matched = listed.find((session) => canActorReadSession(session, actor)) ?? null;
    if (matched) return matched;
    const suffix = conversationId.slice("cnv_".length);
    const deterministicSessionId = safeSessionId(`ses_${suffix}`);
    return deterministicSessionId ? getSessionById(deterministicSessionId) : null;
  }

  async function getSessionByTraceId(traceId) {
    const safeId = safeTraceId(traceId);
    if (!safeId) return null;
    if (traceSessionCache?.has(safeId)) {
      const cached = traceSessionCache.get(safeId) ?? null;
      return canActorReadSession(cached, actor) ? cached : null;
    }
    const session = await accessStore?.getAgentSessionByTraceId?.(safeId) ?? null;
    traceSessionCache?.set(safeId, session);
    return canActorReadSession(session, actor) ? session : null;
  }

  async function listSessions({ limit, offset = 0, includeRouteId } = {}) {
    let sessions = await accessStore?.listAgentSessionsForUser?.({
      ownerUserId: actor?.id,
      actorRole: actor?.role,
      ownerScoped: true,
      limit,
      offset,
      includeArchived: false
    }) ?? [];
    const includeSessionId = safeSessionId(includeRouteId);
    if (includeRouteId && !sessions.some((session) => session.id === includeSessionId)) {
      const included = await getSessionByRouteId(includeRouteId);
      if (included) sessions = [included, ...sessions];
    }
    return sessions.filter((session) => canActorReadSession(session, actor));
  }

  function resultForTrace(traceId) {
    const safeId = safeTraceId(traceId);
    return safeId ? results?.get?.(safeId) ?? null : null;
  }

  function traceSnapshotSync(traceId) {
    return traceStore.snapshot(traceId);
  }

  async function traceSnapshot(traceId) {
    const memory = traceSnapshotSync(traceId);
    if (hasTraceProjection(memory) && TERMINAL_STATUSES.has(normalizeStatus(memory.status))) return memory;
    const durable = await durableTraceSnapshot(runtimeStore, traceId);
    if (isProjectionDiagnosticTrace(durable) && hasTraceProjection(memory)) return mergeTraceProjectionDiagnostic(memory, durable);
    if (durable && shouldPreferDurableTrace(memory, durable)) return durable;
    if (hasTraceProjection(memory)) return memory;
    return durable ?? memory;
  }

  function subscribeTrace(traceId, listener) {
    return traceStore.subscribe(traceId, listener);
  }

  function subscribeDurableProjection(traceId, listener, pollOptions = {}) {
    const runtimeStore = options.runtimeStore ?? null;
    if (typeof runtimeStore?.readWorkbenchProjectionOutbox !== "function") return () => {};
    const pollMs = Math.max(500, Number(pollOptions.pollMs ?? 2000));
    let cursor = 0;
    let stopped = false;
    const poll = async () => {
      if (stopped) return;
      try {
        const rows = await runtimeStore.readWorkbenchProjectionOutbox({ afterSeq: cursor, limit: 50, traceId });
        for (const row of rows) {
          if (stopped) break;
          cursor = row.outboxSeq;
          listener(row);
        }
      } catch { /* durable projection poll errors are non-fatal for SSE consumers */ }
    };
    void poll();
    const timer = setInterval(() => { void poll(); }, pollMs);
    return () => {
      stopped = true;
      clearInterval(timer);
    };
  }

  return {
    queryFacts,
    getSessionById,
    getSessionByRouteId,
    getSessionByTraceId,
    listSessions,
    resultForTrace,
    traceSnapshot,
    traceSnapshotSync,
    subscribeTrace,
    subscribeDurableProjection,
    canReadOwner: (ownerUserId) => canActorReadOwner(ownerUserId, actor)
  };
}

function logWorkbenchFactsQueryFailure(logger, params = {}, error = null) {
  const families = Array.isArray(params.families) ? params.families : [];
  const event = {
    event: "workbench_facts_query_failed",
    ok: false,
    families,
    hasSessionId: Boolean(params.sessionId),
    hasTraceId: Boolean(params.traceId),
    hasOwnerUserId: Boolean(params.ownerUserId),
    errorName: error?.name ?? "Error",
    errorCode: error?.code ?? null,
    message: error instanceof Error ? error.message : String(error ?? "unknown"),
    valuesRedacted: true
  };
  try {
    if (typeof logger?.warn === "function") logger.warn(JSON.stringify(event));
    else if (typeof logger?.log === "function") logger.log(JSON.stringify(event));
  } catch {}
}

async function durableTraceSnapshot(runtimeStore, traceId) {
  if (!runtimeStore || typeof runtimeStore.queryAgentTraceEvents !== "function") return null;
  const safeId = safeTraceId(traceId);
  if (!safeId) return null;
  const cached = terminalDurableTraceCache.get(safeId);
  if (cached) {
    terminalDurableTraceCache.delete(safeId);
    terminalDurableTraceCache.set(safeId, cached);
    return cached;
  }
  let result;
  try {
    result = await runtimeStore.queryAgentTraceEvents({ traceId: safeId });
  } catch (error) {
    return projectionStoreUnavailableTrace(safeId, error);
  }
  const events = Array.isArray(result?.events) ? result.events : [];
  if (events.length === 0) return null;
  const normalizedEvents = events.map((event, index) => ({ ...event, traceId: safeId, seq: eventSeq(event, index) }));
  const firstEvent = normalizedEvents[0] ?? null;
  const lastEvent = normalizedEvents.at(-1) ?? null;
  const status = durableTraceStatus(normalizedEvents);
  const terminalEvidence = traceTerminalEvidence({ events: normalizedEvents, status });
  const snapshot = {
    traceId: safeId,
    status,
    createdAt: firstEvent?.createdAt ?? null,
    updatedAt: lastEvent?.createdAt ?? firstEvent?.createdAt ?? null,
    startedAt: firstEvent?.createdAt ?? null,
    finishedAt: TERMINAL_STATUSES.has(status) ? lastEvent?.createdAt ?? null : null,
    eventCount: normalizedEvents.length,
    events: normalizedEvents,
    assistantStreams: [],
    eventLabels: normalizedEvents.map((event) => event.label).filter(Boolean),
    lastEvent,
    terminalEvidence,
    outputTruncated: normalizedEvents.some((event) => event.outputTruncated === true),
    valuesPrinted: false
  };
  rememberTerminalDurableTrace(snapshot);
  return snapshot;
}

function rememberTerminalDurableTrace(snapshot) {
  const traceId = safeTraceId(snapshot?.traceId);
  if (!traceId || !TERMINAL_STATUSES.has(normalizeStatus(snapshot?.status))) return;
  terminalDurableTraceCache.delete(traceId);
  terminalDurableTraceCache.set(traceId, snapshot);
  while (terminalDurableTraceCache.size > TERMINAL_DURABLE_TRACE_CACHE_MAX) {
    const oldest = terminalDurableTraceCache.keys().next().value;
    terminalDurableTraceCache.delete(oldest);
  }
}

function projectionStoreUnavailableTrace(traceId, error) {
  const message = "运行记录投影存储暂不可读，Trace 更新暂不可见。";
  const projection = {
    projectionStatus: "blocked",
    projectionHealth: "unavailable",
    blocker: {
      code: "projection_store_unavailable",
      layer: "runtime-store",
      category: "projection-store-unavailable",
      message,
      causeCode: error?.code ?? null,
      retryable: true,
      valuesPrinted: false
    },
    valuesRedacted: true
  };
  return {
    traceId,
    status: "missing",
    eventCount: 0,
    events: [],
    projection,
    projectionStatus: projection.projectionStatus,
    projectionHealth: projection.projectionHealth,
    blocker: projection.blocker,
    updatedAt: new Date().toISOString(),
    valuesPrinted: false
  };
}

function isProjectionDiagnosticTrace(snapshot) {
  return Boolean(snapshot?.projection?.blocker || snapshot?.projectionHealth || snapshot?.blocker);
}

function mergeTraceProjectionDiagnostic(memory, diagnostic) {
  return {
    ...memory,
    projection: diagnostic.projection ?? memory?.projection ?? null,
    projectionStatus: diagnostic.projectionStatus ?? diagnostic.projection?.projectionStatus ?? memory?.projectionStatus ?? null,
    projectionHealth: diagnostic.projectionHealth ?? diagnostic.projection?.projectionHealth ?? memory?.projectionHealth ?? null,
    blocker: diagnostic.blocker ?? diagnostic.projection?.blocker ?? memory?.blocker ?? null,
    valuesPrinted: false
  };
}

function hasTraceProjection(snapshot) {
  return Boolean(snapshot?.status !== "missing" && (Number(snapshot?.eventCount ?? 0) > 0 || snapshot?.lastEvent));
}

function shouldPreferDurableTrace(memory, durable) {
  if (!durable) return false;
  if (!hasTraceProjection(memory)) return true;
  const memoryStatus = normalizeWorkbenchStatus(memory?.status);
  const durableStatus = normalizeWorkbenchStatus(durable?.status);
  if (TERMINAL_STATUSES.has(durableStatus) && !TERMINAL_STATUSES.has(memoryStatus)) return true;
  return traceLastSeq(durable) > traceLastSeq(memory);
}

function traceLastSeq(snapshot) {
  const events = Array.isArray(snapshot?.events) ? snapshot.events : [];
  const lastEvent = snapshot?.lastEvent ?? events.at(-1) ?? null;
  const indexedMax = events.reduce((max, event, index) => Math.max(max, eventSeq(event, index)), 0);
  return Math.max(indexedMax, lastEvent ? eventSeq(lastEvent, Math.max(0, events.length - 1)) : 0);
}

function canActorReadSession(session, actor) {
  if (!session || !actor) return false;
  if (normalizeStatus(session.status) === "archived") return false;
  if (actor.role === "admin") return true;
  return session.ownerUserId === actor.id;
}

function canActorReadOwner(ownerUserId, actor) {
  if (!ownerUserId || !actor) return true;
  if (actor.role === "admin") return true;
  return ownerUserId === actor.id;
}

function eventSeq(event, index) {
  const seq = Number(event?.seq);
  return Number.isFinite(seq) && seq > 0 ? Math.trunc(seq) : index + 1;
}

function normalizeStatus(value) {
  const text = String(value ?? "").trim().toLowerCase().replace(/_/gu, "-");
  if (text === "cancelled") return "canceled";
  return text || "unknown";
}

function normalizeWorkbenchFactsResult(facts = {}) {
  const normalized = emptyWorkbenchFacts();
  for (const key of Object.keys(normalized)) {
    normalized[key] = Array.isArray(facts?.[key]) ? facts[key].map((item) => ({ ...item })) : [];
  }
  return normalized;
}

function emptyWorkbenchFacts() {
  return {
    sessions: [],
    messages: [],
    parts: [],
    turns: [],
    traceEvents: [],
    checkpoints: []
  };
}