fix: remove workbench read-side fallback
This commit is contained in:
lyon
@@ -1678,7 +1678,7 @@ test("workbench realtime stream surfaces facts blocker instead of legacy trace f
|
||||
|
||||
try {
|
||||
const { port } = server.address();
|
||||
const eventsPromise = getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}`, 4);
|
||||
const eventsPromise = getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}&traceId=${encodeURIComponent(traceId)}`, 4);
|
||||
const events = await eventsPromise;
|
||||
assert.deepEqual(events.map((event) => event.event), [
|
||||
"workbench.connected",
|
||||
@@ -1687,7 +1687,7 @@ test("workbench realtime stream surfaces facts blocker instead of legacy trace f
|
||||
"workbench.turn.snapshot"
|
||||
]);
|
||||
assert.equal(events[0].data.filters.sessionId, session.id);
|
||||
assert.equal(events[1].data.error.code, "workbench_facts_session_missing");
|
||||
assert.equal(events[1].data.error.code, "workbench_facts_trace_missing");
|
||||
assert.equal(events[2].data.traceId, traceId);
|
||||
assert.equal(events[2].data.snapshot.status, "unknown");
|
||||
assert.equal(events[2].data.snapshot.eventCount, 0);
|
||||
|
||||
@@ -15,7 +15,7 @@ import {
|
||||
} from "./server-http-utils.ts";
|
||||
import { createWorkbenchReadModel } from "./workbench-read-model.ts";
|
||||
import { createWorkbenchRuntimeClient } from "./workbench-runtime-client.ts";
|
||||
import { createWorkbenchTurnProjection, createWorkbenchTurnTimingProjection, durableTraceStatus, RUNNING_STATUSES, TERMINAL_STATUSES } from "./workbench-turn-projection.ts";
|
||||
import { durableTraceStatus, RUNNING_STATUSES, TERMINAL_STATUSES } from "./workbench-turn-projection.ts";
|
||||
import { emitHttpServerRequestSpan } from "./otel-trace.ts";
|
||||
|
||||
const DEFAULT_PAGE_LIMIT = 50;
|
||||
@@ -26,7 +26,7 @@ const WORKBENCH_REALTIME_DRAIN_TIMEOUT_MS = 2500;
|
||||
const WORKBENCH_SESSION_LIST_PAGE_FAMILIES = Object.freeze(["sessions"]);
|
||||
const WORKBENCH_SESSION_DETAIL_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]);
|
||||
const WORKBENCH_SESSION_MESSAGE_PAGE_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]);
|
||||
const WORKBENCH_TRACE_EVENT_METADATA_FAMILIES = Object.freeze(["sessions", "messages", "turns", "checkpoints"]);
|
||||
const WORKBENCH_TRACE_EVENT_METADATA_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]);
|
||||
const WORKBENCH_TRACE_EVENT_PAGE_FAMILIES = Object.freeze(["traceEvents"]);
|
||||
const activeWorkbenchRealtimeConnections = new Set();
|
||||
|
||||
@@ -203,12 +203,25 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
|
||||
}
|
||||
});
|
||||
|
||||
const initialSession = requestedSessionId ? await readModel.getSessionById(requestedSessionId) : null;
|
||||
const initialSessionSnapshot = objectValue(initialSession?.session);
|
||||
const streamSessionId = safeSessionId(initialSession?.id ?? requestedSessionId) ?? null;
|
||||
const streamThreadId = safeOpaqueId(initialSession?.threadId ?? initialSessionSnapshot.threadId) ?? (textValue(initialSession?.threadId ?? initialSessionSnapshot.threadId) || null);
|
||||
const resolveInitialSession = requestedSessionId && requestedAfterSeq <= 0;
|
||||
const initialContext = resolveInitialSession
|
||||
? await realtimeInitialSessionContext(readModel, auth.actor, requestedSessionId)
|
||||
: { facts: emptyFactSet(), session: null, blocker: null };
|
||||
if (initialContext.blocker) {
|
||||
writeEvent("workbench.error", {
|
||||
type: "error",
|
||||
reason: "initial-session",
|
||||
sessionId: requestedSessionId,
|
||||
traceId: requestedTraceId,
|
||||
error: initialContext.blocker
|
||||
});
|
||||
}
|
||||
const initialFactSession = initialContext.session;
|
||||
const streamSessionId = factSessionId(initialFactSession) ?? requestedSessionId ?? null;
|
||||
const streamThreadId = safeOpaqueId(initialFactSession?.threadId) ?? (textValue(initialFactSession?.threadId) || null);
|
||||
const activeTraceId = requestedTraceId
|
||||
?? safeTraceId(initialSession?.lastTraceId ?? initialSessionSnapshot.lastTraceId ?? initialSessionSnapshot.currentTraceId)
|
||||
?? factLastTraceId(initialFactSession)
|
||||
?? factLatestTraceIdForSession(initialContext.facts, streamSessionId)
|
||||
?? null;
|
||||
realtimeSessionId = streamSessionId ?? requestedSessionId ?? null;
|
||||
realtimeTraceId = activeTraceId ?? requestedTraceId ?? null;
|
||||
@@ -240,7 +253,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
|
||||
outboxCursor = row.outboxSeq;
|
||||
const rowTraceId = safeTraceId(row.traceId) ?? activeTraceId;
|
||||
const rowSessionId = safeSessionId(row.sessionId ?? streamSessionId) ?? streamSessionId;
|
||||
if (rowTraceId && !outboxTraceSnapshots.has(rowTraceId)) {
|
||||
if (requestedAfterSeq <= 0 && rowTraceId && !outboxTraceSnapshots.has(rowTraceId)) {
|
||||
outboxTraceSnapshots.add(rowTraceId);
|
||||
await writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "outbox-session" });
|
||||
}
|
||||
@@ -701,9 +714,15 @@ async function handleWorkbenchSessionList(request, response, url, options, actor
|
||||
const offset = cursorOffset(url.searchParams.get("cursor") ?? url.searchParams.get("after"));
|
||||
const includeSessionId = safeSessionId(url.searchParams.get("includeSessionId"));
|
||||
const includeRouteId = includeSessionId ?? safeConversationId(url.searchParams.get("includeSessionId"));
|
||||
const runtime = options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent });
|
||||
const runtime = workbenchRuntimeReader(request, options);
|
||||
if (typeof runtime?.listWorkbenchSessions !== "function") {
|
||||
throw runtimeDependencyError("workbench_runtime_unconfigured", "Workbench runtime service is required for session list.", false);
|
||||
if (typeof runtime?.queryWorkbenchFacts !== "function") {
|
||||
throw runtimeDependencyError("workbench_runtime_unconfigured", "Workbench runtime service is required for session list.", false);
|
||||
}
|
||||
const readModel = workbenchRuntimeFactsReadModel(request, options, actor);
|
||||
const payload = await handleWorkbenchSessionListFromFacts(readModel, url, options, actor, { limit, offset, includeRouteId });
|
||||
recordWorkbenchCacheMetric(options, url, payload, startedAt);
|
||||
return sendJson(response, 200, payload);
|
||||
}
|
||||
const payload = await runtime.listWorkbenchSessions({
|
||||
limit,
|
||||
@@ -716,48 +735,17 @@ async function handleWorkbenchSessionList(request, response, url, options, actor
|
||||
return sendJson(response, 200, payload);
|
||||
}
|
||||
|
||||
function runtimeDependencyError(code, message, retryable = true) {
|
||||
const error = new Error(message || code);
|
||||
error.name = "WorkbenchRuntimeDependencyError";
|
||||
error.code = code;
|
||||
error.data = { serviceId: "hwlab-workbench-runtime", retryable, transient: retryable, valuesRedacted: true };
|
||||
return error;
|
||||
}
|
||||
|
||||
function workbenchRuntimeFactsReadModel(request, options, actor) {
|
||||
const runtime = options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent });
|
||||
if (typeof runtime?.queryWorkbenchFacts !== "function") {
|
||||
throw runtimeDependencyError("workbench_runtime_unconfigured", "Workbench runtime service is required for Workbench facts reads.", false);
|
||||
}
|
||||
return {
|
||||
queryFacts: (query = {}) => runtime.queryWorkbenchFacts({
|
||||
...query,
|
||||
actor: { id: actor.id, role: actor.role ?? "user", valuesRedacted: true }
|
||||
})
|
||||
};
|
||||
}
|
||||
|
||||
function workbenchRuntimeOutboxReader(request, options) {
|
||||
return options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent });
|
||||
}
|
||||
|
||||
async function handleWorkbenchSessionListBunLegacy(response, url, options, actor) {
|
||||
if (url.searchParams.has("projectId") || url.searchParams.has("workspaceId")) return sendJson(response, 400, workbenchError("workbench_authority_removed", "Workbench session list is keyed by sessionId only."));
|
||||
async function handleWorkbenchSessionListFromFacts(readModel, url, options, actor, { limit, offset, includeRouteId }) {
|
||||
const perf = options.backendPerformance;
|
||||
const limit = boundedSessionListLimit(url.searchParams.get("limit"));
|
||||
const offset = cursorOffset(url.searchParams.get("cursor") ?? url.searchParams.get("after"));
|
||||
const includeSessionId = safeSessionId(url.searchParams.get("includeSessionId"));
|
||||
const includeRouteId = includeSessionId ?? safeConversationId(url.searchParams.get("includeSessionId"));
|
||||
const readModel = createWorkbenchReadModel(options, actor);
|
||||
const pageQuery = { ownerUserId: actor.role === "admin" ? undefined : actor.id, limit: limit + offset + 1, families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES, sessionsOrder: "updated_desc", sessionProjection: "summary" };
|
||||
const includeQueryPromise = includeRouteId ? queryWorkbenchSessionInclude(readModel, includeRouteId, perf) : null;
|
||||
const pageResult = await (perf ? perf.measure("workbench_session_page_query", () => readModel.queryFacts(pageQuery)) : readModel.queryFacts(pageQuery));
|
||||
if (pageResult.error) return sendJson(response, 503, workbenchProjectionStoreError(pageResult.error));
|
||||
if (pageResult.error) throw pageResult.error;
|
||||
let pageFacts = pageResult.facts;
|
||||
let naturalPage = visibleFactSessions(pageFacts, actor).slice(offset, offset + limit + 1);
|
||||
if (includeRouteId && !naturalPage.some((session) => factSessionMatchesRouteId(session, includeRouteId))) {
|
||||
const includeResult = await includeQueryPromise;
|
||||
if (includeResult.error) return sendJson(response, 503, workbenchProjectionStoreError(includeResult.error));
|
||||
if (includeResult.error) throw includeResult.error;
|
||||
const included = visibleFactSessions(includeResult.facts, actor)[0] ?? null;
|
||||
if (included) {
|
||||
pageFacts = mergeFactSets(pageFacts, includeResult.facts);
|
||||
@@ -766,11 +754,11 @@ async function handleWorkbenchSessionListBunLegacy(response, url, options, actor
|
||||
}
|
||||
const pageSessions = naturalPage.slice(0, limit);
|
||||
const summaryResult = await (perf ? perf.measure("workbench_session_summary_query", () => queryFactsForSessionSummaries(readModel, pageSessions)) : queryFactsForSessionSummaries(readModel, pageSessions));
|
||||
if (summaryResult.error) return sendJson(response, 503, workbenchProjectionStoreError(summaryResult.error));
|
||||
if (summaryResult.error) throw summaryResult.error;
|
||||
pageFacts = mergeFactSets(pageFacts, summaryResult.facts);
|
||||
const summaries = pageSessions.map((session) => factSessionSummary(session, pageFacts)).filter(Boolean);
|
||||
const hasMore = naturalPage.length > limit;
|
||||
sendJson(response, 200, {
|
||||
return {
|
||||
ok: true,
|
||||
status: "succeeded",
|
||||
contractVersion: "workbench-sessions-v1",
|
||||
@@ -779,9 +767,60 @@ async function handleWorkbenchSessionListBunLegacy(response, url, options, actor
|
||||
cursor: offset > 0 ? cursorFromOffset(offset) : null,
|
||||
hasMore,
|
||||
nextCursor: hasMore ? cursorFromOffset(offset + limit) : null,
|
||||
cache: pageResult.cache ?? summaryResult.cache ?? null,
|
||||
valuesRedacted: true,
|
||||
secretMaterialStored: false
|
||||
});
|
||||
};
|
||||
}
|
||||
|
||||
function runtimeDependencyError(code, message, retryable = true) {
|
||||
const error = new Error(message || code);
|
||||
error.name = "WorkbenchRuntimeDependencyError";
|
||||
error.code = code;
|
||||
error.data = { serviceId: "hwlab-workbench-runtime", retryable, transient: retryable, valuesRedacted: true };
|
||||
return error;
|
||||
}
|
||||
|
||||
function workbenchRuntimeReader(request, options) {
|
||||
return options.workbenchRuntime
|
||||
?? options.runtimeStore
|
||||
?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent });
|
||||
}
|
||||
|
||||
function workbenchRuntimeFactsReadModel(request, options, actor) {
|
||||
const runtime = workbenchRuntimeReader(request, options);
|
||||
if (typeof runtime?.queryWorkbenchFacts !== "function") {
|
||||
throw runtimeDependencyError("workbench_runtime_unconfigured", "Workbench runtime service is required for Workbench facts reads.", false);
|
||||
}
|
||||
return {
|
||||
async queryFacts(query = {}) {
|
||||
try {
|
||||
return await runtime.queryWorkbenchFacts({
|
||||
...query,
|
||||
actor: { id: actor.id, role: actor.role ?? "user", valuesRedacted: true }
|
||||
});
|
||||
} catch (error) {
|
||||
return { facts: emptyFactSet(), count: 0, durable: true, error: workbenchRuntimeFactsQueryError(error), persistence: null };
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function workbenchRuntimeFactsQueryError(error) {
|
||||
const normalized = error instanceof Error ? error : new Error(String(error ?? "Workbench facts query failed."));
|
||||
const data = objectValue(normalized.data);
|
||||
normalized.data = {
|
||||
...data,
|
||||
retryable: data.retryable !== false,
|
||||
transient: data.transient !== false,
|
||||
retryAfterMs: data.retryAfterMs ?? 5000,
|
||||
valuesRedacted: true
|
||||
};
|
||||
return normalized;
|
||||
}
|
||||
|
||||
function workbenchRuntimeOutboxReader(request, options) {
|
||||
return workbenchRuntimeReader(request, options);
|
||||
}
|
||||
|
||||
function queryWorkbenchSessionInclude(readModel, includeRouteId, perf = null) {
|
||||
@@ -790,13 +829,6 @@ function queryWorkbenchSessionInclude(readModel, includeRouteId, perf = null) {
|
||||
return promise.catch((error) => ({ error }));
|
||||
}
|
||||
|
||||
function sessionMatchesRouteId(session, routeId) {
|
||||
const sessionId = safeSessionId(routeId);
|
||||
if (sessionId && session?.id === sessionId) return true;
|
||||
const conversationId = safeConversationId(routeId);
|
||||
return Boolean(conversationId && session?.conversationId === conversationId);
|
||||
}
|
||||
|
||||
async function queryFactsByRouteId(readModel, routeId, query = {}) {
|
||||
const baseQuery = { ...query, limit: query.limit ?? MAX_PAGE_LIMIT };
|
||||
const sessionId = safeSessionId(routeId);
|
||||
@@ -828,7 +860,7 @@ async function queryFactsForSessionSummaries(readModel, sessions = []) {
|
||||
// observer. Avoid parallel fan-out here so one list request cannot consume
|
||||
// multiple durable-read connections while another page is refreshing.
|
||||
const bySession = sessionIds.length > 0
|
||||
? await readModel.queryFacts({ sessionIds, families: ["messages", "turns"] })
|
||||
? await readModel.queryFacts({ sessionIds, families: ["messages", "parts", "turns"] })
|
||||
: { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
|
||||
if (bySession.error) return bySession;
|
||||
const byTrace = traceIds.length > 0
|
||||
@@ -1593,7 +1625,7 @@ async function handleWorkbenchMessagePage(request, response, url, options, actor
|
||||
const hasCursor = Boolean(textValue(cursorParam));
|
||||
const result = await queryFactsByRouteId(readModel, sessionId, {
|
||||
families: WORKBENCH_SESSION_MESSAGE_PAGE_FAMILIES,
|
||||
...(hasCursor ? {} : { limit, messagesOrder: "updated_desc", partsOrder: "updated_desc", turnsOrder: "updated_desc", checkpointsOrder: "updated_desc" })
|
||||
...(hasCursor ? {} : { limit: limit + 1, messagesOrder: "updated_desc", partsOrder: "updated_desc", turnsOrder: "updated_desc", checkpointsOrder: "updated_desc" })
|
||||
});
|
||||
if (result.error) return sendJson(response, 503, workbenchProjectionStoreError(result.error));
|
||||
const session = visibleFactSessions(result.facts, actor)[0] ?? null;
|
||||
@@ -1603,7 +1635,7 @@ async function handleWorkbenchMessagePage(request, response, url, options, actor
|
||||
return sendJson(response, 404, body);
|
||||
}
|
||||
const messages = factMessagesForSession(session, result.facts);
|
||||
const offset = hasCursor ? cursorOffset(cursorParam) : Math.max(0, messages.length - limit);
|
||||
const offset = hasCursor ? cursorOffset(cursorParam) : 0;
|
||||
const page = messages.slice(offset, offset + limit);
|
||||
const nextOffset = offset + page.length;
|
||||
const resolvedSessionId = factSessionId(session);
|
||||
@@ -1634,7 +1666,7 @@ async function handleWorkbenchTurnSnapshot(request, response, url, options, acto
|
||||
return sendJson(response, 400, body);
|
||||
}
|
||||
const readModel = workbenchRuntimeFactsReadModel(request, options, actor);
|
||||
const result = await readModel.queryFacts({ traceId, limit: MAX_PAGE_LIMIT, families: WORKBENCH_TRACE_EVENT_METADATA_FAMILIES });
|
||||
const result = await readModel.queryFacts({ traceId, limit: MAX_PAGE_LIMIT, families: [...WORKBENCH_TRACE_EVENT_METADATA_FAMILIES, ...WORKBENCH_TRACE_EVENT_PAGE_FAMILIES] });
|
||||
if (result.error) {
|
||||
const body = workbenchProjectionStoreError(result.error);
|
||||
recordWorkbenchTurnReadMetric(options, url, 503, body, startedAt);
|
||||
@@ -1715,7 +1747,8 @@ async function handleWorkbenchTraceEventPage(request, response, url, options, ac
|
||||
const missingTraceEvents = traceEventPageMissing(page, projection, pageOptions);
|
||||
if (missingTraceEvents) {
|
||||
const blocker = traceEventsReadModelBlocker("workbench_trace_events_missing", "Workbench trace events are missing from the durable read model.", { traceId, projection, session, route: url.pathname });
|
||||
page.blocker = blocker;
|
||||
attachWorkbenchReadModelDiagnosticOtel(request, { route: "/v1/workbench/traces/:id/events", code: "workbench_trace_events_missing", sessionId: factSessionId(session), traceId, count: pageResult.count });
|
||||
return sendJson(response, 404, workbenchTraceEventsReadModelError(blocker, { traceId, projection, session, turn: traceTurn }));
|
||||
}
|
||||
const responseProjection = page.blocker ? blockedTraceEventProjection(projection, page.blocker) : projection;
|
||||
const body = {
|
||||
@@ -1891,300 +1924,6 @@ function workbenchTraceEventsReadModelError(blocker, { traceId, projection = {},
|
||||
});
|
||||
}
|
||||
|
||||
async function visibleSessionById(store, sessionId, actor) {
|
||||
const safeId = safeSessionId(sessionId);
|
||||
if (!safeId) return null;
|
||||
const session = await store?.getAgentSession?.(safeId) ?? null;
|
||||
return canActorReadSession(session, actor) ? session : null;
|
||||
}
|
||||
|
||||
async function visibleSessionByRouteId(store, routeId, actor) {
|
||||
const sessionId = safeSessionId(routeId);
|
||||
if (sessionId) return visibleSessionById(store, sessionId, actor);
|
||||
const conversationId = safeConversationId(routeId);
|
||||
if (!conversationId) return null;
|
||||
const listed = await store?.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 ? visibleSessionById(store, deterministicSessionId, actor) : null;
|
||||
}
|
||||
|
||||
async function visibleSessionByTrace(store, traceId, actor) {
|
||||
const safeId = safeTraceId(traceId);
|
||||
if (!safeId) return null;
|
||||
const session = await store?.getAgentSessionByTraceId?.(safeId) ?? null;
|
||||
return canActorReadSession(session, actor) ? session : null;
|
||||
}
|
||||
|
||||
async function sessionProjectionOptions(readModel, session, options) {
|
||||
const snapshot = objectValue(session?.session);
|
||||
const traceId = safeTraceId(session?.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId) ?? null;
|
||||
if (!traceId) return options;
|
||||
const trace = await readModel.traceSnapshot(traceId);
|
||||
const result = readModel.resultForTrace(traceId);
|
||||
const projection = createWorkbenchTurnProjection({ traceId, result, session, trace });
|
||||
const projectionDiagnostic = readModel.projectionDiagnostics({ traceId, result, trace, projection });
|
||||
return { ...options, trace, result, projection, projectionDiagnostic };
|
||||
}
|
||||
|
||||
function sessionListSummaries(readModel, sessions, options) {
|
||||
return sessions
|
||||
.map((session) => sessionSummary(session, sessionListProjectionOptions(readModel, session, options)))
|
||||
.filter(Boolean);
|
||||
}
|
||||
|
||||
function sessionListProjectionOptions(readModel, session, options) {
|
||||
const snapshot = objectValue(session?.session);
|
||||
const traceId = safeTraceId(session?.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId ?? snapshot.traceId) ?? null;
|
||||
if (!traceId) return options;
|
||||
const memoryResult = readModel.resultForTrace(traceId);
|
||||
const hasDurableProjection = Boolean(snapshot.projectionStatus || snapshot.terminal !== undefined || snapshot.sealedAt);
|
||||
const result = hasDurableProjection ? null : (memoryResult ?? compactSessionTraceResult(snapshot, session, traceId));
|
||||
const memoryTrace = hasDurableProjection ? null : traceSnapshotSync(options, traceId);
|
||||
const trace = hasDurableProjection ? null : compactSessionTraceSnapshot(snapshot, session, traceId, result, memoryTrace);
|
||||
if (hasDurableProjection) {
|
||||
const projection = createWorkbenchTurnProjection({
|
||||
traceId,
|
||||
turnId: traceId,
|
||||
result: { status: snapshot.projectionStatus ?? snapshot.status, terminal: snapshot.terminal, timing: snapshot.timing, finalResponse: snapshot.finalResponse, valuesRedacted: true },
|
||||
session: null,
|
||||
trace: null
|
||||
});
|
||||
const projectionDiagnostic = readModel.projectionDiagnostics({ traceId, result: null, trace: null, projection });
|
||||
return { ...options, trace: null, result: null, projection, projectionDiagnostic };
|
||||
}
|
||||
const projection = createWorkbenchTurnProjection({ traceId, result, session, trace });
|
||||
const projectionDiagnostic = readModel.projectionDiagnostics({ traceId, result, trace, projection });
|
||||
return { ...options, trace, result, projection, projectionDiagnostic };
|
||||
}
|
||||
|
||||
function compactSessionTraceResult(snapshot, session, traceId) {
|
||||
const record = compactSessionTraceRecord(snapshot, traceId);
|
||||
if (record) return traceResultFromCompactRecord(record, traceId);
|
||||
const snapshotTraceId = safeTraceId(snapshot.lastTraceId ?? snapshot.currentTraceId ?? snapshot.traceId ?? session?.lastTraceId) ?? null;
|
||||
if (snapshotTraceId !== traceId) return null;
|
||||
const traceSummary = optionalObject(snapshot.traceSummary);
|
||||
const finalResponse = optionalObject(snapshot.finalResponse) ?? messageAuthorityTextValue(snapshot.finalResponse);
|
||||
const agentRun = optionalObject(snapshot.agentRun);
|
||||
if (!traceSummary && !finalResponse && !agentRun) return null;
|
||||
return compactTraceResult({ traceId, status: snapshot.status, finalResponse, traceSummary, agentRun, session });
|
||||
}
|
||||
|
||||
function compactSessionTraceRecord(snapshot, traceId) {
|
||||
for (const source of [snapshot.traceResults, snapshot.traceResult]) {
|
||||
const container = objectValue(source);
|
||||
if (safeTraceId(container.traceId) === traceId) return container;
|
||||
const byTraceId = objectValue(container[traceId]);
|
||||
if (Object.keys(byTraceId).length > 0) return { ...byTraceId, traceId };
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function traceResultFromCompactRecord(record, traceId) {
|
||||
const traceSummary = optionalObject(record.traceSummary);
|
||||
const finalResponse = optionalObject(record.finalResponse) ?? messageAuthorityTextValue(record.finalResponse);
|
||||
const agentRun = optionalObject(record.agentRun) ?? optionalObject(traceSummary?.agentRun);
|
||||
if (!traceSummary && !finalResponse && !agentRun) return null;
|
||||
return compactTraceResult({ traceId, status: record.status ?? record.terminalStatus, finalResponse, traceSummary, agentRun, session: record });
|
||||
}
|
||||
|
||||
function compactTraceResult({ traceId, status, finalResponse, traceSummary, agentRun, session }) {
|
||||
return {
|
||||
traceId,
|
||||
status: normalizeStatus(firstTextValue(status, traceSummary?.terminalStatus, agentRun?.terminalStatus, agentRun?.status)),
|
||||
ownerUserId: session?.ownerUserId ?? null,
|
||||
conversationId: session?.conversationId ?? null,
|
||||
sessionId: session?.sessionId ?? session?.id ?? null,
|
||||
threadId: session?.threadId ?? null,
|
||||
finalResponse: finalResponse || null,
|
||||
traceSummary: traceSummary ?? null,
|
||||
agentRun: agentRun ?? null,
|
||||
updatedAt: session?.updatedAt ?? traceSummary?.updatedAt ?? agentRun?.updatedAt ?? null,
|
||||
valuesRedacted: true,
|
||||
secretMaterialStored: false
|
||||
};
|
||||
}
|
||||
|
||||
function compactSessionTraceSnapshot(snapshot, session, traceId, result, memoryTrace) {
|
||||
const traceSummary = optionalObject(result?.traceSummary) ?? optionalObject(snapshot.traceSummary);
|
||||
const agentRun = optionalObject(result?.agentRun) ?? optionalObject(traceSummary?.agentRun);
|
||||
const finalResponse = optionalObject(result?.finalResponse) ?? optionalObject(snapshot.finalResponse) ?? messageAuthorityTextValue(result?.finalResponse ?? snapshot.finalResponse);
|
||||
const status = normalizeStatus(firstTextValue(result?.status, traceSummary?.terminalStatus, agentRun?.terminalStatus, memoryTrace?.status));
|
||||
if (!traceSummary && !agentRun && !finalResponse && memoryTrace) return memoryTrace;
|
||||
const eventCount = compactTraceEventCount(traceSummary, agentRun, memoryTrace);
|
||||
const terminal = TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status);
|
||||
const compactTrace = {
|
||||
...(memoryTrace ?? {}),
|
||||
traceId,
|
||||
status,
|
||||
eventCount,
|
||||
events: Array.isArray(memoryTrace?.events) ? memoryTrace.events : [],
|
||||
lastEvent: memoryTrace?.lastEvent ?? null,
|
||||
updatedAt: result?.updatedAt ?? traceSummary?.updatedAt ?? agentRun?.updatedAt ?? memoryTrace?.updatedAt ?? session?.updatedAt ?? null,
|
||||
finalResponse: finalResponse || null,
|
||||
terminalEvidence: terminal && traceSummary ? { status, terminalStatus: traceSummary.terminalStatus ?? status, traceSummary, valuesRedacted: true } : memoryTrace?.terminalEvidence ?? null,
|
||||
agentRun: agentRun ?? memoryTrace?.agentRun ?? null,
|
||||
valuesPrinted: false
|
||||
};
|
||||
return compactTrace;
|
||||
}
|
||||
|
||||
function compactTraceEventCount(traceSummary, agentRun, memoryTrace) {
|
||||
const memoryEvents = Array.isArray(memoryTrace?.events) ? memoryTrace.events : [];
|
||||
const memoryHasProjection = memoryTrace?.lastEvent || memoryEvents.length > 0;
|
||||
const candidates = memoryHasProjection
|
||||
? [memoryTrace?.eventCount, traceSummary?.eventCount, traceSummary?.sourceEventCount, traceSummary?.lastSeq, agentRun?.lastSeq]
|
||||
: [traceSummary?.eventCount, traceSummary?.sourceEventCount, traceSummary?.lastSeq, agentRun?.lastSeq, memoryTrace?.eventCount];
|
||||
for (const value of candidates) {
|
||||
const count = Number(value);
|
||||
if (Number.isFinite(count) && count >= 0) return Math.trunc(count);
|
||||
}
|
||||
return Array.isArray(memoryTrace?.events) ? memoryTrace.events.length : 0;
|
||||
}
|
||||
|
||||
function firstTextValue(...values) {
|
||||
for (const value of values) {
|
||||
const text = textValue(value);
|
||||
if (text) return text;
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
||||
function optionalObject(value) {
|
||||
const object = objectValue(value);
|
||||
return Object.keys(object).length > 0 ? object : null;
|
||||
}
|
||||
|
||||
function sessionSummary(session, options) {
|
||||
if (!session?.id) return null;
|
||||
const snapshot = objectValue(session.session);
|
||||
const currentTurn = currentWorkbenchTurnProjection(session, options);
|
||||
const projectionDiagnostic = options.projectionDiagnostic ?? null;
|
||||
const messages = sessionMessages(session, { ...options, currentTurn });
|
||||
const status = currentTurn ? currentTurn.status : sessionLifecycleProjectionStatus(session);
|
||||
return {
|
||||
sessionId: session.id,
|
||||
threadId: safeOpaqueId(session.threadId) ?? (textValue(session.threadId) || null),
|
||||
agentId: session.agentId ?? "hwlab-code-agent",
|
||||
status,
|
||||
running: RUNNING_STATUSES.has(status),
|
||||
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
|
||||
lastTraceId: currentTurn ? currentTurn.traceId : null,
|
||||
projection: projectionDiagnostic,
|
||||
projectionStatus: projectionDiagnostic?.projectionStatus ?? null,
|
||||
projectionHealth: projectionDiagnostic?.projectionHealth ?? null,
|
||||
staleMs: projectionDiagnostic?.staleMs ?? null,
|
||||
blocker: projectionDiagnostic?.blocker ?? null,
|
||||
providerProfile: textValue(snapshot.providerProfile) || null,
|
||||
messageCount: messages.length,
|
||||
firstUserMessagePreview: firstUserPreview(messages),
|
||||
updatedAt: session.updatedAt ?? snapshot.updatedAt ?? null,
|
||||
turnSummary: currentTurn ? {
|
||||
turnId: currentTurn.turnId,
|
||||
traceId: currentTurn.traceId,
|
||||
status,
|
||||
running: RUNNING_STATUSES.has(status),
|
||||
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
|
||||
eventCount: currentTurn.eventCount,
|
||||
projection: projectionDiagnostic,
|
||||
projectionStatus: projectionDiagnostic?.projectionStatus ?? null,
|
||||
projectionHealth: projectionDiagnostic?.projectionHealth ?? null,
|
||||
staleMs: projectionDiagnostic?.staleMs ?? null,
|
||||
blocker: projectionDiagnostic?.blocker ?? null,
|
||||
updatedAt: currentTurn.updatedAt
|
||||
} : null,
|
||||
valuesRedacted: true
|
||||
};
|
||||
}
|
||||
|
||||
function sessionDetail(session, options) {
|
||||
return {
|
||||
...sessionSummary(session, options),
|
||||
metadata: {
|
||||
startedAt: session.startedAt ?? null,
|
||||
endedAt: session.endedAt ?? null,
|
||||
ownerUserId: session.ownerUserId ?? null
|
||||
},
|
||||
messagePageUrl: `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages`,
|
||||
valuesRedacted: true,
|
||||
secretMaterialStored: false
|
||||
};
|
||||
}
|
||||
|
||||
function sessionMessages(session, options = {}) {
|
||||
const snapshot = objectValue(session.session);
|
||||
const raw = Array.isArray(snapshot.messages) ? snapshot.messages : Array.isArray(snapshot.chatMessages) ? snapshot.chatMessages : [];
|
||||
const currentTurn = options.currentTurn ? options.currentTurn : currentWorkbenchTurnProjection(session, options);
|
||||
const finalTraceId = currentTurn ? currentTurn.traceId : null;
|
||||
const terminalProjection = terminalMessageProjection(snapshot, options, currentTurn);
|
||||
const messages = raw
|
||||
.map((message, index) => messageFact(message, index, session, snapshot))
|
||||
.filter(Boolean)
|
||||
.map((message) => applyTerminalMessageProjection(message, terminalProjection))
|
||||
.map((message) => applyProjectionMessageDiagnostic(message, options.projectionDiagnostic, currentTurn));
|
||||
const hasAssistantLikeFinal = messages.some((message) => isAssistantLikeRole(message.role) && (!finalTraceId || message.traceId === finalTraceId));
|
||||
const finalText = terminalProjection?.text ?? null;
|
||||
if (!hasAssistantLikeFinal && finalText) {
|
||||
const finalMessageStatus = terminalProjection ? terminalProjection.status : sessionLifecycleProjectionStatus(session);
|
||||
const finalMessageSource = terminalProjection ? terminalProjection.source : "finalResponse";
|
||||
messages.push(messageFact({ role: "assistant", text: finalText, status: finalMessageStatus, traceId: finalTraceId, source: finalMessageSource }, messages.length, session, snapshot));
|
||||
}
|
||||
return messages;
|
||||
}
|
||||
|
||||
function applyProjectionMessageDiagnostic(message, diagnostic, currentTurn) {
|
||||
if (!diagnostic || !isAssistantLikeRole(message.role)) return message;
|
||||
if (currentTurn?.traceId && message.traceId && message.traceId !== currentTurn.traceId) return message;
|
||||
const trace = message.runnerTrace && typeof message.runnerTrace === "object" ? message.runnerTrace : null;
|
||||
return {
|
||||
...message,
|
||||
projection: diagnostic,
|
||||
projectionStatus: diagnostic.projectionStatus ?? null,
|
||||
projectionHealth: diagnostic.projectionHealth ?? null,
|
||||
staleMs: diagnostic.staleMs ?? null,
|
||||
blocker: diagnostic.blocker ?? null,
|
||||
runnerTrace: trace ? { ...trace, projection: diagnostic, projectionStatus: diagnostic.projectionStatus ?? null, projectionHealth: diagnostic.projectionHealth ?? null, staleMs: diagnostic.staleMs ?? null, blocker: diagnostic.blocker ?? null } : trace
|
||||
};
|
||||
}
|
||||
|
||||
function terminalMessageProjection(snapshot, options, currentTurn) {
|
||||
if (!currentTurn) return null;
|
||||
const traceId = currentTurn.traceId;
|
||||
const status = currentTurn.status;
|
||||
if (!TERMINAL_STATUSES.has(status) || RUNNING_STATUSES.has(status)) return null;
|
||||
const text = projectionText(currentTurn.finalResponse);
|
||||
return { traceId, status, text, source: "turn-projection" };
|
||||
}
|
||||
|
||||
function applyTerminalMessageProjection(message, projection) {
|
||||
if (!projection || !isAssistantLikeRole(message.role)) return message;
|
||||
if (projection.traceId && message.traceId && message.traceId !== projection.traceId) return message;
|
||||
const text = projection.text || "";
|
||||
const parts = projectTerminalMessageParts(message.parts, message.messageId, projection.traceId ?? message.traceId, projection, text);
|
||||
return {
|
||||
...message,
|
||||
traceId: projection.traceId ?? message.traceId,
|
||||
turnId: message.turnId ?? projection.traceId ?? null,
|
||||
status: projection.status,
|
||||
parts,
|
||||
text,
|
||||
textPreview: text ? text.slice(0, 240) : message.textPreview ?? null,
|
||||
updatedAt: message.updatedAt ?? null
|
||||
};
|
||||
}
|
||||
|
||||
function projectTerminalMessageParts(parts, messageId, traceId, projection, text) {
|
||||
const sourceParts = Array.isArray(parts) ? parts : [];
|
||||
const textIndex = sourceParts.findIndex((part) => part?.type === "final_response" || part?.partType === "final_response");
|
||||
if (textIndex >= 0) {
|
||||
return sourceParts.map((part, index) => index === textIndex ? { ...part, traceId, type: "final_response", text: text || part.text || null, status: projection.status } : part);
|
||||
}
|
||||
if (!text) return sourceParts;
|
||||
return [partFact({ type: "final_response", text, status: projection.status }, 0, messageId, traceId), ...sourceParts];
|
||||
}
|
||||
|
||||
function projectionText(...values) {
|
||||
for (const value of values) {
|
||||
if (value && typeof value === "object") {
|
||||
@@ -2198,33 +1937,6 @@ function projectionText(...values) {
|
||||
return null;
|
||||
}
|
||||
|
||||
function messageFact(message, index, session, snapshot) {
|
||||
const role = textValue(message?.role) || (index % 2 === 0 ? "user" : "assistant");
|
||||
const traceId = safeTraceId(message?.traceId ?? message?.turnTraceId ?? session.lastTraceId ?? snapshot.lastTraceId) ?? null;
|
||||
const messageId = safeMessageId(message?.messageId ?? message?.id) || `msg_${hash(`${session.id}:${index}:${role}:${traceId ?? "none"}`).slice(0, 24)}`;
|
||||
const status = normalizeStatus(message?.status ?? (isAssistantLikeRole(role) ? session.status : "completed"));
|
||||
const legacyText = projectionText(message?.text, message?.content, message?.message, message?.finalResponse);
|
||||
const rawParts = Array.isArray(message?.parts) ? message.parts : [];
|
||||
const parts = rawParts.length > 0
|
||||
? rawParts.map((part, partIndex) => partFact(part, partIndex, messageId, traceId)).filter(Boolean)
|
||||
: !isAssistantLikeRole(role) && legacyText ? [partFact({ type: "text", text: legacyText }, 0, messageId, traceId)] : [];
|
||||
const text = factMessageAuthorityText({ role, status, parts, legacyText });
|
||||
return {
|
||||
messageId,
|
||||
role,
|
||||
sessionId: session.id,
|
||||
traceId,
|
||||
turnId: safeTurnId(message?.turnId) || traceId,
|
||||
status,
|
||||
parts,
|
||||
text: text || "",
|
||||
textPreview: text ? text.slice(0, 240) : null,
|
||||
createdAt: textValue(message?.createdAt ?? message?.timestamp ?? message?.at) || session.startedAt || session.updatedAt || null,
|
||||
updatedAt: textValue(message?.updatedAt) || session.updatedAt || null,
|
||||
valuesRedacted: message?.valuesRedacted !== false
|
||||
};
|
||||
}
|
||||
|
||||
function partFact(part, index, messageId, traceId) {
|
||||
const type = textValue(part?.type) || "text";
|
||||
const text = messageAuthorityTextValue(part?.text ?? part?.content ?? part?.message);
|
||||
@@ -2241,45 +1953,6 @@ function partFact(part, index, messageId, traceId) {
|
||||
};
|
||||
}
|
||||
|
||||
function turnSnapshot({ projection, result, session, trace }) {
|
||||
const turnId = projection.turnId;
|
||||
const traceId = projection.traceId;
|
||||
const status = projection.status;
|
||||
const messages = session ? sessionMessages(session, { result, trace }) : [];
|
||||
const userMessage = messages.find((message) => message.role === "user") ?? null;
|
||||
const assistantMessage = [...messages].reverse().find((message) => isAssistantLikeRole(message.role)) ?? null;
|
||||
const finalText = projectionText(projection.finalResponse);
|
||||
return {
|
||||
turnId,
|
||||
traceId,
|
||||
status,
|
||||
running: RUNNING_STATUSES.has(status),
|
||||
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
|
||||
sessionId: safeSessionId(result?.sessionId ?? result?.session?.sessionId ?? session?.id) ?? null,
|
||||
threadId: safeOpaqueId(result?.threadId ?? result?.session?.threadId ?? session?.threadId) ?? (textValue(session?.threadId) || null),
|
||||
userMessageId: userMessage?.messageId ?? null,
|
||||
assistantMessageId: assistantMessage?.messageId ?? null,
|
||||
assistantText: finalText ?? null,
|
||||
finalResponse: projection.finalResponse ?? null,
|
||||
timing: projection.timing ?? null,
|
||||
startedAt: projection.timing?.startedAt ?? null,
|
||||
lastEventAt: projection.timing?.lastEventAt ?? null,
|
||||
finishedAt: projection.timing?.finishedAt ?? null,
|
||||
durationMs: projection.timing?.durationMs ?? null,
|
||||
agentRun: result?.agentRun ?? null,
|
||||
trace: {
|
||||
traceId,
|
||||
status: trace?.status ?? "missing",
|
||||
eventCount: trace?.eventCount ?? 0,
|
||||
updatedAt: trace?.updatedAt ?? null
|
||||
},
|
||||
urls: {
|
||||
self: `/v1/workbench/turns/${encodeURIComponent(turnId)}`,
|
||||
traceEvents: `/v1/workbench/traces/${encodeURIComponent(traceId)}/events`
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function tracePageOptions(url) {
|
||||
const cursor = textValue(url.searchParams.get("cursor"));
|
||||
const cursorProjectedSeq = cursor.startsWith("projected:") ? Number.parseInt(cursor.slice("projected:".length), 10) : NaN;
|
||||
@@ -2289,8 +1962,31 @@ function tracePageOptions(url) {
|
||||
return { afterProjectedSeq, limit: boundedLimit(url.searchParams.get("limit")) };
|
||||
}
|
||||
|
||||
function traceSnapshotSync(options, traceId) {
|
||||
return (options.traceStore ?? defaultCodeAgentTraceStore).snapshot(traceId);
|
||||
async function realtimeInitialSessionContext(readModel, actor, sessionId) {
|
||||
const safeId = safeSessionId(sessionId);
|
||||
if (!safeId || typeof readModel?.queryFacts !== "function") {
|
||||
const blocker = traceEventsReadModelBlocker("workbench_facts_query_unavailable", "Workbench realtime initial session must be read from durable facts; the facts query interface is unavailable.", { session: { sessionId: safeId }, route: "/v1/workbench/events" });
|
||||
return { facts: emptyFactSet(), session: null, blocker };
|
||||
}
|
||||
try {
|
||||
const result = await readModel.queryFacts({ sessionId: safeId, families: WORKBENCH_SESSION_DETAIL_FAMILIES, limit: MAX_PAGE_LIMIT });
|
||||
if (result.error || result.durable === false) {
|
||||
const blocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench realtime initial session facts query failed; realtime must surface the read-model error instead of falling back to legacy session snapshots.", { session: { sessionId: safeId }, route: "/v1/workbench/events" });
|
||||
blocker.causeCode = result.error?.code ?? null;
|
||||
blocker.causeName = result.error?.name ?? null;
|
||||
blocker.message = result.error?.message ? `${blocker.message} Cause: ${result.error.message}` : result.durable === false ? `${blocker.message} Cause: durable facts reader unavailable.` : blocker.message;
|
||||
return { facts: emptyFactSet(), session: null, blocker };
|
||||
}
|
||||
const facts = result.facts ?? emptyFactSet();
|
||||
const session = visibleFactSessions(facts, actor).find((item) => factSessionId(item) === safeId) ?? null;
|
||||
return { facts, session, blocker: null };
|
||||
} catch (error) {
|
||||
const blocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench realtime initial session facts query failed; realtime must surface the read-model error instead of falling back to legacy session snapshots.", { session: { sessionId: safeId }, route: "/v1/workbench/events" });
|
||||
blocker.causeCode = error?.code ?? null;
|
||||
blocker.causeName = error?.name ?? null;
|
||||
blocker.message = error?.message ? `${blocker.message} Cause: ${error.message}` : blocker.message;
|
||||
return { facts: emptyFactSet(), session: null, blocker };
|
||||
}
|
||||
}
|
||||
|
||||
async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId, reason }) {
|
||||
@@ -2312,8 +2008,8 @@ async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId,
|
||||
error: context.factsReadBlocker
|
||||
});
|
||||
}
|
||||
const sessionId = safeSessionId(context.result?.sessionId ?? context.result?.session?.sessionId ?? context.session?.id) ?? null;
|
||||
const threadId = safeOpaqueId(context.result?.threadId ?? context.result?.session?.threadId ?? context.session?.threadId) ?? (textValue(context.session?.threadId) || null);
|
||||
const sessionId = factSessionId(context.factSession) ?? safeSessionId(context.session?.id) ?? null;
|
||||
const threadId = safeOpaqueId(context.factSession?.threadId ?? context.session?.threadId) ?? (textValue(context.factSession?.threadId ?? context.session?.threadId) || null);
|
||||
const realtimeTraceSnapshot = context.factsReadBlocker ? blockedRealtimeTraceSnapshot(context) : traceSnapshotForRealtime(context.trace);
|
||||
const traceSeq = context.factsReadBlocker ? 0 : traceSnapshotLastSeq(context.trace);
|
||||
writeEvent("workbench.trace.snapshot", {
|
||||
@@ -2438,38 +2134,59 @@ async function writeTraceRealtimeSnapshotSafe(input) {
|
||||
|
||||
async function visibleTraceContext(options, actor, traceId) {
|
||||
const readModel = createWorkbenchReadModel(options, actor);
|
||||
const result = readModel.resultForTrace(traceId);
|
||||
const session = await readModel.getSessionByTraceId(traceId);
|
||||
if (result?.ownerUserId && !readModel.canReadOwner(result.ownerUserId)) return { visible: false };
|
||||
const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId));
|
||||
if (!resultVisible && !session) return { visible: false };
|
||||
const trace = await readModel.traceSnapshot(traceId);
|
||||
const projection = createWorkbenchTurnProjection({ turnId: traceId, traceId, result, session, trace });
|
||||
const factSessionIdValue = safeSessionId(session?.id ?? result?.sessionId ?? result?.session?.sessionId) ?? null;
|
||||
if (typeof readModel.queryFacts !== "function") {
|
||||
const projection = factProjectionForTrace(emptyFactSet(), traceId);
|
||||
const factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_unavailable", "Workbench durable facts query is unavailable; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" });
|
||||
return { visible: true, turnId: traceId, traceId, status: "unknown", projection, result: null, session: null, trace: factTraceSnapshot(emptyFactSet(), traceId), facts: emptyFactSet(), factSession: null, factsReadBlocker };
|
||||
}
|
||||
|
||||
let facts = emptyFactSet();
|
||||
let factsReadBlocker = null;
|
||||
if (factSessionIdValue && typeof readModel.queryFacts === "function") {
|
||||
try {
|
||||
const factResult = await readModel.queryFacts({ sessionId: factSessionIdValue, families: WORKBENCH_SESSION_DETAIL_FAMILIES, limit: 100 });
|
||||
try {
|
||||
const factResult = await readModel.queryFacts({ traceId, families: [...WORKBENCH_SESSION_DETAIL_FAMILIES, ...WORKBENCH_TRACE_EVENT_PAGE_FAMILIES], limit: MAX_PAGE_LIMIT });
|
||||
if (factResult.error || factResult.durable === false) {
|
||||
const projection = factProjectionForTrace(facts, traceId);
|
||||
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench durable facts query failed; realtime must surface the read-model error instead of silently falling back to legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" });
|
||||
factsReadBlocker.causeCode = factResult.error?.code ?? null;
|
||||
factsReadBlocker.causeName = factResult.error?.name ?? null;
|
||||
factsReadBlocker.message = factResult.error?.message ? `${factsReadBlocker.message} Cause: ${factResult.error.message}` : factResult.durable === false ? `${factsReadBlocker.message} Cause: durable facts reader unavailable.` : factsReadBlocker.message;
|
||||
} else {
|
||||
facts = factResult?.facts ?? facts;
|
||||
} catch (error) {
|
||||
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench durable facts query failed; realtime must surface the read-model error instead of silently falling back to legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" });
|
||||
factsReadBlocker.causeCode = error?.code ?? null;
|
||||
factsReadBlocker.causeName = error?.name ?? null;
|
||||
factsReadBlocker.message = error?.message ? `${factsReadBlocker.message} Cause: ${error.message}` : factsReadBlocker.message;
|
||||
}
|
||||
} catch (error) {
|
||||
const projection = factProjectionForTrace(facts, traceId);
|
||||
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench durable facts query failed; realtime must surface the read-model error instead of silently falling back to legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" });
|
||||
factsReadBlocker.causeCode = error?.code ?? null;
|
||||
factsReadBlocker.causeName = error?.name ?? null;
|
||||
factsReadBlocker.message = error?.message ? `${factsReadBlocker.message} Cause: ${error.message}` : factsReadBlocker.message;
|
||||
}
|
||||
const factSession = factArray(facts.sessions).find((item) => factSessionId(item) === factSessionIdValue) ?? null;
|
||||
if (!factsReadBlocker) {
|
||||
if (!factSessionIdValue) {
|
||||
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_id_missing", "Workbench realtime could not resolve a durable session id; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" });
|
||||
} else if (typeof readModel.queryFacts !== "function") {
|
||||
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_unavailable", "Workbench durable facts query is unavailable; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" });
|
||||
} else if (!factSession) {
|
||||
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_missing", "Workbench durable facts did not contain the trace session; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" });
|
||||
}
|
||||
|
||||
const resolved = factsReadBlocker ? { session: null, facts } : await visibleFactSessionForTrace(readModel, facts, actor, traceId);
|
||||
if (resolved.error) {
|
||||
const projection = factProjectionForTrace(facts, traceId);
|
||||
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench durable facts session lookup failed; realtime must surface the read-model error instead of using legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" });
|
||||
factsReadBlocker.causeCode = resolved.error?.code ?? null;
|
||||
factsReadBlocker.causeName = resolved.error?.name ?? null;
|
||||
factsReadBlocker.message = resolved.error?.message ? `${factsReadBlocker.message} Cause: ${resolved.error.message}` : factsReadBlocker.message;
|
||||
} else if (resolved.facts) {
|
||||
facts = resolved.facts;
|
||||
}
|
||||
return { visible: true, turnId: projection.turnId, traceId, status: projection.status, projection, result, session, trace, facts, factSession, factsReadBlocker };
|
||||
|
||||
const factSession = resolved.session ?? null;
|
||||
const trace = factTraceSnapshot(facts, traceId);
|
||||
const turn = factTurnForTrace(facts, traceId, traceId);
|
||||
let projection = factsReadBlocker ? blockedTraceEventProjection(factProjectionForTrace(facts, traceId), factsReadBlocker) : factProjectionForTrace(facts, traceId);
|
||||
const visible = Boolean(factsReadBlocker || factSession || turn || factCheckpointForTrace(facts, traceId) || trace.eventCount > 0);
|
||||
if (!visible) {
|
||||
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_trace_missing", "Workbench realtime could not find durable facts for the requested trace; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" });
|
||||
projection = blockedTraceEventProjection(projection, factsReadBlocker);
|
||||
}
|
||||
if (!factsReadBlocker && !factSession) {
|
||||
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_missing", "Workbench durable facts did not contain the trace session; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, turn, route: "/v1/workbench/events" });
|
||||
projection = blockedTraceEventProjection(projection, factsReadBlocker);
|
||||
}
|
||||
const snapshot = factTurnSnapshot({ turn, session: factSession, facts, traceId, turnId: traceId });
|
||||
return { visible: true, turnId: snapshot.turnId ?? traceId, traceId, status: snapshot.status, projection, result: null, session: null, trace, facts, factSession, factsReadBlocker };
|
||||
}
|
||||
|
||||
function traceSnapshotForRealtime(snapshot) {
|
||||
@@ -2501,36 +2218,6 @@ function traceSnapshotLastSeq(snapshot) {
|
||||
return events.reduce((max, event, index) => Math.max(max, eventSeq(event, index)), 0);
|
||||
}
|
||||
|
||||
function currentWorkbenchTurnProjection(session, options = {}) {
|
||||
const traceId = safeTraceId(session?.lastTraceId) ?? null;
|
||||
if (!traceId) return null;
|
||||
const trace = options.trace?.traceId === traceId ? options.trace : traceSnapshotSync(options, traceId);
|
||||
const result = traceScopedResult(options, traceId);
|
||||
return createWorkbenchTurnProjection({ turnId: traceId, traceId, result, session, trace });
|
||||
}
|
||||
|
||||
function traceScopedResult(options, traceId) {
|
||||
const result = options.result?.traceId === traceId || (!options.result?.traceId && options.result) ? options.result : options.codeAgentChatResults?.get?.(traceId) ?? null;
|
||||
return result && typeof result === "object" ? result : null;
|
||||
}
|
||||
|
||||
function sessionLifecycleProjectionStatus(session) {
|
||||
return normalizeStatus(session?.status);
|
||||
}
|
||||
|
||||
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 handleWorkbenchReadModelFailure(response, url, options = {}, error) {
|
||||
const classified = classifyWorkbenchReadModelFailure(error);
|
||||
const route = workbenchReadRouteTemplate(url?.pathname ?? "/v1/workbench");
|
||||
|
||||
Reference in New Issue
Block a user