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

fix: aggregate workbench terminal projection rows (#2153)

Browse Source 
Co-authored-by: root <root@lyon.remote>
This commit is contained in:
Lyon
2026-06-25 19:29:57 +08:00
committed by GitHub
parent f137f683fa
commit 5f3b9f7fad
2 changed files with 153 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/workbench-projection-writer.test.ts
+97 -12
View File
@@ -196,7 +196,7 @@ test("workbench projection writer keeps event writes durable when checkpoint rea
assert.equal(factWrites[0].params.facts.checkpoints[0].projectionStatus, "projecting");
});
test("workbench projection writer keeps terminal session facts durable and skips unsafe trace backfill when reads are blocked", async () => {
test("workbench projection writer keeps terminal session facts durable without trace backfill queries", async () => {
const factWrites = [];
let allocationSeq = 0;
let queryCount = 0;
@@ -261,7 +261,7 @@ test("workbench projection writer keeps terminal session facts durable and skips
});
assert.equal(owner.id, "ses_writer_blocked_backfill");
assert.equal(queryCount, 2);
assert.equal(queryCount, 1);
assert.equal(factWrites.length, 1);
assert.equal(factWrites[0].params.facts.checkpoints[0].traceId, "trc_writer_blocked_backfill");
assert.equal(factWrites[0].params.facts.turns[0].finalResponse.text, "BLOCKED_BACKFILL_DONE");
@@ -269,7 +269,7 @@ test("workbench projection writer keeps terminal session facts durable and skips
assert.equal(allocationSeq, 0);
});
test("workbench projection writer backfills terminal final and completion trace events from sealed facts", async () => {
test("workbench projection writer does not synthesize terminal trace events from sealed facts", async () => {
const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:20:00.000Z" });
const accessController = {
async recordAgentSessionOwner(input) {
@@ -347,15 +347,14 @@ test("workbench projection writer backfills terminal final and completion trace
}
});
const loaded = runtimeStore.queryWorkbenchFacts({ traceId: "trc_writer_terminal_backfill", families: ["traceEvents", "checkpoints"], limit: 20 });
const loaded = runtimeStore.queryWorkbenchFacts({ traceId: "trc_writer_terminal_backfill", families: ["traceEvents", "turns", "messages"], 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");
assert.deepEqual(events.map((event) => event.projectedSeq), [1, 2, 3]);
assert.deepEqual(events.map((event) => event.label), ["agentrun:backend:admitted", "agentrun:run:created", "agentrun:runner-job:queued"]);
assert.equal(events.some((event) => event.source === "workbench-terminal-projection"), false);
const terminalTurn = loaded.facts.turns.find((turn) => turn.terminal === true);
assert.equal(terminalTurn.finalResponse.text, "BACKFILL_DONE");
assert.equal(loaded.facts.messages.some((message) => message.role !== "user" && message.text === "BACKFILL_DONE"), true);
});
test("workbench projection writer does not duplicate terminal trace events that already exist", async () => {
@@ -462,6 +461,85 @@ test("workbench projection writer does not duplicate terminal trace events that
assert.equal(loaded.facts.checkpoints[0].projectedSeq, 3);
});
test("workbench projection event writer aggregates assistant updates and skips late non-terminal rows after seal", async () => {
const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:32:00.000Z" });
const traceId = "trc_writer_assistant_aggregate";
const sessionId = "ses_writer_assistant_aggregate";
await writeWorkbenchProjectionEvent({
runtimeStore,
event: {
traceId,
sessionId,
source: "agentrun",
sourceSeq: 1,
type: "backend",
eventType: "backend",
status: "running",
label: "agentrun:backend:running",
createdAt: "2026-06-20T11:31:00.000Z"
}
});
await writeWorkbenchProjectionEvent({
runtimeStore,
event: {
traceId,
sessionId,
source: "agentrun",
sourceSeq: 25,
type: "assistant",
eventType: "assistant",
status: "running",
label: "agentrun:assistant:message",
text: "AGGREGATED_FINAL",
createdAt: "2026-06-20T11:31:10.000Z"
}
});
await writeWorkbenchProjectionEvent({
runtimeStore,
event: {
traceId,
sessionId,
source: "agentrun",
sourceSeq: 26,
type: "assistant",
eventType: "assistant",
status: "completed",
label: "agentrun:assistant:message",
text: "AGGREGATED_FINAL",
finalResponse: { text: "AGGREGATED_FINAL", status: "completed", traceId },
terminal: true,
createdAt: "2026-06-20T11:31:12.000Z"
}
});
await writeWorkbenchProjectionEvent({
runtimeStore,
event: {
traceId,
sessionId,
source: "agentrun",
sourceSeq: 24,
type: "assistant",
eventType: "assistant",
status: "running",
label: "agentrun:assistant:message",
text: "AGGREGATED_FINAL",
createdAt: "2026-06-20T11:31:09.000Z"
}
});
const loaded = runtimeStore.queryWorkbenchFacts({ traceId, families: ["traceEvents", "messages", "checkpoints"], limit: 20 });
const events = loaded.facts.traceEvents;
assert.equal(events.length, 2);
assert.deepEqual(events.map((event) => event.projectedSeq), [1, 2]);
assert.equal(events[1].sourceEventId, `${traceId}:assistant-message:msg_writer_assistant_aggregate_agent`);
assert.equal(events[1].status, "completed");
assert.equal(events[1].text, "AGGREGATED_FINAL");
assert.equal(events.some((event) => event.sourceSeq === 24), false);
assert.equal(loaded.facts.messages.filter((message) => message.role !== "user").length, 1);
assert.equal(loaded.facts.checkpoints[0].terminal, true);
});
test("workbench projection writer does not rewrite prior-turn messages from merged owner session", async () => {
const factWrites = [];
const runtimeStore = {
@@ -662,7 +740,14 @@ test("workbench projection writer does not replace prior user prompt with cancel
agentRun: { runId: "run_writer_canceled_prompt_preserved", commandId: "cmd_writer_canceled_prompt_preserved", status: "cancelled", commandState: "cancelled", terminalStatus: "cancelled", lastSeq: 12 },
updatedAt: "2026-06-20T11:04:30.000Z"
},
session: { sessionStatus: "canceled", lastTraceId: traceId }
session: {
sessionStatus: "canceled",
lastTraceId: traceId,
messages: [
{ messageId: "msg_writer_canceled_prompt_preserved_user", role: "user", text: "当前 AgentRun 请求已取消；traceId/runId/commandId 已保留，可重试上一条消息。", status: "sent", turnId: traceId, traceId },
{ messageId: "msg_writer_canceled_prompt_preserved_agent", role: "agent", text: "", status: "canceled", turnId: traceId, traceId }
]
}
});
const loaded = runtimeStore.queryWorkbenchFacts({ traceId, families: ["messages", "parts", "turns"], limit: 20 });
internal/cloud/workbench-projection-writer.ts
+56 -8
View File
@@ -121,6 +121,16 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
const occurredAt = latestTimestamp(previousTiming?.lastEventAt, sourceOccurredAt) ?? sourceOccurredAt;
const previousTerminal = checkpointIsTerminal(resolvedPreviousCheckpoint);
const suppressedAfterSeal = previousTerminal && !terminal;
if (suppressedAfterSeal) {
emitWorkbenchProjectionOtel("workbench.projection.event.write", traceId, "ok", {
"workbench.projection.phase": "event-suppressed-after-seal",
"workbench.projection.terminal": false,
"workbench.projection.suppressed_after_seal": true,
"workbench.projection.session_id": sessionId,
"workbench.projection.turn_id": turnId
});
return { written: false, suppressedAfterSeal: true, valuesPrinted: false };
}
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);
@@ -162,7 +172,7 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
});
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 eventId = isAssistantMessageTraceEvent(event) ? stableFactId("wte", { traceId, sourceEventId }) : textValue(event.id) || stableFactId("wte", { traceId, sourceEventId });
const previousFinalText = finalResponseTextValue(resolvedPreviousCheckpoint?.finalResponse, resolvedPreviousCheckpoint?.assistantText, resolvedPreviousCheckpoint?.finalText);
const messageFinalText = terminal ? finalResponseTextValue(event.finalResponse, event.assistantText, event.reply, previousFinalText) : null;
const userMessageFact = sessionId && !suppressedAfterSeal && resolvedPreviousCheckpoint?.userMessage ? normalizeMessageFact(resolvedPreviousCheckpoint.userMessage, 0, { traceId, sessionId, turnId, terminal: false, terminalStatus: "sent", finalText: null, timestamp: projectedAt, timing }) : null;
@@ -335,16 +345,15 @@ 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.terminal_trace_backfill_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 terminalTraceBackfill ? { ...result, terminalTraceBackfill } : result;
return result;
} catch (error) {
const safeId = safeTraceId(traceId);
emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "error", {
@@ -505,7 +514,7 @@ function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, own
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 }), {
const messages = normalizeMessages(workbenchProjectionInputMessages({ session, payload, params, traceId: safeId, timestamp, terminal, terminalStatus, finalText, previousCheckpoint }), {
traceId: safeId,
sessionId: resolvedSessionId,
conversationId: resolvedConversationId,
@@ -621,9 +630,9 @@ function workbenchProjectionFactCountAttributes(facts = {}) {
};
}
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;
function workbenchProjectionInputMessages({ session = {}, payload = {}, params = {}, traceId = null, timestamp = null, terminal = false, terminalStatus = "completed", finalText = null, previousCheckpoint = null } = {}) {
if (Array.isArray(session?.messages) && session.messages.length > 0) return preserveTerminalUserMessage(session.messages, { terminal, traceId, previousCheckpoint });
if (Array.isArray(payload?.messages) && payload.messages.length > 0) return preserveTerminalUserMessage(payload.messages, { terminal, traceId, previousCheckpoint });
if (!traceId) return [];
const turnId = textValue(payload?.turnId) || traceId;
const assistantMessageId = eventProjectionAssistantMessageId(traceId, payload);
@@ -667,6 +676,35 @@ function workbenchProjectionUserMessage({ payload = {}, params = {}, traceId = n
};
}
function preserveTerminalUserMessage(messages = [], { terminal = false, traceId = null, previousCheckpoint = null } = {}) {
if (!terminal || !traceId || !previousCheckpoint?.userMessage) return messages;
const previousUserMessage = previousCheckpoint.userMessage;
const previousText = textValue(previousUserMessage.text ?? previousUserMessage.content ?? previousUserMessage.message);
if (!previousText) return messages;
let replaced = false;
const normalizedTraceId = safeTraceId(traceId) ?? traceId;
const preserved = messages.map((message) => {
const role = textValue(message?.role) || "agent";
const messageTraceId = safeTraceId(message?.traceId ?? message?.turnId ?? normalizedTraceId) ?? normalizedTraceId;
if (role !== "user" || messageTraceId !== normalizedTraceId) return message;
replaced = true;
return {
...message,
...previousUserMessage,
messageId: textValue(previousUserMessage.messageId ?? previousUserMessage.id ?? message?.messageId ?? message?.id) || eventProjectionUserMessageId(normalizedTraceId, message),
role: "user",
status: "sent",
text: previousText,
traceId: normalizedTraceId,
turnId: textValue(previousUserMessage.turnId ?? message?.turnId) || normalizedTraceId,
terminal: false,
sealed: false
};
});
if (replaced) return preserved;
return [previousUserMessage, ...messages];
}
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)) {
@@ -971,6 +1009,10 @@ function stableFactId(prefix, value) {
}
function workbenchSourceEventId(event = {}, { traceId, sourceSeq, occurredAt } = {}) {
if (isAssistantMessageTraceEvent(event)) {
const messageId = eventProjectionAssistantMessageId(traceId, event);
return `${safeTraceId(traceId) ?? traceId}:assistant-message:${messageId}`;
}
const explicit = textValue(event.sourceEventId ?? event.id);
if (explicit) return explicit;
const source = textValue(event.source);
@@ -990,6 +1032,12 @@ function workbenchSourceEventId(event = {}, { traceId, sourceSeq, occurredAt } =
});
}
function isAssistantMessageTraceEvent(event = {}) {
const type = textValue(event?.eventType ?? event?.type).toLowerCase();
const label = textValue(event?.label).toLowerCase();
return type === "assistant" || /assistant:message|assistant_message/u.test(label);
}
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])]));