Lyon
573 lines
25 KiB
TypeScript
573 lines
25 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 { 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
|
|
});
|
|
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
|
|
});
|
|
return ownerRecord;
|
|
} catch (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
|
|
});
|
|
return null;
|
|
}
|
|
}
|
|
|
|
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 occurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt ?? projectedAt);
|
|
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 explicitStartedAt = optionalTimestampValue(event.startedAt ?? event.traceStartedAt ?? event.runnerStartedAt);
|
|
const startedAt = explicitStartedAt ?? previousTiming?.startedAt ?? (sourceSeq <= 1 ? occurredAt : null);
|
|
const lastEventAt = latestTimestamp(previousTiming?.lastEventAt, occurredAt, projectedAt);
|
|
const finishedAt = terminal ? latestTimestamp(previousTiming?.lastEventAt, optionalTimestampValue(event.finishedAt), occurredAt, projectedAt) : null;
|
|
const timing = eventTimingProjection({ startedAt, lastEventAt, finishedAt, terminal });
|
|
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 messageFinalText = finalResponseTextValue(event.finalResponse, event.assistantText, event.reply);
|
|
const messageFact = sessionId ? 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"
|
|
}, 0, { traceId, sessionId, turnId, terminal, terminalStatus: status, finalText: messageFinalText, timestamp: projectedAt, timing }) : null;
|
|
const facts = {
|
|
sessions: sessionId ? [{
|
|
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,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
occurredAt,
|
|
updatedAt: projectedAt
|
|
}],
|
|
messages: messageFact ? [messageFact] : [],
|
|
checkpoints: [{
|
|
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,
|
|
valuesRedacted: true
|
|
},
|
|
updatedAt: projectedAt
|
|
}]
|
|
};
|
|
return runtimeStore.writeWorkbenchFacts({ facts }, {
|
|
traceId,
|
|
sessionId,
|
|
agentSessionId: requestMeta.agentSessionId ?? sessionId,
|
|
valuesPrinted: false
|
|
});
|
|
}
|
|
|
|
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 facts = buildWorkbenchProjectionFacts({ traceId, ownerUserId, ownerRole, sessionId, projectId, conversationId, threadId, status, session, payload, params });
|
|
if (isEmptyFacts(facts)) return null;
|
|
try {
|
|
return await runtimeStore.writeWorkbenchFacts({ facts }, {
|
|
traceId: facts.checkpoints[0]?.traceId ?? traceId,
|
|
sessionId: facts.sessions[0]?.sessionId ?? sessionId,
|
|
ownerUserId,
|
|
projectId,
|
|
valuesPrinted: false
|
|
});
|
|
} catch (error) {
|
|
const safeId = safeTraceId(traceId);
|
|
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
|
|
});
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, payload = null, params = {} } = {}) {
|
|
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 timing = projectionTimingForStatus(projection.timing, terminal);
|
|
const diagnostic = projectionDiagnostics({ traceId: safeId, result: payload, trace: payload?.runnerTrace ?? null, projection });
|
|
const finalText = finalResponseTextValue(projection.finalResponse, payload?.finalResponse, session?.finalResponse, payload?.assistantText);
|
|
const messages = normalizeMessages(session?.messages, {
|
|
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) => message.text ? [{
|
|
partId: stableFactId("wpt", { messageId: message.messageId, index: 0 }),
|
|
messageId: message.messageId,
|
|
sessionId: message.sessionId,
|
|
turnId: message.turnId,
|
|
traceId: message.traceId,
|
|
partIndex: 0,
|
|
partType: "text",
|
|
status: message.status,
|
|
projectedSeq: message.projectedSeq,
|
|
sourceSeq: message.sourceSeq,
|
|
sourceEventId: message.sourceEventId,
|
|
terminal: message.terminal,
|
|
sealed: message.sealed,
|
|
text: message.text,
|
|
updatedAt: message.updatedAt
|
|
}] : []),
|
|
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: projection.finalResponse,
|
|
diagnostic,
|
|
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,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
updatedAt: timestamp
|
|
}] : []
|
|
};
|
|
}
|
|
|
|
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 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 eventTimingProjection({ startedAt = null, lastEventAt = null, finishedAt = null, terminal = false } = {}) {
|
|
const normalizedStartedAt = optionalTimestampValue(startedAt);
|
|
const normalizedLastEventAt = optionalTimestampValue(lastEventAt);
|
|
const normalizedFinishedAt = terminal ? optionalTimestampValue(finishedAt ?? normalizedLastEventAt) : null;
|
|
return {
|
|
startedAt: normalizedStartedAt,
|
|
lastEventAt: normalizedLastEventAt,
|
|
finishedAt: normalizedFinishedAt,
|
|
durationMs: terminal ? elapsedBetween(normalizedStartedAt, normalizedFinishedAt ?? normalizedLastEventAt) : 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 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
|
|
});
|
|
}
|