mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: backfill terminal workbench trace events

Browse Source
This commit is contained in:
lyon
2026-06-25 09:22:58 +08:00
parent 49d42fde14
commit f3967a5f6d
2 changed files with 229 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/workbench-projection-writer.test.ts
+89
View File
@@ -90,6 +90,95 @@ test("workbench projection writer commits terminal owner evidence as sealed dura
assert.equal(facts.checkpoints[0].timing.finishedAt, "2026-06-20T11:00:00.000Z");
});
test("workbench projection writer backfills terminal final and completion trace events from sealed facts", async () => {
const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:20:00.000Z" });
const accessController = {
async recordAgentSessionOwner(input) {
return {
id: input.sessionId,
projectId: input.projectId,
ownerUserId: input.ownerUserId,
conversationId: input.conversationId,
threadId: input.threadId,
lastTraceId: input.traceId,
status: input.status,
session: input.session,
updatedAt: "2026-06-20T11:20:00.000Z"
};
}
};
for (const [sourceSeq, label] of [
[1, "agentrun:backend:admitted"],
[2, "agentrun:run:created"],
[3, "agentrun:runner-job:queued"]
]) {
await writeWorkbenchProjectionEvent({
runtimeStore,
event: {
traceId: "trc_writer_terminal_backfill",
sessionId: "ses_writer_terminal_backfill",
source: "agentrun",
sourceSeq,
type: "backend",
eventType: "backend",
status: "running",
label,
runId: "run_writer_terminal_backfill",
commandId: "cmd_writer_terminal_backfill",
createdAt: `2026-06-20T11:19:0${sourceSeq}.000Z`
}
});
}
await writeWorkbenchProjectionSession({
accessController,
runtimeStore,
traceId: "trc_writer_terminal_backfill",
ownerUserId: "usr_writer",
ownerRole: "user",
sessionId: "ses_writer_terminal_backfill",
projectId: "prj_writer",
conversationId: "cnv_writer_terminal_backfill",
threadId: "thread-writer-terminal-backfill",
status: "completed",
payload: {
traceId: "trc_writer_terminal_backfill",
status: "completed",
finalResponse: { text: "BACKFILL_DONE", status: "completed", traceId: "trc_writer_terminal_backfill" },
runnerTrace: {
traceId: "trc_writer_terminal_backfill",
status: "completed",
startedAt: "2026-06-20T11:19:01.000Z",
lastEventAt: "2026-06-20T11:19:13.000Z",
finishedAt: "2026-06-20T11:19:13.000Z",
eventCount: 3,
events: []
},
agentRun: { runId: "run_writer_terminal_backfill", commandId: "cmd_writer_terminal_backfill", status: "completed", terminalStatus: "completed", lastSeq: 3 },
updatedAt: "2026-06-20T11:19:13.000Z"
},
session: {
sessionStatus: "completed",
messages: [
{ messageId: "msg_writer_backfill_user", role: "user", text: "question", status: "sent", turnId: "trc_writer_terminal_backfill", traceId: "trc_writer_terminal_backfill" },
{ messageId: "msg_writer_backfill_agent", role: "agent", text: "BACKFILL_DONE", status: "completed", turnId: "trc_writer_terminal_backfill", traceId: "trc_writer_terminal_backfill" }
],
finalResponse: { text: "BACKFILL_DONE", status: "completed", traceId: "trc_writer_terminal_backfill" }
}
});
const loaded = runtimeStore.queryWorkbenchFacts({ traceId: "trc_writer_terminal_backfill", families: ["traceEvents", "checkpoints"], limit: 20 });
const events = loaded.facts.traceEvents;
assert.deepEqual(events.map((event) => event.projectedSeq), [1, 2, 3, 4, 5]);
assert.deepEqual(events.slice(-2).map((event) => event.label), ["agentrun:assistant:message", "agentrun:result:completed"]);
assert.equal(events.at(-2).text, "BACKFILL_DONE");
assert.equal(events.at(-1).terminal, true);
assert.equal(events.at(-1).sealed, true);
assert.equal(loaded.facts.checkpoints[0].projectedSeq, 5);
assert.equal(loaded.facts.checkpoints[0].projectionStatus, "caught_up");
});
test("workbench projection writer does not rewrite prior-turn messages from merged owner session", async () => {
const factWrites = [];
const runtimeStore = {
internal/cloud/workbench-projection-writer.ts
+140 -1
View File
@@ -324,14 +324,16 @@ async function writeWorkbenchProjectionFacts({ runtimeStore = null, traceStore =
projectId,
valuesPrinted: false
});
const terminalTraceBackfill = await backfillTerminalTraceEvents({ runtimeStore, traceStore, traceId: safeId, facts, payload, status });
emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "ok", {
...workbenchProjectionFactCountAttributes(facts),
"workbench.projection.terminal_trace_backfill_written": terminalTraceBackfill?.written ?? 0,
"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;
return terminalTraceBackfill ? { ...result, terminalTraceBackfill } : result;
} catch (error) {
const safeId = safeTraceId(traceId);
emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "error", {
@@ -353,6 +355,143 @@ async function writeWorkbenchProjectionFacts({ runtimeStore = null, traceStore =
}
}
async function backfillTerminalTraceEvents({ runtimeStore = null, traceStore = defaultCodeAgentTraceStore, traceId = null, facts = {}, payload = null, status = null } = {}) {
if (typeof runtimeStore?.queryWorkbenchFacts !== "function" || typeof runtimeStore?.allocateWorkbenchProjectedSeq !== "function") return null;
const safeId = safeTraceId(traceId ?? facts?.checkpoints?.[0]?.traceId ?? facts?.turns?.[0]?.traceId);
const checkpoint = facts?.checkpoints?.[0] ?? null;
const turn = facts?.turns?.[0] ?? null;
if (!safeId || checkpoint?.terminal !== true || turn?.terminal !== true) return null;
const terminalStatus = normalizeWorkbenchStatus(turn?.status ?? checkpoint?.status ?? status);
if (terminalStatus !== "completed") return null;
const finalText = finalResponseTextValue(turn?.finalResponse, checkpoint?.finalResponse, turn?.assistantText, checkpoint?.assistantText, payload?.finalResponse, payload?.assistantText);
if (!finalText) return null;
const sessionId = textValue(checkpoint?.sessionId ?? facts?.sessions?.[0]?.sessionId) || null;
const turnId = textValue(turn?.turnId ?? checkpoint?.turnId) || safeId;
const runId = textValue(checkpoint?.runId ?? payload?.agentRun?.runId) || null;
const commandId = textValue(checkpoint?.commandId ?? payload?.agentRun?.commandId) || null;
const assistantMessageId = textValue(turn?.messageId ?? facts?.messages?.find((message) => message.role !== "user" && message.traceId === safeId)?.messageId) || null;
try {
const existing = await runtimeStore.queryWorkbenchFacts({ traceId: safeId, families: ["traceEvents"], limit: 500 });
const events = Array.isArray(existing?.facts?.traceEvents) ? existing.facts.traceEvents : [];
const missingAssistantFinal = !events.some((event) => terminalAssistantTraceEventMatches(event, finalText));
const missingCompletion = !events.some((event) => completedTerminalTraceEvent(event));
if (!missingAssistantFinal && !missingCompletion) return { written: 0, missingAssistantFinal: false, missingCompletion: false, valuesPrinted: false };
const timestamp = latestTimestamp(turn?.finishedAt, checkpoint?.finishedAt, turn?.lastEventAt, checkpoint?.lastEventAt, turn?.updatedAt, checkpoint?.updatedAt, payload?.updatedAt) ?? new Date().toISOString();
const baseSourceSeq = Math.max(nonNegativeInteger(checkpoint?.sourceSeq), nonNegativeInteger(turn?.sourceSeq), ...events.map((event) => nonNegativeInteger(event?.sourceSeq)));
let offset = 1;
let written = 0;
if (missingAssistantFinal) {
await writeWorkbenchProjectionEvent({
runtimeStore,
event: {
traceId: safeId,
sessionId,
turnId,
messageId: assistantMessageId,
source: "workbench-terminal-projection",
sourceSeq: baseSourceSeq + offset,
sourceEventId: `${safeId}:workbench-terminal-projection:assistant-final`,
type: "assistant",
eventType: "assistant",
status: "completed",
label: "agentrun:assistant:message",
message: finalText,
text: finalText,
finalResponse: { text: finalText, status: "completed", traceId: safeId, valuesPrinted: false },
final: true,
replyAuthority: true,
terminal: true,
runId,
commandId,
timing: checkpoint?.timing,
startedAt: checkpoint?.startedAt,
lastEventAt: checkpoint?.lastEventAt,
finishedAt: checkpoint?.finishedAt,
durationMs: checkpoint?.durationMs,
createdAt: timestamp,
occurredAt: timestamp,
updatedAt: timestamp,
valuesPrinted: false
}
});
written += 1;
offset += 1;
}
if (missingCompletion || missingAssistantFinal) {
await writeWorkbenchProjectionEvent({
runtimeStore,
event: {
traceId: safeId,
sessionId,
turnId,
messageId: assistantMessageId,
source: "workbench-terminal-projection",
sourceSeq: baseSourceSeq + offset,
sourceEventId: `${safeId}:workbench-terminal-projection:completion`,
type: "result",
eventType: "terminal",
status: "completed",
label: "agentrun:result:completed",
message: "AgentRun result is ready for HWLAB short-connection polling.",
finalResponse: { text: finalText, status: "completed", traceId: safeId, valuesPrinted: false },
terminal: true,
sealed: true,
runId,
commandId,
timing: checkpoint?.timing,
startedAt: checkpoint?.startedAt,
lastEventAt: checkpoint?.lastEventAt,
finishedAt: checkpoint?.finishedAt,
durationMs: checkpoint?.durationMs,
createdAt: timestamp,
occurredAt: timestamp,
updatedAt: timestamp,
valuesPrinted: false
}
});
written += 1;
}
return { written, missingAssistantFinal, missingCompletion, valuesPrinted: false };
} catch (error) {
emitWorkbenchProjectionOtel("workbench.projection.terminal_trace_backfill", safeId, "error", {
"workbench.projection.phase": "terminal-trace-backfill",
"workbench.projection.session_id": sessionId,
"workbench.projection.turn_id": turnId
}, error);
appendProjectionDiagnostic(traceStore, safeId, {
type: "facts",
status: "degraded",
label: "projection-writer:terminal-trace-backfill-failed",
errorCode: error?.code ?? "workbench_terminal_trace_backfill_failed",
message: error?.message ?? "Workbench terminal trace event backfill failed.",
terminal: false,
valuesPrinted: false
});
return { written: 0, errorCode: error?.code ?? "workbench_terminal_trace_backfill_failed", valuesPrinted: false };
}
}
function terminalAssistantTraceEventMatches(event = {}, finalText = null) {
if (!isAssistantTraceEvent(event)) return false;
const text = finalResponseTextValue(event?.finalResponse, event?.text, event?.message, event?.content, event?.payload?.text, event?.payload?.message);
if (!text || text !== finalText) return false;
return event?.terminal === true || event?.final === true || event?.replyAuthority === true || normalizeWorkbenchStatus(event?.status) === "completed";
}
function completedTerminalTraceEvent(event = {}) {
const status = normalizeWorkbenchStatus(event?.status ?? event?.terminalStatus ?? event?.payload?.terminalStatus);
if (status !== "completed") return false;
if (event?.terminal === true) return true;
const label = textValue(event?.label ?? event?.type ?? event?.eventType).toLowerCase();
return /result:completed|terminal:completed|turn:completed/u.test(label);
}
function isAssistantTraceEvent(event = {}) {
const type = textValue(event?.type ?? event?.eventType).toLowerCase();
if (type === "assistant" || type === "assistant_message") return true;
return /assistant:message|assistant_message/u.test(textValue(event?.label).toLowerCase());
}
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;