Lyon
836 lines
39 KiB
TypeScript
836 lines
39 KiB
TypeScript
/*
|
|
* SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; draft-2026-06-20-p1-zero-split-durable-realtime; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0.
|
|
* 职责: WorkbenchProjectionWriter 组件入口。唯一封装 Code Agent/AgentRun facts 到 Workbench session projection facts 的持久化写入。
|
|
*/
|
|
import { createHash } from "node:crypto";
|
|
|
|
import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
|
|
import { emitCodeAgentOtelSpan } from "./otel-trace.ts";
|
|
import { safeTraceId } from "./server-http-utils.ts";
|
|
import { createWorkbenchTurnProjection, normalizeWorkbenchStatus, projectionDiagnostics, TERMINAL_STATUSES } from "./workbench-turn-projection.ts";
|
|
|
|
export async function writeWorkbenchProjectionSession({ accessController, runtimeStore = null, traceStore = defaultCodeAgentTraceStore, traceId, ownerUserId, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, payload = null, params = {}, preserveLastTraceId = false } = {}) {
|
|
if (!ownerUserId || typeof accessController?.recordAgentSessionOwner !== "function") return null;
|
|
const safeId = safeTraceId(traceId);
|
|
try {
|
|
const ownerRecord = await accessController.recordAgentSessionOwner({
|
|
ownerUserId,
|
|
ownerRole,
|
|
sessionId,
|
|
projectId,
|
|
conversationId,
|
|
threadId,
|
|
traceId: preserveLastTraceId ? undefined : safeId,
|
|
status,
|
|
session
|
|
});
|
|
const factsWrite = await writeWorkbenchProjectionFacts({
|
|
runtimeStore,
|
|
traceStore,
|
|
traceId: safeId,
|
|
ownerUserId,
|
|
ownerRole,
|
|
sessionId: ownerRecord?.id ?? sessionId,
|
|
projectId: ownerRecord?.projectId ?? projectId,
|
|
conversationId: ownerRecord?.conversationId ?? conversationId,
|
|
threadId: ownerRecord?.threadId ?? threadId,
|
|
status: ownerRecord?.status ?? status,
|
|
session: ownerRecord?.session ?? session,
|
|
payload,
|
|
params
|
|
});
|
|
if (!factsWrite) {
|
|
throw workbenchProjectionError("workbench_projection_facts_not_written", "Workbench projection facts were not durably written.");
|
|
}
|
|
return ownerRecord;
|
|
} catch (error) {
|
|
emitWorkbenchProjectionOtel("workbench.projection.session.persist", safeId, "error", {
|
|
"workbench.projection.phase": "session-owner",
|
|
"workbench.projection.has_owner_user_id": Boolean(ownerUserId),
|
|
"workbench.projection.has_session_id": Boolean(sessionId)
|
|
}, error);
|
|
if (safeId) appendProjectionDiagnostic(traceStore, safeId, {
|
|
type: "session-owner",
|
|
status: "degraded",
|
|
label: "projection-writer:session-owner:persist-failed",
|
|
errorCode: "workbench_projection_session_persist_failed",
|
|
message: error?.message ?? "Workbench projection session persistence failed.",
|
|
valuesPrinted: false
|
|
});
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
export async function writeWorkbenchSessionAdmissionFact({ runtimeStore = null, session = null, ownerUserId = null, ownerRole = null, projectId = null, conversationId = null, threadId = null, status = "idle", now = new Date().toISOString() } = {}) {
|
|
if (typeof runtimeStore?.writeWorkbenchFacts !== "function" || !session || typeof session !== "object") return null;
|
|
const sessionId = textValue(session.id ?? session.sessionId);
|
|
if (!sessionId) return null;
|
|
const createdAt = timestampValue(session.startedAt ?? session.createdAt ?? session.session?.createdAt ?? now);
|
|
const updatedAt = timestampValue(session.updatedAt ?? createdAt);
|
|
const normalizedStatus = normalizeWorkbenchStatus(session.status ?? session.session?.sessionStatus ?? status) || "idle";
|
|
const fact = {
|
|
sessionId,
|
|
ownerUserId: textValue(session.ownerUserId ?? ownerUserId) || null,
|
|
ownerRole: textValue(session.ownerRole ?? ownerRole) || null,
|
|
projectId: textValue(session.projectId ?? projectId) || null,
|
|
agentId: textValue(session.agentId) || "hwlab-code-agent",
|
|
conversationId: textValue(session.conversationId ?? conversationId) || null,
|
|
threadId: textValue(session.threadId ?? threadId) || null,
|
|
status: normalizedStatus,
|
|
lastTraceId: safeTraceId(session.lastTraceId ?? session.traceId) || null,
|
|
projectedSeq: 0,
|
|
sourceSeq: 0,
|
|
sourceEventId: `${sessionId}:session-admission`,
|
|
terminal: false,
|
|
sealed: false,
|
|
sessionJson: {
|
|
...(session.session && typeof session.session === "object" ? session.session : {}),
|
|
sessionStatus: normalizedStatus,
|
|
source: textValue(session.session?.source) || "manual-session-create",
|
|
providerProfile: textValue(session.session?.providerProfile) || null,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
},
|
|
createdAt,
|
|
updatedAt,
|
|
valuesRedacted: true
|
|
};
|
|
return runtimeStore.writeWorkbenchFacts({ facts: { sessions: [fact] } }, {
|
|
sessionId,
|
|
ownerUserId: fact.ownerUserId,
|
|
projectId: fact.projectId,
|
|
valuesPrinted: false
|
|
});
|
|
}
|
|
|
|
export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event = {}, requestMeta = {} } = {}) {
|
|
if (typeof runtimeStore?.writeWorkbenchFacts !== "function") return null;
|
|
if (typeof runtimeStore?.allocateWorkbenchProjectedSeq !== "function") throw new Error("Workbench projection event writer requires a durable projectedSeq allocator.");
|
|
const traceId = safeTraceId(event.traceId ?? requestMeta.traceId);
|
|
if (!traceId) return null;
|
|
const projectedAt = new Date().toISOString();
|
|
const sourceSeq = nonNegativeInteger(event.sourceSeq ?? event.seq);
|
|
const sessionId = textValue(event.sessionId ?? requestMeta.sessionId) || null;
|
|
const turnId = textValue(event.turnId) || traceId;
|
|
const terminal = event.terminal === true || TERMINAL_STATUSES.has(normalizeWorkbenchStatus(event.status));
|
|
const status = terminal ? normalizeWorkbenchStatus(event.status) : "running";
|
|
const previousCheckpoint = await latestWorkbenchCheckpoint(runtimeStore, traceId);
|
|
const previousTiming = normalizeTimingProjection(previousCheckpoint?.timing) ?? normalizeTimingProjection(previousCheckpoint);
|
|
const sourceOccurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt ?? projectedAt);
|
|
const occurredAt = latestTimestamp(previousTiming?.lastEventAt, sourceOccurredAt) ?? sourceOccurredAt;
|
|
const previousTerminal = checkpointIsTerminal(previousCheckpoint);
|
|
const suppressedAfterSeal = previousTerminal && !terminal;
|
|
const explicitEventTiming = normalizeTimingProjection(event.timing) ?? normalizeTimingProjection(event);
|
|
const explicitDurationMs = durationValue(event.durationMs ?? explicitEventTiming?.durationMs);
|
|
const explicitStartedAt = optionalTimestampValue(event.startedAt ?? event.traceStartedAt ?? event.runnerStartedAt ?? explicitEventTiming?.startedAt);
|
|
// Once a trace has a durable checkpoint, startedAt is sealed as the card
|
|
// elapsed-time source. Later trace/result hydration can carry older or newer
|
|
// startedAt values, but it must not make the visible Code Agent card jump by
|
|
// switching to a different clock origin.
|
|
const provisionalLastEventAt = latestTimestamp(previousTiming?.lastEventAt, explicitEventTiming?.lastEventAt, occurredAt);
|
|
const provisionalFinishedAt = terminal ? latestTimestamp(explicitEventTiming?.finishedAt, optionalTimestampValue(event.finishedAt), provisionalLastEventAt, occurredAt) : null;
|
|
const startedAt = previousTiming?.startedAt ?? explicitStartedAt ?? startedAtFromDuration(provisionalFinishedAt, explicitDurationMs) ?? (sourceSeq <= 1 ? occurredAt : null);
|
|
const lastEventAt = provisionalLastEventAt;
|
|
const finishedAt = terminal ? latestTimestamp(provisionalFinishedAt, lastEventAt) : null;
|
|
const timing = eventTimingProjection({ startedAt, lastEventAt, finishedAt, terminal, durationMs: explicitDurationMs ?? previousTiming?.durationMs });
|
|
const timingAuthorityIssue = terminalTimingAuthorityIssue(timing, { terminal, traceId, source: "event", status, label: event.label, sourceSeq });
|
|
if (timingAuthorityIssue) emitTerminalTimingAuthorityDiagnostic(defaultCodeAgentTraceStore, traceId, timingAuthorityIssue, {
|
|
"workbench.projection.phase": "event-timing",
|
|
"workbench.projection.session_id": sessionId,
|
|
"workbench.projection.turn_id": turnId,
|
|
"workbench.projection.source_seq": sourceSeq
|
|
});
|
|
const eventTiming = suppressedAfterSeal ? previousTiming ?? timing : timing;
|
|
const sourceEventId = workbenchSourceEventId(event, { traceId, sourceSeq, occurredAt });
|
|
const allocation = await runtimeStore.allocateWorkbenchProjectedSeq({
|
|
traceId,
|
|
sessionId,
|
|
turnId,
|
|
runId: textValue(event.runId) || null,
|
|
commandId: textValue(event.commandId) || null,
|
|
sourceSeq,
|
|
sourceEventId,
|
|
eventType: textValue(event.eventType ?? event.type ?? event.label) || "event",
|
|
occurredAt,
|
|
updatedAt: projectedAt
|
|
}, {
|
|
traceId,
|
|
sessionId,
|
|
agentSessionId: requestMeta.agentSessionId ?? sessionId,
|
|
valuesPrinted: false
|
|
});
|
|
const projectedSeq = nonNegativeInteger(allocation?.projectedSeq);
|
|
if (projectedSeq <= 0) throw new Error("Workbench projection allocator returned an invalid projectedSeq.");
|
|
const eventId = textValue(event.id) || stableFactId("wte", { traceId, sourceEventId });
|
|
const previousFinalText = finalResponseTextValue(previousCheckpoint?.finalResponse, previousCheckpoint?.assistantText, previousCheckpoint?.finalText);
|
|
const messageFinalText = terminal ? finalResponseTextValue(event.finalResponse, event.assistantText, event.reply, previousFinalText) : null;
|
|
const userMessageFact = sessionId && !suppressedAfterSeal && previousCheckpoint?.userMessage ? normalizeMessageFact(previousCheckpoint.userMessage, 0, { traceId, sessionId, turnId, terminal: false, terminalStatus: "sent", finalText: null, timestamp: projectedAt, timing }) : null;
|
|
const messageFact = sessionId && !suppressedAfterSeal ? normalizeMessageFact({
|
|
messageId: eventProjectionAssistantMessageId(traceId, event),
|
|
role: "agent",
|
|
status,
|
|
traceId,
|
|
turnId,
|
|
text: terminal && messageFinalText ? messageFinalText : "",
|
|
projectedSeq,
|
|
sourceSeq,
|
|
sourceEventId,
|
|
terminal,
|
|
sealed: terminal,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
createdAt: occurredAt,
|
|
updatedAt: projectedAt,
|
|
source: "workbench-event-projection"
|
|
}, userMessageFact ? 1 : 0, { traceId, sessionId, turnId, terminal, terminalStatus: status, finalText: messageFinalText, timestamp: projectedAt, timing }) : null;
|
|
const facts = {
|
|
sessions: sessionId && !suppressedAfterSeal ? [{
|
|
sessionId,
|
|
status,
|
|
lastTraceId: traceId,
|
|
projectedSeq,
|
|
sourceSeq,
|
|
sourceEventId,
|
|
terminal,
|
|
sealed: terminal,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
updatedAt: projectedAt
|
|
}] : [],
|
|
traceEvents: [{
|
|
...event,
|
|
id: eventId,
|
|
traceId,
|
|
sessionId,
|
|
turnId,
|
|
messageId: textValue(event.messageId) || null,
|
|
sourceSeq,
|
|
sourceEventId,
|
|
projectedSeq,
|
|
eventType: textValue(event.eventType ?? event.type ?? event.label) || "event",
|
|
terminal,
|
|
sealed: terminal,
|
|
timing: eventTiming,
|
|
startedAt: eventTiming?.startedAt ?? null,
|
|
lastEventAt: eventTiming?.lastEventAt ?? null,
|
|
finishedAt: eventTiming?.finishedAt ?? null,
|
|
durationMs: eventTiming?.durationMs ?? null,
|
|
suppressedAfterSeal,
|
|
sourceOccurredAt: sourceOccurredAt !== occurredAt ? sourceOccurredAt : null,
|
|
createdAt: occurredAt,
|
|
occurredAt,
|
|
updatedAt: projectedAt
|
|
}],
|
|
messages: [userMessageFact, messageFact].filter(Boolean),
|
|
parts: [userMessageFact, messageFact].filter(Boolean).flatMap((message) => messagePartFacts(message, { finalText: message === messageFact ? messageFinalText : null })),
|
|
checkpoints: !suppressedAfterSeal ? [{
|
|
traceId,
|
|
sessionId,
|
|
turnId,
|
|
runId: textValue(event.runId) || null,
|
|
commandId: textValue(event.commandId) || null,
|
|
projectedSeq,
|
|
sourceSeq,
|
|
sourceEventId,
|
|
projectionStatus: terminal ? "caught_up" : "projecting",
|
|
projectionHealth: "healthy",
|
|
terminal,
|
|
sealed: terminal,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
diagnostic: {
|
|
lastEventLabel: textValue(event.label) || null,
|
|
lastEventStatus: textValue(event.status) || null,
|
|
timingAuthorityIssue,
|
|
valuesRedacted: true
|
|
},
|
|
updatedAt: projectedAt
|
|
}] : []
|
|
};
|
|
try {
|
|
const result = await runtimeStore.writeWorkbenchFacts({ facts }, {
|
|
traceId,
|
|
sessionId,
|
|
agentSessionId: requestMeta.agentSessionId ?? sessionId,
|
|
valuesPrinted: false
|
|
});
|
|
emitWorkbenchProjectionOtel("workbench.projection.event.write", traceId, "ok", {
|
|
...workbenchProjectionFactCountAttributes(facts),
|
|
"workbench.projection.phase": terminal ? "event-terminal" : "event-running",
|
|
"workbench.projection.terminal": terminal,
|
|
"workbench.projection.suppressed_after_seal": suppressedAfterSeal,
|
|
"workbench.projection.session_id": sessionId,
|
|
"workbench.projection.turn_id": turnId
|
|
});
|
|
return result;
|
|
} catch (error) {
|
|
emitWorkbenchProjectionOtel("workbench.projection.event.write", traceId, "error", {
|
|
...workbenchProjectionFactCountAttributes(facts),
|
|
"workbench.projection.phase": terminal ? "event-terminal" : "event-running",
|
|
"workbench.projection.terminal": terminal,
|
|
"workbench.projection.suppressed_after_seal": suppressedAfterSeal,
|
|
"workbench.projection.session_id": sessionId,
|
|
"workbench.projection.turn_id": turnId
|
|
}, error);
|
|
appendProjectionDiagnostic(defaultCodeAgentTraceStore, traceId, {
|
|
type: "facts",
|
|
status: "degraded",
|
|
label: "projection-writer:event-facts:persist-failed",
|
|
errorCode: error?.code ?? "workbench_event_facts_persist_failed",
|
|
message: error?.message ?? "Workbench event facts persistence failed.",
|
|
terminal: false,
|
|
valuesPrinted: false
|
|
});
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
function eventProjectionAssistantMessageId(traceId, event = {}) {
|
|
const explicit = textValue(event.messageId ?? event.assistantMessageId ?? event.assistantMessage?.messageId);
|
|
if (explicit) return explicit;
|
|
const traceSuffix = (safeTraceId(traceId) || String(traceId || "trace"))
|
|
.replace(/^trc_/u, "")
|
|
.replace(/[^A-Za-z0-9_.:-]/gu, "_")
|
|
.slice(0, 48) || "trace";
|
|
return `msg_${traceSuffix}_agent`;
|
|
}
|
|
|
|
async function writeWorkbenchProjectionFacts({ runtimeStore = null, traceStore = defaultCodeAgentTraceStore, traceId = null, ownerUserId = null, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, payload = null, params = {} } = {}) {
|
|
if (typeof runtimeStore?.writeWorkbenchFacts !== "function") return null;
|
|
const safeId = safeTraceId(traceId ?? session?.traceId ?? payload?.traceId ?? params?.traceId);
|
|
const previousCheckpoint = safeId ? await latestWorkbenchCheckpoint(runtimeStore, safeId) : null;
|
|
const facts = buildWorkbenchProjectionFacts({ traceId, ownerUserId, ownerRole, sessionId, projectId, conversationId, threadId, status, session, payload, params, previousCheckpoint });
|
|
if (isEmptyFacts(facts)) return null;
|
|
const timingAuthorityIssue = facts.checkpoints?.[0]?.diagnostic?.timingAuthorityIssue ?? null;
|
|
if (timingAuthorityIssue) emitTerminalTimingAuthorityDiagnostic(traceStore, safeId, timingAuthorityIssue, {
|
|
"workbench.projection.phase": "facts-timing",
|
|
"workbench.projection.session_id": facts.sessions[0]?.sessionId ?? sessionId ?? null,
|
|
"workbench.projection.turn_id": facts.turns[0]?.turnId ?? null
|
|
});
|
|
try {
|
|
const result = await runtimeStore.writeWorkbenchFacts({ facts }, {
|
|
traceId: facts.checkpoints[0]?.traceId ?? traceId,
|
|
sessionId: facts.sessions[0]?.sessionId ?? sessionId,
|
|
ownerUserId,
|
|
projectId,
|
|
valuesPrinted: false
|
|
});
|
|
emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "ok", {
|
|
...workbenchProjectionFactCountAttributes(facts),
|
|
"workbench.projection.phase": "session-facts",
|
|
"workbench.projection.terminal": facts.turns.some((turn) => turn.terminal === true),
|
|
"workbench.projection.session_id": facts.sessions[0]?.sessionId ?? sessionId ?? null,
|
|
"workbench.projection.turn_id": facts.turns[0]?.turnId ?? null
|
|
});
|
|
return result;
|
|
} catch (error) {
|
|
const safeId = safeTraceId(traceId);
|
|
emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "error", {
|
|
...workbenchProjectionFactCountAttributes(facts),
|
|
"workbench.projection.phase": "session-facts",
|
|
"workbench.projection.terminal": facts.turns.some((turn) => turn.terminal === true),
|
|
"workbench.projection.session_id": facts.sessions[0]?.sessionId ?? sessionId ?? null,
|
|
"workbench.projection.turn_id": facts.turns[0]?.turnId ?? null
|
|
}, error);
|
|
if (safeId) appendProjectionDiagnostic(traceStore, safeId, {
|
|
type: "facts",
|
|
status: "degraded",
|
|
label: "projection-writer:facts:persist-failed",
|
|
errorCode: error?.code ?? "workbench_facts_persist_failed",
|
|
message: error?.message ?? "Workbench facts persistence failed.",
|
|
terminal: false
|
|
});
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, payload = null, params = {}, previousCheckpoint = null } = {}) {
|
|
const safeId = safeTraceId(traceId ?? session?.traceId ?? payload?.traceId ?? params?.traceId);
|
|
const resolvedSessionId = textValue(sessionId ?? session?.sessionId ?? payload?.sessionId ?? params?.sessionId) || null;
|
|
const resolvedConversationId = textValue(conversationId ?? session?.conversationId ?? payload?.conversationId ?? params?.conversationId) || null;
|
|
const resolvedThreadId = textValue(threadId ?? session?.threadId ?? payload?.threadId ?? params?.threadId) || null;
|
|
const agentRun = payload?.agentRun ?? session?.agentRun ?? null;
|
|
const timestamp = timestampValue(payload?.updatedAt ?? session?.updatedAt ?? payload?.createdAt ?? session?.createdAt);
|
|
const normalizedStatus = normalizeWorkbenchStatus(payload?.status ?? session?.sessionStatus ?? status);
|
|
const projection = createWorkbenchTurnProjection({ traceId: safeId, result: payload, session: { id: resolvedSessionId, status: normalizedStatus, session }, trace: payload?.runnerTrace ?? null });
|
|
const projectedStatus = normalizeWorkbenchStatus(projection.status ?? normalizedStatus);
|
|
const terminal = projection.terminal === true;
|
|
const terminalStatus = terminal ? (TERMINAL_STATUSES.has(projectedStatus) ? projectedStatus : normalizedStatus) : projectedStatus;
|
|
const projectedAt = new Date().toISOString();
|
|
const timing = terminalTimingAtLeastProjectedAt(projectionTimingForStatus(projection.timing, terminal), terminal, projectedAt);
|
|
const timingAuthorityIssue = terminalTimingAuthorityIssue(timing, { terminal, traceId: safeId, source: "facts", status: terminalStatus, label: payload?.lastEventLabel, sourceSeq: projection.lastProjectedSeq });
|
|
const baseDiagnostic = projectionDiagnostics({ traceId: safeId, result: payload, trace: payload?.runnerTrace ?? null, projection });
|
|
const diagnostic = timingAuthorityIssue
|
|
? { ...baseDiagnostic, projectionHealth: baseDiagnostic.projectionHealth === "unavailable" ? "unavailable" : "degraded", blocker: baseDiagnostic.blocker ?? timingAuthorityIssue, timingAuthorityIssue, valuesRedacted: true }
|
|
: baseDiagnostic;
|
|
const previousFinalText = finalResponseTextValue(previousCheckpoint?.finalResponse, previousCheckpoint?.assistantText, previousCheckpoint?.finalText);
|
|
const finalText = terminal ? finalResponseTextValue(projection.finalResponse, payload?.finalResponse, session?.finalResponse, payload?.assistantText, previousFinalText) : null;
|
|
const projectedFinalResponse = terminal && finalText ? { text: finalText, status: terminalStatus, traceId: safeId, valuesPrinted: false } : null;
|
|
const messages = normalizeMessages(workbenchProjectionInputMessages({ session, payload, params, traceId: safeId, timestamp, terminal, terminalStatus, finalText }), {
|
|
traceId: safeId,
|
|
sessionId: resolvedSessionId,
|
|
conversationId: resolvedConversationId,
|
|
threadId: resolvedThreadId,
|
|
terminal,
|
|
terminalStatus,
|
|
finalText,
|
|
timestamp,
|
|
timing
|
|
});
|
|
return {
|
|
sessions: resolvedSessionId ? [{
|
|
sessionId: resolvedSessionId,
|
|
ownerUserId,
|
|
ownerRole,
|
|
projectId: projectId ?? session?.projectId ?? payload?.projectId ?? params?.projectId ?? null,
|
|
conversationId: resolvedConversationId,
|
|
threadId: resolvedThreadId,
|
|
status: terminal ? terminalStatus : "running",
|
|
lastTraceId: safeId,
|
|
projectedSeq: nonNegativeInteger(projection.lastProjectedSeq),
|
|
sourceSeq: nonNegativeInteger(projection.lastProjectedSeq),
|
|
sourceEventId: safeId,
|
|
terminal,
|
|
sealed: terminal,
|
|
sessionJson: session,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
createdAt: timestampValue(session?.createdAt ?? payload?.createdAt ?? timestamp),
|
|
updatedAt: timestamp
|
|
}] : [],
|
|
messages,
|
|
parts: messages.flatMap((message) => messagePartFacts(message, { finalText: message.role !== "user" ? finalText : null })),
|
|
turns: safeId && resolvedSessionId ? [{
|
|
turnId: projection.turnId ?? safeId,
|
|
sessionId: resolvedSessionId,
|
|
traceId: safeId,
|
|
messageId: messages.find((message) => message.role !== "user")?.messageId ?? null,
|
|
status: projection.status ?? normalizedStatus,
|
|
projectedSeq: nonNegativeInteger(projection.lastProjectedSeq),
|
|
sourceSeq: nonNegativeInteger(projection.lastProjectedSeq),
|
|
sourceEventId: safeId,
|
|
terminal: projection.terminal,
|
|
sealed: projection.terminal,
|
|
finalResponse: projectedFinalResponse,
|
|
diagnostic,
|
|
assistantText: finalText,
|
|
userMessage: previousCheckpoint?.userMessage ?? null,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
updatedAt: timestamp
|
|
}] : [],
|
|
checkpoints: safeId ? [{
|
|
traceId: safeId,
|
|
sessionId: resolvedSessionId,
|
|
turnId: projection.turnId ?? safeId,
|
|
runId: textValue(agentRun?.runId ?? projection.sourceRunId) || null,
|
|
commandId: textValue(agentRun?.commandId ?? projection.sourceCommandId) || null,
|
|
projectedSeq: nonNegativeInteger(projection.lastProjectedSeq),
|
|
sourceSeq: nonNegativeInteger(agentRun?.lastSeq ?? projection.lastProjectedSeq),
|
|
sourceEventId: safeId,
|
|
projectionStatus: projection.terminal ? "caught_up" : "projecting",
|
|
projectionHealth: diagnostic.projectionHealth === "unavailable" ? "unavailable" : diagnostic.projectionHealth === "stalled" ? "stalled" : "healthy",
|
|
terminal: projection.terminal,
|
|
sealed: projection.terminal,
|
|
diagnostic,
|
|
finalResponse: projectedFinalResponse,
|
|
assistantText: finalText,
|
|
userMessage: messages.find((message) => message.role === "user") ?? null,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
updatedAt: timestamp
|
|
}] : []
|
|
};
|
|
}
|
|
|
|
function workbenchProjectionError(code, message) {
|
|
const error = new Error(message);
|
|
error.code = code;
|
|
error.valuesRedacted = true;
|
|
return error;
|
|
}
|
|
|
|
function emitWorkbenchProjectionOtel(name, traceId, status = "ok", attributes = {}, error = null) {
|
|
void emitCodeAgentOtelSpan(name, safeTraceId(traceId) ?? "trc_unassigned", process.env, {
|
|
status,
|
|
error,
|
|
attributes: {
|
|
...attributes,
|
|
"workbench.projection.zero_implicit_fallback": true,
|
|
"workbench.projection.error_visible": status === "error"
|
|
}
|
|
});
|
|
}
|
|
|
|
function workbenchProjectionFactCountAttributes(facts = {}) {
|
|
return {
|
|
"workbench.facts.sessions": Array.isArray(facts.sessions) ? facts.sessions.length : 0,
|
|
"workbench.facts.messages": Array.isArray(facts.messages) ? facts.messages.length : 0,
|
|
"workbench.facts.parts": Array.isArray(facts.parts) ? facts.parts.length : 0,
|
|
"workbench.facts.turns": Array.isArray(facts.turns) ? facts.turns.length : 0,
|
|
"workbench.facts.trace_events": Array.isArray(facts.traceEvents) ? facts.traceEvents.length : 0,
|
|
"workbench.facts.checkpoints": Array.isArray(facts.checkpoints) ? facts.checkpoints.length : 0
|
|
};
|
|
}
|
|
|
|
function workbenchProjectionInputMessages({ session = {}, payload = {}, params = {}, traceId = null, timestamp = null, terminal = false, terminalStatus = "completed", finalText = null } = {}) {
|
|
if (Array.isArray(session?.messages) && session.messages.length > 0) return session.messages;
|
|
if (Array.isArray(payload?.messages) && payload.messages.length > 0) return payload.messages;
|
|
if (!traceId) return [];
|
|
const turnId = textValue(payload?.turnId) || traceId;
|
|
const assistantMessageId = textValue(payload?.assistantMessageId ?? payload?.messageId ?? payload?.assistantMessage?.messageId) || stableFactId("msg", { traceId, role: "agent" });
|
|
const createdAt = timestampValue(timestamp ?? payload?.createdAt ?? payload?.updatedAt);
|
|
const messages = [];
|
|
const userMessage = workbenchProjectionUserMessage({ payload, params, traceId, turnId, timestamp: createdAt });
|
|
if (userMessage) messages.push(userMessage);
|
|
messages.push({
|
|
messageId: assistantMessageId,
|
|
role: "agent",
|
|
status: terminal ? terminalStatus : "running",
|
|
text: terminal && finalText ? finalText : "",
|
|
traceId,
|
|
turnId,
|
|
timing: payload?.timing,
|
|
startedAt: payload?.startedAt,
|
|
lastEventAt: payload?.lastEventAt,
|
|
finishedAt: payload?.finishedAt,
|
|
durationMs: payload?.durationMs,
|
|
createdAt,
|
|
updatedAt: createdAt,
|
|
valuesPrinted: false
|
|
});
|
|
return messages;
|
|
}
|
|
|
|
function workbenchProjectionUserMessage({ payload = {}, params = {}, traceId = null, turnId = null, timestamp = null } = {}) {
|
|
const userText = textValue(params?.message ?? params?.prompt ?? payload?.prompt ?? payload?.message);
|
|
if (!traceId || !userText) return null;
|
|
const createdAt = timestampValue(timestamp ?? payload?.createdAt ?? payload?.updatedAt);
|
|
return {
|
|
messageId: textValue(payload?.userMessageId ?? payload?.userMessage?.messageId) || stableFactId("msg", { traceId, role: "user" }),
|
|
role: "user",
|
|
status: "sent",
|
|
text: userText,
|
|
traceId,
|
|
turnId: textValue(turnId) || traceId,
|
|
createdAt,
|
|
updatedAt: createdAt,
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
|
|
function normalizeMessages(messages, context) {
|
|
const normalized = Array.isArray(messages) ? messages.map((message, index) => normalizeMessageFact(message, index, context)).filter(Boolean) : [];
|
|
if (context?.terminal && context?.finalText && context?.traceId && !normalized.some((message) => message.role !== "user" && message.traceId === context.traceId)) {
|
|
const projected = normalizeMessageFact({ role: "agent", status: context.terminalStatus ?? "completed", text: context.finalText, traceId: context.traceId, source: "workbench-terminal-projection" }, normalized.length, context);
|
|
if (projected) normalized.push(projected);
|
|
}
|
|
return normalized;
|
|
}
|
|
|
|
function messagePartFacts(message = {}, { finalText = null } = {}) {
|
|
const messageId = textValue(message.messageId);
|
|
if (!messageId || !message.sessionId) return [];
|
|
const role = textValue(message.role) || "agent";
|
|
const terminal = message.terminal === true;
|
|
const finalResponseText = role !== "user" && terminal ? finalResponseTextValue(finalText, message.finalResponse) : null;
|
|
const text = finalResponseText ?? textValue(message.text);
|
|
if (!text) return [];
|
|
const partType = finalResponseText ? "final_response" : "text";
|
|
return [{
|
|
partId: stableFactId("wpt", { messageId, index: 0, partType }),
|
|
messageId,
|
|
sessionId: message.sessionId,
|
|
turnId: message.turnId,
|
|
traceId: message.traceId,
|
|
partIndex: 0,
|
|
partType,
|
|
status: message.status,
|
|
projectedSeq: message.projectedSeq,
|
|
sourceSeq: message.sourceSeq,
|
|
sourceEventId: message.sourceEventId,
|
|
terminal,
|
|
sealed: finalResponseText ? true : message.sealed,
|
|
text,
|
|
createdAt: message.createdAt,
|
|
updatedAt: message.updatedAt,
|
|
valuesPrinted: false
|
|
}];
|
|
}
|
|
|
|
function normalizeMessageFact(message = {}, index, { traceId, sessionId, conversationId, threadId, terminal, terminalStatus = "completed", finalText = null, timestamp, timing: contextTiming }) {
|
|
const messageId = textValue(message.messageId ?? message.id) || stableFactId("msg", { traceId, index, role: message.role });
|
|
if (!sessionId || !messageId) return null;
|
|
const role = textValue(message.role) || "agent";
|
|
const messageTraceId = safeTraceId(message.traceId ?? message.turnId);
|
|
const appliesToContextTrace = role !== "user" && Boolean(traceId && (!messageTraceId || messageTraceId === traceId));
|
|
const contextTerminalStatus = normalizeWorkbenchStatus(terminalStatus) || "completed";
|
|
const status = appliesToContextTrace && terminal ? contextTerminalStatus : normalizeWorkbenchStatus(message.status ?? (terminal && role !== "user" ? contextTerminalStatus : role === "user" ? "sent" : "running"));
|
|
const messageTerminal = role !== "user" && TERMINAL_STATUSES.has(status);
|
|
const messageTiming = normalizeTimingProjection(message.timing) ?? normalizeTimingProjection(message);
|
|
const contextTimingProjection = appliesToContextTrace ? normalizeTimingProjection(contextTiming) : null;
|
|
const timing = role === "user" ? messageTiming ?? emptyTimingProjection() : contextTimingProjection ?? messageTiming ?? emptyTimingProjection();
|
|
const terminalCompleted = messageTerminal && status === "completed";
|
|
const text = appliesToContextTrace && terminal && finalText ? finalText : textValue(message.text ?? message.content ?? message.message);
|
|
return {
|
|
...message,
|
|
messageId,
|
|
sessionId,
|
|
conversationId,
|
|
threadId,
|
|
turnId: textValue(message.turnId) || traceId,
|
|
traceId: safeTraceId(message.traceId ?? traceId) ?? traceId,
|
|
role,
|
|
status,
|
|
projectedSeq: nonNegativeInteger(message.projectedSeq ?? message.seq ?? index + 1),
|
|
sourceSeq: nonNegativeInteger(message.sourceSeq ?? message.seq ?? index + 1),
|
|
sourceEventId: textValue(message.sourceEventId) || `${messageId}:0`,
|
|
terminal: messageTerminal,
|
|
sealed: messageTerminal,
|
|
text,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
createdAt: timestampValue(message.createdAt ?? timestamp),
|
|
updatedAt: timestampValue(message.updatedAt ?? timestamp),
|
|
...(terminalCompleted ? { error: null, blocker: null, projection: null, projectionStatus: null, projectionHealth: null, staleMs: null } : {}),
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
|
|
function finalResponseTextValue(...values) {
|
|
for (const value of values) {
|
|
if (value && typeof value === "object") {
|
|
const nested = finalResponseTextValue(value.text, value.content, value.message, value.summary, value.preview);
|
|
if (nested) return nested;
|
|
continue;
|
|
}
|
|
const text = textValue(value);
|
|
if (text) return text;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
async function latestWorkbenchCheckpoint(runtimeStore, traceId) {
|
|
if (typeof runtimeStore?.queryWorkbenchFacts !== "function") return null;
|
|
const result = await runtimeStore.queryWorkbenchFacts({ traceId, families: ["checkpoints"], limit: 1 });
|
|
return Array.isArray(result?.facts?.checkpoints) ? result.facts.checkpoints[0] ?? null : null;
|
|
}
|
|
|
|
function projectionTimingForStatus(value, terminal) {
|
|
const timing = normalizeTimingProjection(value) ?? emptyTimingProjection();
|
|
if (terminal) return timing;
|
|
return { ...timing, finishedAt: null, durationMs: null, valuesRedacted: timing.valuesRedacted !== false };
|
|
}
|
|
|
|
function terminalTimingAtLeastProjectedAt(value, terminal, projectedAt) {
|
|
const timing = normalizeTimingProjection(value) ?? emptyTimingProjection();
|
|
if (!terminal) return timing;
|
|
const finishedAt = timing.finishedAt ?? timing.lastEventAt ?? optionalTimestampValue(projectedAt);
|
|
return {
|
|
...timing,
|
|
finishedAt,
|
|
durationMs: elapsedBetween(timing.startedAt, finishedAt) ?? timing.durationMs,
|
|
valuesRedacted: timing.valuesRedacted !== false
|
|
};
|
|
}
|
|
|
|
function checkpointIsTerminal(value) {
|
|
const source = value && typeof value === "object" ? value : null;
|
|
if (!source) return false;
|
|
if (source.terminal === true || source.sealed === true) return true;
|
|
const status = normalizeWorkbenchStatus(source.status ?? source.terminalStatus ?? source.traceStatus);
|
|
return TERMINAL_STATUSES.has(status);
|
|
}
|
|
|
|
function eventTimingProjection({ startedAt = null, lastEventAt = null, finishedAt = null, terminal = false, durationMs = null } = {}) {
|
|
const normalizedStartedAt = optionalTimestampValue(startedAt);
|
|
const normalizedLastEventAt = optionalTimestampValue(lastEventAt);
|
|
const normalizedFinishedAt = terminal ? optionalTimestampValue(finishedAt ?? normalizedLastEventAt) : null;
|
|
const normalizedDurationMs = durationValue(durationMs);
|
|
return {
|
|
startedAt: normalizedStartedAt,
|
|
lastEventAt: normalizedLastEventAt,
|
|
finishedAt: normalizedFinishedAt,
|
|
durationMs: terminal ? elapsedBetween(normalizedStartedAt, normalizedFinishedAt ?? normalizedLastEventAt) ?? normalizedDurationMs : null,
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
function normalizeTimingProjection(value) {
|
|
const source = value && typeof value === "object" ? value : null;
|
|
if (!source) return null;
|
|
const startedAt = optionalTimestampValue(source.startedAt);
|
|
const lastEventAt = optionalTimestampValue(source.lastEventAt);
|
|
const finishedAt = optionalTimestampValue(source.finishedAt);
|
|
const durationMs = durationValue(source.durationMs);
|
|
if (!startedAt && !lastEventAt && !finishedAt && durationMs === null) return null;
|
|
return { ...source, startedAt, lastEventAt, finishedAt, durationMs, valuesRedacted: source.valuesRedacted !== false };
|
|
}
|
|
|
|
function emptyTimingProjection() {
|
|
return { startedAt: null, lastEventAt: null, finishedAt: null, durationMs: null, valuesRedacted: true };
|
|
}
|
|
|
|
function durationValue(value) {
|
|
const number = Number(value);
|
|
return Number.isFinite(number) && number >= 0 ? Math.trunc(number) : null;
|
|
}
|
|
|
|
function elapsedBetween(startedAt, endedAt) {
|
|
const start = Date.parse(String(startedAt ?? ""));
|
|
const end = Date.parse(String(endedAt ?? ""));
|
|
if (!Number.isFinite(start) || !Number.isFinite(end) || end < start) return null;
|
|
return Math.trunc(end - start);
|
|
}
|
|
|
|
function startedAtFromDuration(finishedAt, durationMs) {
|
|
const end = Date.parse(String(finishedAt ?? ""));
|
|
const duration = durationValue(durationMs);
|
|
if (!Number.isFinite(end) || duration === null) return null;
|
|
return new Date(Math.max(0, end - duration)).toISOString();
|
|
}
|
|
|
|
function terminalTimingAuthorityIssue(timing, { terminal = false, traceId = null, source = null, status = null, label = null, sourceSeq = null } = {}) {
|
|
if (!terminal) return null;
|
|
const startedAt = optionalTimestampValue(timing?.startedAt);
|
|
const finishedAt = optionalTimestampValue(timing?.finishedAt);
|
|
const durationMs = durationValue(timing?.durationMs);
|
|
if (startedAt && finishedAt && durationMs !== null) return null;
|
|
return {
|
|
code: "workbench_terminal_timing_authority_missing",
|
|
layer: "workbench-projection",
|
|
category: "timing-authority",
|
|
traceId,
|
|
source,
|
|
status,
|
|
label: textValue(label) || null,
|
|
sourceSeq: nonNegativeInteger(sourceSeq),
|
|
missingStartedAt: !startedAt,
|
|
missingFinishedAt: !finishedAt,
|
|
missingDurationMs: durationMs === null,
|
|
retryable: true,
|
|
message: "Workbench terminal projection is missing durable timing authority; the UI must surface this diagnostic instead of hiding elapsed time.",
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
|
|
function emitTerminalTimingAuthorityDiagnostic(traceStore, traceId, issue, attributes = {}) {
|
|
if (!issue) return null;
|
|
emitWorkbenchProjectionOtel("workbench.projection.terminal_timing_authority", traceId, "error", {
|
|
...attributes,
|
|
"workbench.projection.timing_authority_missing": true,
|
|
"workbench.projection.missing_started_at": issue.missingStartedAt === true,
|
|
"workbench.projection.missing_finished_at": issue.missingFinishedAt === true,
|
|
"workbench.projection.missing_duration_ms": issue.missingDurationMs === true
|
|
}, workbenchProjectionError(issue.code, issue.message));
|
|
return appendProjectionDiagnostic(traceStore, traceId, {
|
|
type: "projection-timing",
|
|
status: "degraded",
|
|
label: "projection-writer:terminal-timing:missing-authority",
|
|
errorCode: issue.code,
|
|
message: issue.message,
|
|
terminal: false,
|
|
timingAuthorityIssue: issue,
|
|
valuesPrinted: false
|
|
});
|
|
}
|
|
|
|
function latestTimestamp(...values) {
|
|
let latest = null;
|
|
let latestMs = Number.NEGATIVE_INFINITY;
|
|
for (const value of values) {
|
|
const timestamp = optionalTimestampValue(value);
|
|
const ms = Date.parse(String(timestamp ?? ""));
|
|
if (!Number.isFinite(ms) || ms < latestMs) continue;
|
|
latest = timestamp;
|
|
latestMs = ms;
|
|
}
|
|
return latest;
|
|
}
|
|
|
|
function isEmptyFacts(facts) {
|
|
return Object.values(facts).every((items) => !Array.isArray(items) || items.length === 0);
|
|
}
|
|
|
|
function stableFactId(prefix, value) {
|
|
return `${prefix}_${createHash("sha256").update(JSON.stringify(sortJson(value))).digest("hex").slice(0, 32)}`;
|
|
}
|
|
|
|
function workbenchSourceEventId(event = {}, { traceId, sourceSeq, occurredAt } = {}) {
|
|
const explicit = textValue(event.sourceEventId ?? event.id);
|
|
if (explicit) return explicit;
|
|
const source = textValue(event.source);
|
|
const label = textValue(event.label ?? event.type ?? event.eventType) || "event";
|
|
if (source && sourceSeq > 0) return `${source}:${sourceSeq}:${label}`;
|
|
return stableFactId("wte_source", {
|
|
traceId,
|
|
source: source || null,
|
|
sourceSeq,
|
|
label,
|
|
eventType: textValue(event.eventType ?? event.type) || null,
|
|
status: textValue(event.status) || null,
|
|
itemId: textValue(event.itemId ?? event.messageId) || null,
|
|
runId: textValue(event.runId) || null,
|
|
commandId: textValue(event.commandId) || null,
|
|
occurredAt
|
|
});
|
|
}
|
|
|
|
function sortJson(value) {
|
|
if (Array.isArray(value)) return value.map(sortJson);
|
|
if (value && typeof value === "object") return Object.fromEntries(Object.keys(value).sort().map((key) => [key, sortJson(value[key])]));
|
|
return value;
|
|
}
|
|
|
|
function nonNegativeInteger(value) {
|
|
const parsed = Number.parseInt(value ?? "", 10);
|
|
return Number.isInteger(parsed) && parsed >= 0 ? parsed : 0;
|
|
}
|
|
|
|
function timestampValue(value) {
|
|
const ms = Date.parse(String(value ?? ""));
|
|
return Number.isFinite(ms) ? new Date(ms).toISOString() : new Date().toISOString();
|
|
}
|
|
|
|
function optionalTimestampValue(value) {
|
|
const ms = Date.parse(String(value ?? ""));
|
|
return Number.isFinite(ms) ? new Date(ms).toISOString() : null;
|
|
}
|
|
|
|
function textValue(value) {
|
|
return String(value ?? "").trim();
|
|
}
|
|
|
|
export function appendProjectionDiagnostic(traceStore = defaultCodeAgentTraceStore, traceId, event = {}) {
|
|
const safeId = safeTraceId(traceId);
|
|
if (!safeId || typeof traceStore?.append !== "function") return null;
|
|
return traceStore.append(safeId, {
|
|
type: "projection-diagnostic",
|
|
status: "degraded",
|
|
label: "projection:diagnostic",
|
|
...event,
|
|
valuesPrinted: false
|
|
});
|
|
}
|