Fix workbench trace projection authority (#1696)
This commit is contained in:
Lyon
@@ -1222,6 +1222,58 @@ test("workbench read model keeps non-terminal durable tool completed events out
|
||||
}
|
||||
});
|
||||
|
||||
test("workbench trace events API blocks historical projectedSeq collisions instead of handing them to renderer", async () => {
|
||||
const traceStore = createCodeAgentTraceStore();
|
||||
const traceId = "trc_workbench_projected_seq_collision";
|
||||
const session = {
|
||||
id: "ses_workbench_projected_seq_collision",
|
||||
projectId: "prj_hwpod_workbench",
|
||||
agentId: "hwlab-code-agent",
|
||||
status: "running",
|
||||
ownerUserId: ACTOR.id,
|
||||
conversationId: "cnv_workbench_projected_seq_collision",
|
||||
threadId: "thread-workbench-projected-seq-collision",
|
||||
lastTraceId: traceId,
|
||||
updatedAt: "2026-06-20T12:20:00.000Z",
|
||||
session: { sessionStatus: "running", lastTraceId: traceId, messages: [{ role: "user", text: "collision", traceId, status: "sent" }] }
|
||||
};
|
||||
const accessController = {
|
||||
store: {
|
||||
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
|
||||
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
|
||||
},
|
||||
async ensureBootstrap() {},
|
||||
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
|
||||
};
|
||||
const runtimeStore = createDurableFactsRuntimeStore({
|
||||
sessions: [{
|
||||
session,
|
||||
events: [
|
||||
{ id: "wte_collision_a", projectedSeq: 1, sourceSeq: 1, sourceEventId: "source-a", type: "backend", status: "running", label: "projected:1:a", createdAt: "2026-06-20T12:19:58.000Z" },
|
||||
{ id: "wte_collision_b", projectedSeq: 1, sourceSeq: 2, sourceEventId: "source-b", type: "assistant", status: "running", label: "projected:1:b", message: "later", createdAt: "2026-06-20T12:19:59.000Z" }
|
||||
],
|
||||
status: "running",
|
||||
projectionStatus: "projecting",
|
||||
lastProjectedSeq: 1
|
||||
}]
|
||||
});
|
||||
const server = createCloudApiServer({ accessController, traceStore, runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() });
|
||||
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
|
||||
try {
|
||||
const { port } = server.address();
|
||||
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`);
|
||||
assert.equal(trace.status, 200);
|
||||
assert.equal(trace.body.status, "blocked");
|
||||
assert.equal(trace.body.projectionStatus, "blocked");
|
||||
assert.equal(trace.body.projectionHealth, "degraded");
|
||||
assert.equal(trace.body.blocker.code, "projected_seq_collision");
|
||||
assert.deepEqual(trace.body.events, []);
|
||||
} finally {
|
||||
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
|
||||
}
|
||||
});
|
||||
|
||||
test("workbench realtime stream accepts session authority without project or workspace", async () => {
|
||||
const traceStore = createCodeAgentTraceStore();
|
||||
const results = createCodeAgentChatResultStore();
|
||||
|
||||
@@ -933,22 +933,23 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT
|
||||
const pageResult = await readModel.queryFacts({ traceId, families: WORKBENCH_TRACE_EVENT_PAGE_FAMILIES, afterProjectedSeq: pageOptions.afterProjectedSeq, limit: pageOptions.limit + 1 });
|
||||
if (pageResult.error) return sendJson(response, 503, workbenchProjectionStoreError(pageResult.error));
|
||||
const page = traceEventPageFromFacts(pageResult.facts.traceEvents, pageOptions, { total: projection.lastProjectedSeq, traceStatus: session.status });
|
||||
const responseProjection = page.blocker ? blockedTraceEventProjection(projection, page.blocker) : projection;
|
||||
sendJson(response, 200, {
|
||||
ok: true,
|
||||
status: "succeeded",
|
||||
status: page.blocker ? "blocked" : "succeeded",
|
||||
contractVersion: "workbench-trace-events-v1",
|
||||
traceId,
|
||||
sessionId: factSessionId(session),
|
||||
threadId: safeOpaqueId(session?.threadId) ?? (textValue(session?.threadId) || null),
|
||||
...page,
|
||||
projection,
|
||||
projectionStatus: projection.projectionStatus,
|
||||
projectionHealth: projection.projectionHealth,
|
||||
lastProjectedSeq: projection.lastProjectedSeq,
|
||||
sourceRunId: projection.sourceRunId,
|
||||
sourceCommandId: projection.sourceCommandId,
|
||||
staleMs: projection.staleMs,
|
||||
blocker: projection.blocker,
|
||||
projection: responseProjection,
|
||||
projectionStatus: responseProjection.projectionStatus,
|
||||
projectionHealth: responseProjection.projectionHealth,
|
||||
lastProjectedSeq: responseProjection.lastProjectedSeq,
|
||||
sourceRunId: responseProjection.sourceRunId,
|
||||
sourceCommandId: responseProjection.sourceCommandId,
|
||||
staleMs: responseProjection.staleMs,
|
||||
blocker: responseProjection.blocker,
|
||||
valuesRedacted: true,
|
||||
secretMaterialStored: false
|
||||
});
|
||||
@@ -956,6 +957,8 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT
|
||||
|
||||
function traceEventPageFromFacts(sourceEvents, options, metadata = {}) {
|
||||
const rows = factArray(sourceEvents).filter((event) => factProjectedSeq(event)).sort(compareFactTraceEventsAsc);
|
||||
const collision = projectedSeqCollision(rows);
|
||||
if (collision) return blockedTraceEventPage(options, metadata, collision);
|
||||
const pageRows = rows.slice(0, options.limit);
|
||||
const events = pageRows.map(factTraceEventDto).filter(Boolean);
|
||||
const toProjectedSeq = events.length ? events.at(-1).projectedSeq : null;
|
||||
@@ -980,6 +983,62 @@ function traceEventPageFromFacts(sourceEvents, options, metadata = {}) {
|
||||
};
|
||||
}
|
||||
|
||||
function projectedSeqCollision(rows = []) {
|
||||
const seen = new Map();
|
||||
for (const row of rows) {
|
||||
const projectedSeq = factProjectedSeq(row);
|
||||
if (!projectedSeq) continue;
|
||||
const id = textValue(row?.id ?? row?.sourceEventId) || null;
|
||||
const existing = seen.get(projectedSeq);
|
||||
if (existing && existing.id !== id) {
|
||||
return {
|
||||
code: "projected_seq_collision",
|
||||
category: "workbench-projection",
|
||||
layer: "workbench-trace-events",
|
||||
message: "Trace projection contains duplicate projectedSeq values; canonical event order is blocked until projection is rebuilt.",
|
||||
projectedSeq,
|
||||
eventIds: [existing.id, id].filter(Boolean),
|
||||
retryable: false,
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
if (!existing) seen.set(projectedSeq, { id });
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function blockedTraceEventPage(options, metadata = {}, blocker) {
|
||||
const total = Number.isFinite(Number(metadata.total)) ? Math.trunc(Number(metadata.total)) : null;
|
||||
return {
|
||||
events: [],
|
||||
eventCount: total ?? options.afterProjectedSeq,
|
||||
range: {
|
||||
afterProjectedSeq: options.afterProjectedSeq,
|
||||
fromProjectedSeq: null,
|
||||
toProjectedSeq: null,
|
||||
limit: options.limit,
|
||||
returned: 0,
|
||||
total
|
||||
},
|
||||
hasMore: false,
|
||||
nextProjectedSeq: options.afterProjectedSeq,
|
||||
nextCursor: null,
|
||||
traceStatus: metadata.traceStatus ?? "unknown",
|
||||
updatedAt: null,
|
||||
blocker
|
||||
};
|
||||
}
|
||||
|
||||
function blockedTraceEventProjection(projection = {}, blocker) {
|
||||
return {
|
||||
...projection,
|
||||
projectionStatus: "blocked",
|
||||
projectionHealth: "degraded",
|
||||
blocker,
|
||||
valuesRedacted: true
|
||||
};
|
||||
}
|
||||
|
||||
async function visibleSessionById(store, sessionId, actor) {
|
||||
const safeId = safeSessionId(sessionId);
|
||||
if (!safeId) return null;
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "bun:test";
|
||||
|
||||
import { createCloudRuntimeStore } from "../db/runtime-store.ts";
|
||||
import { writeWorkbenchProjectionEvent, writeWorkbenchProjectionSession } from "./workbench-projection-writer.ts";
|
||||
|
||||
test("workbench projection writer commits terminal owner evidence as sealed durable facts", async () => {
|
||||
@@ -77,10 +78,15 @@ test("workbench projection writer commits terminal owner evidence as sealed dura
|
||||
|
||||
test("workbench projection event commit writes trace event and checkpoint facts", async () => {
|
||||
const factWrites = [];
|
||||
const baseStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:00:00.000Z" });
|
||||
const runtimeStore = {
|
||||
allocateWorkbenchProjectedSeq(params, requestMeta) {
|
||||
return baseStore.allocateWorkbenchProjectedSeq(params, requestMeta);
|
||||
},
|
||||
async writeWorkbenchFacts(params, requestMeta) {
|
||||
const result = baseStore.writeWorkbenchFacts(params, requestMeta);
|
||||
factWrites.push({ params, requestMeta });
|
||||
return { written: true, facts: params.facts };
|
||||
return result;
|
||||
}
|
||||
};
|
||||
|
||||
@@ -117,9 +123,45 @@ test("workbench projection event commit writes trace event and checkpoint facts"
|
||||
});
|
||||
|
||||
assert.equal(factWrites.length, 2);
|
||||
assert.equal(factWrites[0].params.facts.traceEvents[0].projectedSeq, 7);
|
||||
assert.equal(factWrites[0].params.facts.traceEvents[0].projectedSeq, 1);
|
||||
assert.equal(factWrites[0].params.facts.checkpoints[0].projectionStatus, "projecting");
|
||||
assert.equal(factWrites[1].params.facts.sessions[0].status, "completed");
|
||||
assert.equal(factWrites[1].params.facts.traceEvents[0].sealed, true);
|
||||
assert.equal(factWrites[1].params.facts.traceEvents[0].projectedSeq, 2);
|
||||
assert.equal(factWrites[1].params.facts.checkpoints[0].projectionStatus, "caught_up");
|
||||
});
|
||||
|
||||
test("workbench projection event writer allocates projectedSeq idempotently by source event identity", async () => {
|
||||
const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T12:00:00.000Z" });
|
||||
|
||||
const repeatedSourceEvent = {
|
||||
traceId: "trc_writer_idempotent",
|
||||
sessionId: "ses_writer_idempotent",
|
||||
seq: 1,
|
||||
source: "agentrun",
|
||||
sourceSeq: 70,
|
||||
type: "backend",
|
||||
status: "running",
|
||||
label: "agentrun:backend:running",
|
||||
runId: "run_writer_idempotent",
|
||||
commandId: "cmd_writer_idempotent",
|
||||
createdAt: "2026-06-20T12:00:01.000Z"
|
||||
};
|
||||
await writeWorkbenchProjectionEvent({ runtimeStore, event: repeatedSourceEvent });
|
||||
await writeWorkbenchProjectionEvent({
|
||||
runtimeStore,
|
||||
event: {
|
||||
...repeatedSourceEvent,
|
||||
seq: 1,
|
||||
sourceSeq: 71,
|
||||
label: "agentrun:backend:runner-ready",
|
||||
createdAt: "2026-06-20T12:00:02.000Z"
|
||||
}
|
||||
});
|
||||
await writeWorkbenchProjectionEvent({ runtimeStore, event: repeatedSourceEvent });
|
||||
|
||||
const loaded = runtimeStore.queryWorkbenchFacts({ traceId: "trc_writer_idempotent", families: ["traceEvents", "checkpoints"], limit: 10 });
|
||||
assert.deepEqual(loaded.facts.traceEvents.map((event) => event.projectedSeq), [1, 2]);
|
||||
assert.deepEqual(loaded.facts.traceEvents.map((event) => event.sourceSeq), [70, 71]);
|
||||
assert.equal(loaded.facts.checkpoints[0].projectedSeq, 2);
|
||||
});
|
||||
|
||||
@@ -54,17 +54,36 @@ export async function writeWorkbenchProjectionSession({ accessController, runtim
|
||||
|
||||
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 sourceSeq = nonNegativeInteger(event.sourceSeq ?? event.seq);
|
||||
const projectedSeq = nonNegativeInteger(event.seq ?? event.projectedSeq ?? sourceSeq);
|
||||
const occurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt);
|
||||
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 sourceEventId = textValue(event.sourceEventId ?? event.id) || `${traceId}:${sourceSeq || projectedSeq}`;
|
||||
const eventId = textValue(event.id) || stableFactId("wte", { traceId, sourceEventId, sourceSeq, projectedSeq, label: event.label, occurredAt });
|
||||
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: timestampValue(event.updatedAt ?? occurredAt)
|
||||
}, {
|
||||
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 facts = {
|
||||
sessions: sessionId ? [{
|
||||
sessionId,
|
||||
@@ -279,6 +298,26 @@ 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])]));
|
||||
|
||||
@@ -806,6 +806,41 @@ test("configured postgres runtime persists and queries Workbench durable facts",
|
||||
assert.ok(queryClient.calls.some((call) => call.sql.startsWith("SELECT session_json FROM workbench_sessions WHERE session_id = $1")));
|
||||
});
|
||||
|
||||
test("configured postgres runtime allocates Workbench projectedSeq idempotently by source event", async () => {
|
||||
const queryClient = createFakePostgresClient({ migrationReady: true });
|
||||
const store = createConfiguredCloudRuntimeStore({
|
||||
env: {
|
||||
HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres",
|
||||
HWLAB_CLOUD_RUNTIME_DURABLE: "true"
|
||||
},
|
||||
dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab",
|
||||
queryClient,
|
||||
now: () => "2026-06-20T12:30:00.000Z"
|
||||
});
|
||||
const traceId = "trc_pg_projected_seq_idempotent";
|
||||
|
||||
const first = await store.allocateWorkbenchProjectedSeq({ traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-a" });
|
||||
await store.writeWorkbenchFacts({ facts: {
|
||||
traceEvents: [{ id: "wte_pg_source_a", traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-a", projectedSeq: first.projectedSeq, eventType: "backend", occurredAt: "2026-06-20T12:30:01.000Z" }],
|
||||
checkpoints: [{ traceId, sessionId: "ses_pg_projected_seq_idempotent", projectedSeq: first.projectedSeq, sourceSeq: 1, sourceEventId: "source-a", projectionStatus: "projecting", projectionHealth: "healthy", updatedAt: "2026-06-20T12:30:01.000Z" }]
|
||||
} });
|
||||
const second = await store.allocateWorkbenchProjectedSeq({ traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-b" });
|
||||
await store.writeWorkbenchFacts({ facts: {
|
||||
traceEvents: [{ id: "wte_pg_source_b", traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-b", projectedSeq: second.projectedSeq, eventType: "backend", occurredAt: "2026-06-20T12:30:02.000Z" }],
|
||||
checkpoints: [{ traceId, sessionId: "ses_pg_projected_seq_idempotent", projectedSeq: second.projectedSeq, sourceSeq: 1, sourceEventId: "source-b", projectionStatus: "projecting", projectionHealth: "healthy", updatedAt: "2026-06-20T12:30:02.000Z" }]
|
||||
} });
|
||||
const repeated = await store.allocateWorkbenchProjectedSeq({ traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-a" });
|
||||
await store.writeWorkbenchFacts({ facts: { checkpoints: [{ traceId, sessionId: "ses_pg_projected_seq_idempotent", projectedSeq: repeated.projectedSeq, sourceSeq: 1, sourceEventId: "source-a", projectionStatus: "projecting", projectionHealth: "healthy", updatedAt: "2026-06-20T12:30:03.000Z" }] } });
|
||||
|
||||
const loaded = await store.queryWorkbenchFacts({ traceId, families: ["traceEvents", "checkpoints"], limit: 10 });
|
||||
assert.equal(first.projectedSeq, 1);
|
||||
assert.equal(second.projectedSeq, 2);
|
||||
assert.equal(repeated.projectedSeq, 1);
|
||||
assert.deepEqual(loaded.facts.traceEvents.map((event) => event.projectedSeq), [1, 2]);
|
||||
assert.equal(loaded.facts.checkpoints[0].projectedSeq, 2);
|
||||
assert.ok(queryClient.calls.some((call) => call.sql.startsWith("SELECT pg_advisory_xact_lock")));
|
||||
});
|
||||
|
||||
test("configured postgres runtime reuses proven readiness for Workbench durable reads", async () => {
|
||||
const queryClient = createFakePostgresClient({ migrationReady: true });
|
||||
const store = createConfiguredCloudRuntimeStore({
|
||||
@@ -1000,6 +1035,9 @@ function createFakePostgresClient({
|
||||
calls,
|
||||
async query(sql, params = []) {
|
||||
calls.push({ sql, params });
|
||||
if (["BEGIN", "COMMIT", "ROLLBACK"].includes(sql)) {
|
||||
return { rows: [] };
|
||||
}
|
||||
if (sql.includes("information_schema.columns")) {
|
||||
return { rows: schemaRows() };
|
||||
}
|
||||
@@ -1030,6 +1068,23 @@ function createFakePostgresClient({
|
||||
if (sql.startsWith("CREATE INDEX IF NOT EXISTS idx_workbench_")) {
|
||||
return { rows: [] };
|
||||
}
|
||||
if (sql.startsWith("SELECT pg_advisory_xact_lock")) {
|
||||
return { rows: [{ pg_advisory_xact_lock: null }] };
|
||||
}
|
||||
if (sql.startsWith("SELECT projected_seq FROM workbench_trace_events WHERE trace_id = $1 AND source_event_id = $2")) {
|
||||
const rows = [...state.workbench_trace_events.values()]
|
||||
.filter((record) => record.trace_id === params[0] && record.source_event_id === params[1])
|
||||
.sort((left, right) => Number(left.projected_seq) - Number(right.projected_seq));
|
||||
return { rows: rows.slice(0, 1).map((record) => ({ projected_seq: record.projected_seq })) };
|
||||
}
|
||||
if (sql.startsWith("SELECT GREATEST(COALESCE((SELECT MAX(projected_seq) FROM workbench_trace_events")) {
|
||||
const traceId = params[0];
|
||||
const eventMax = [...state.workbench_trace_events.values()]
|
||||
.filter((record) => record.trace_id === traceId)
|
||||
.reduce((max, record) => Math.max(max, Number(record.projected_seq ?? 0)), 0);
|
||||
const checkpointMax = Number(state.workbench_projection_checkpoints.get(traceId)?.projected_seq ?? 0);
|
||||
return { rows: [{ max_projected_seq: Math.max(eventMax, checkpointMax) }] };
|
||||
}
|
||||
if (sql.startsWith("SELECT gateway_session_json FROM gateway_sessions")) {
|
||||
return jsonSelect(state.gateway_sessions, params[0], "gateway_session_json");
|
||||
}
|
||||
@@ -1206,10 +1261,19 @@ function createFakePostgresClient({
|
||||
return { rows: [] };
|
||||
}
|
||||
if (sql.startsWith("INSERT INTO workbench_projection_checkpoints")) {
|
||||
const previous = state.workbench_projection_checkpoints.get(params[0]) ?? null;
|
||||
const projectedSeq = sql.includes("RETURNING projected_seq")
|
||||
? Math.max(Number(previous?.projected_seq ?? 0) + (previous ? 1 : 0), Number(params[5] ?? 0))
|
||||
: previous && Number(params[5] ?? 0) < Number(previous.projected_seq ?? 0)
|
||||
? Number(previous.projected_seq)
|
||||
: Number(params[5] ?? 0);
|
||||
if (previous && !sql.includes("RETURNING projected_seq") && Number(params[5] ?? 0) < Number(previous.projected_seq ?? 0)) {
|
||||
return { rows: [] };
|
||||
}
|
||||
state.workbench_projection_checkpoints.set(params[0], {
|
||||
trace_id: params[0], session_id: params[1], turn_id: params[2], run_id: params[3], command_id: params[4], projected_seq: params[5], source_seq: params[6], source_event_id: params[7], projection_status: params[8], projection_health: params[9], terminal: params[10], sealed: params[11], diagnostic_json: params[12], checkpoint_json: params[13], created_at: params[14], updated_at: params[15]
|
||||
trace_id: params[0], session_id: params[1], turn_id: params[2], run_id: params[3], command_id: params[4], projected_seq: projectedSeq, source_seq: params[6], source_event_id: params[7], projection_status: params[8], projection_health: params[9], terminal: params[10], sealed: params[11], diagnostic_json: params[12], checkpoint_json: params[13], created_at: params[14], updated_at: params[15]
|
||||
});
|
||||
return { rows: [] };
|
||||
return { rows: sql.includes("RETURNING projected_seq") ? [{ projected_seq: projectedSeq }] : [] };
|
||||
}
|
||||
throw new Error(`unexpected sql: ${sql}`);
|
||||
}
|
||||
|
||||
@@ -574,6 +574,33 @@ export class CloudRuntimeStore {
|
||||
};
|
||||
}
|
||||
|
||||
allocateWorkbenchProjectedSeq(params = {}, requestMeta = {}) {
|
||||
const allocation = normalizeWorkbenchProjectionAllocation(params, requestMeta, this.now());
|
||||
const existing = [...this.workbenchTraceEvents.values()].find((record) =>
|
||||
record.traceId === allocation.traceId && record.sourceEventId === allocation.sourceEventId
|
||||
);
|
||||
if (existing) {
|
||||
return {
|
||||
...allocation,
|
||||
projectedSeq: nonNegativeInteger(existing.projectedSeq),
|
||||
reused: true,
|
||||
persistence: this.summary(),
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
const eventMax = [...this.workbenchTraceEvents.values()]
|
||||
.filter((record) => record.traceId === allocation.traceId)
|
||||
.reduce((max, record) => Math.max(max, nonNegativeInteger(record.projectedSeq)), 0);
|
||||
const checkpointMax = nonNegativeInteger(this.workbenchProjectionCheckpoints.get(allocation.traceId)?.projectedSeq);
|
||||
return {
|
||||
...allocation,
|
||||
projectedSeq: Math.max(eventMax, checkpointMax) + 1,
|
||||
reused: false,
|
||||
persistence: this.summary(),
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
|
||||
queryWorkbenchProjectionStates(params = {}) {
|
||||
const dueAt = textOr(params.dueAt, "");
|
||||
const statuses = Array.isArray(params.projectionStatuses) ? new Set(params.projectionStatuses.map((item) => textOr(item, "")).filter(Boolean)) : null;
|
||||
@@ -596,7 +623,12 @@ export class CloudRuntimeStore {
|
||||
for (const fact of facts.parts) this.workbenchParts.set(fact.partId, fact);
|
||||
for (const fact of facts.turns) this.workbenchTurns.set(fact.turnId, fact);
|
||||
for (const fact of facts.traceEvents) this.workbenchTraceEvents.set(fact.id, fact);
|
||||
for (const fact of facts.checkpoints) this.workbenchProjectionCheckpoints.set(fact.traceId, fact);
|
||||
for (const fact of facts.checkpoints) {
|
||||
const previous = this.workbenchProjectionCheckpoints.get(fact.traceId) ?? null;
|
||||
if (!previous || nonNegativeInteger(fact.projectedSeq) >= nonNegativeInteger(previous.projectedSeq)) {
|
||||
this.workbenchProjectionCheckpoints.set(fact.traceId, fact);
|
||||
}
|
||||
}
|
||||
return {
|
||||
written: true,
|
||||
facts,
|
||||
@@ -1198,6 +1230,40 @@ export class PostgresCloudRuntimeStore {
|
||||
};
|
||||
}
|
||||
|
||||
async allocateWorkbenchProjectedSeq(params = {}, requestMeta = {}) {
|
||||
const readiness = this.summary();
|
||||
if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
|
||||
await this.assertReadyForWrites();
|
||||
}
|
||||
const allocation = normalizeWorkbenchProjectionAllocation(params, requestMeta, this.now());
|
||||
const result = await this.withDurableTransaction("workbench.projected-seq.allocate", async (client) => {
|
||||
await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-projection:${allocation.traceId}`]);
|
||||
const existing = await client.query(
|
||||
"SELECT projected_seq FROM workbench_trace_events WHERE trace_id = $1 AND source_event_id = $2 ORDER BY projected_seq ASC LIMIT 1",
|
||||
[allocation.traceId, allocation.sourceEventId]
|
||||
);
|
||||
const existingSeq = nonNegativeInteger(existing.rows?.[0]?.projected_seq);
|
||||
if (existingSeq > 0) return { projectedSeq: existingSeq, reused: true };
|
||||
const maxResult = await client.query(
|
||||
"SELECT GREATEST(COALESCE((SELECT MAX(projected_seq) FROM workbench_trace_events WHERE trace_id = $1), 0), COALESCE((SELECT projected_seq FROM workbench_projection_checkpoints WHERE trace_id = $1), 0))::int AS max_projected_seq",
|
||||
[allocation.traceId]
|
||||
);
|
||||
const proposedSeq = nonNegativeInteger(maxResult.rows?.[0]?.max_projected_seq) + 1;
|
||||
const reserve = await client.query(
|
||||
"INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET projected_seq = GREATEST(workbench_projection_checkpoints.projected_seq + 1, EXCLUDED.projected_seq), updated_at = EXCLUDED.updated_at RETURNING projected_seq",
|
||||
[allocation.traceId, allocation.sessionId, allocation.turnId, allocation.runId, allocation.commandId, proposedSeq, allocation.sourceSeq, allocation.sourceEventId, "projecting", "healthy", false, false, stableJson({ allocator: "workbench-projected-seq", valuesRedacted: true }), stableJson({ ...allocation, projectedSeq: proposedSeq, projectionStatus: "projecting", projectionHealth: "healthy", valuesPrinted: false }), allocation.createdAt, allocation.updatedAt]
|
||||
);
|
||||
return { projectedSeq: nonNegativeInteger(reserve.rows?.[0]?.projected_seq) || proposedSeq, reused: false };
|
||||
});
|
||||
return {
|
||||
...allocation,
|
||||
projectedSeq: result.projectedSeq,
|
||||
reused: result.reused,
|
||||
persistence: this.summary(),
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
|
||||
async writeWorkbenchFacts(params = {}, requestMeta = {}) {
|
||||
const readiness = this.summary();
|
||||
if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
|
||||
@@ -1566,7 +1632,7 @@ export class PostgresCloudRuntimeStore {
|
||||
|
||||
async persistWorkbenchProjectionCheckpoint(record) {
|
||||
await this.query(
|
||||
"INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, run_id = EXCLUDED.run_id, command_id = EXCLUDED.command_id, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, projection_status = EXCLUDED.projection_status, projection_health = EXCLUDED.projection_health, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, diagnostic_json = EXCLUDED.diagnostic_json, checkpoint_json = EXCLUDED.checkpoint_json, updated_at = EXCLUDED.updated_at",
|
||||
"INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET session_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.session_id ELSE workbench_projection_checkpoints.session_id END, turn_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.turn_id ELSE workbench_projection_checkpoints.turn_id END, run_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.run_id ELSE workbench_projection_checkpoints.run_id END, command_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.command_id ELSE workbench_projection_checkpoints.command_id END, projected_seq = GREATEST(workbench_projection_checkpoints.projected_seq, EXCLUDED.projected_seq), source_seq = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.source_seq ELSE workbench_projection_checkpoints.source_seq END, source_event_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.source_event_id ELSE workbench_projection_checkpoints.source_event_id END, projection_status = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.projection_status ELSE workbench_projection_checkpoints.projection_status END, projection_health = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.projection_health ELSE workbench_projection_checkpoints.projection_health END, terminal = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.terminal ELSE workbench_projection_checkpoints.terminal END, sealed = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.sealed ELSE workbench_projection_checkpoints.sealed END, diagnostic_json = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.diagnostic_json ELSE workbench_projection_checkpoints.diagnostic_json END, checkpoint_json = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.checkpoint_json ELSE workbench_projection_checkpoints.checkpoint_json END, updated_at = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.updated_at ELSE workbench_projection_checkpoints.updated_at END",
|
||||
[record.traceId, record.sessionId, record.turnId, record.runId, record.commandId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.projectionStatus, record.projectionHealth, record.terminal, record.sealed, stableJson(record.diagnostic), stableJson(record), record.createdAt, record.updatedAt]
|
||||
);
|
||||
}
|
||||
@@ -1623,6 +1689,33 @@ export class PostgresCloudRuntimeStore {
|
||||
return client.query(sql, params);
|
||||
}
|
||||
|
||||
async withDurableTransaction(label, fn) {
|
||||
const client = await this.getQueryClient();
|
||||
if (typeof client.connect === "function") {
|
||||
const tx = await client.connect();
|
||||
try {
|
||||
await tx.query("BEGIN");
|
||||
const result = await fn(tx);
|
||||
await tx.query("COMMIT");
|
||||
return result;
|
||||
} catch (error) {
|
||||
try { await tx.query("ROLLBACK"); } catch {}
|
||||
throw error;
|
||||
} finally {
|
||||
tx.release?.();
|
||||
}
|
||||
}
|
||||
await client.query("BEGIN");
|
||||
try {
|
||||
const result = await fn(client);
|
||||
await client.query("COMMIT");
|
||||
return result;
|
||||
} catch (error) {
|
||||
try { await client.query("ROLLBACK"); } catch {}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async getQueryClient() {
|
||||
if (this.queryClient) {
|
||||
return this.queryClient;
|
||||
@@ -1839,6 +1932,33 @@ function normalizeWorkbenchProjectionState(value, requestMeta = {}, now) {
|
||||
});
|
||||
}
|
||||
|
||||
function normalizeWorkbenchProjectionAllocation(value, requestMeta = {}, now) {
|
||||
const input = normalizeJsonObject(value);
|
||||
const traceId = textOr(input.traceId ?? requestMeta.traceId, "");
|
||||
const sourceEventId = textOr(input.sourceEventId ?? input.eventId, "");
|
||||
if (!traceId || !sourceEventId) {
|
||||
throw new HwlabProtocolError("workbench projectedSeq allocation requires traceId and sourceEventId", {
|
||||
code: ERROR_CODES.invalidParams,
|
||||
data: { required: ["traceId", "sourceEventId"] }
|
||||
});
|
||||
}
|
||||
const timestamp = timestampOr(input.updatedAt ?? input.occurredAt ?? input.createdAt, now);
|
||||
return pruneUndefined({
|
||||
...input,
|
||||
traceId,
|
||||
sessionId: textOr(input.sessionId ?? requestMeta.sessionId, "") || null,
|
||||
turnId: textOr(input.turnId ?? requestMeta.turnId ?? traceId, "") || null,
|
||||
runId: textOr(input.runId ?? input.sourceRunId, "") || null,
|
||||
commandId: textOr(input.commandId ?? input.sourceCommandId, "") || null,
|
||||
sourceSeq: nonNegativeInteger(input.sourceSeq ?? input.lastSourceSeq ?? input.lastAgentRunSeq),
|
||||
sourceEventId,
|
||||
eventType: textOr(input.eventType ?? input.type, "event"),
|
||||
createdAt: timestampOr(input.createdAt, timestamp),
|
||||
updatedAt: timestamp,
|
||||
valuesPrinted: false
|
||||
});
|
||||
}
|
||||
|
||||
function normalizeWorkbenchFacts(value, requestMeta = {}, now) {
|
||||
const input = normalizeJsonObject(value);
|
||||
return {
|
||||
|
||||
@@ -16,9 +16,6 @@ export function traceDisplayRows(trace: Record<string, unknown> = {}, events: Tr
|
||||
const rows: TraceEventRow[] = [];
|
||||
const renderedSourceEvents = new Set<string>();
|
||||
const renderedToolIdentities = new Set<string>();
|
||||
const assistantRows: AssistantRowState[] = [];
|
||||
let lastAssistantRowIndex = -1;
|
||||
let completionEvent: TraceEvent | null = null;
|
||||
for (let index = 0; index < events.length; index += 1) {
|
||||
const event = events[index];
|
||||
const sourceEventKey = traceSourceEventKey(event);
|
||||
@@ -44,30 +41,20 @@ export function traceDisplayRows(trace: Record<string, unknown> = {}, events: Tr
|
||||
continue;
|
||||
}
|
||||
if (isTerminalAssistantTraceEvent(event)) {
|
||||
const assistantRowIndex = upsertAssistantMessageRow(rows, assistantRows, effectiveTrace, event, { terminal: true });
|
||||
if (assistantRowIndex >= 0) lastAssistantRowIndex = assistantRowIndex;
|
||||
rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: true }));
|
||||
continue;
|
||||
}
|
||||
if (isAssistantTraceEvent(event)) {
|
||||
const assistantRowIndex = upsertAssistantMessageRow(rows, assistantRows, effectiveTrace, event, { terminal: false });
|
||||
if (assistantRowIndex >= 0) lastAssistantRowIndex = assistantRowIndex;
|
||||
rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: false }));
|
||||
continue;
|
||||
}
|
||||
if (isCompletionTraceEvent(event)) {
|
||||
completionEvent = event;
|
||||
const finalResponseText = traceFinalResponseText(effectiveTrace);
|
||||
rows.push(finalResponseText ? traceFinalResponseRow(event, finalResponseText) : traceCompletionSummaryRow(effectiveTrace, event));
|
||||
continue;
|
||||
}
|
||||
rows.push(traceDisplayRow(effectiveTrace, event));
|
||||
}
|
||||
if (completionEvent) {
|
||||
const finalResponseText = traceFinalResponseText(effectiveTrace);
|
||||
if (finalResponseText) {
|
||||
lastAssistantRowIndex = upsertAuthoritativeFinalResponseRow(rows, assistantRows, completionEvent, finalResponseText);
|
||||
}
|
||||
const lastAssistantRow = lastAssistantRowIndex >= 0 ? rows[lastAssistantRowIndex] : undefined;
|
||||
if (lastAssistantRow) rows[lastAssistantRowIndex] = markAssistantRowTerminal(effectiveTrace, lastAssistantRow, completionEvent);
|
||||
else rows.push(traceCompletionSummaryRow(effectiveTrace, completionEvent));
|
||||
}
|
||||
const progressRow = traceBackendProgressSummaryRow(effectiveTrace, rows, events);
|
||||
if (progressRow) rows.push(progressRow);
|
||||
if (rows.length > 0) return rows;
|
||||
@@ -113,18 +100,6 @@ function stripTraceToolOutputFromSummary(value: string): string {
|
||||
return value.replace(/\s+(stdout:|stderr:|exitCode=).*$/isu, "").trim();
|
||||
}
|
||||
|
||||
interface AssistantRowState {
|
||||
rowIndex: number;
|
||||
comparableText: string;
|
||||
sourceEvent: TraceEvent;
|
||||
derivedSnapshotSuffix?: boolean;
|
||||
}
|
||||
|
||||
interface AssistantSnapshotDecision {
|
||||
action: "replace" | "append-suffix" | "keep-existing";
|
||||
suffixText?: string;
|
||||
}
|
||||
|
||||
function traceToolCallRow(trace: Record<string, unknown>, event: TraceEvent): TraceEventRow {
|
||||
const command = cleanShellCommand(event.command);
|
||||
const status = traceStatusToken(event);
|
||||
@@ -211,146 +186,23 @@ function traceAssistantMessageRow(trace: Record<string, unknown>, event: TraceEv
|
||||
};
|
||||
}
|
||||
|
||||
function upsertAssistantMessageRow(rows: TraceEventRow[], assistantRows: AssistantRowState[], trace: Record<string, unknown>, event: TraceEvent, options: { terminal: boolean }): number {
|
||||
const row = traceAssistantMessageRow(trace, event, options);
|
||||
const comparableText = comparableAssistantText(row.body);
|
||||
if (!comparableText) {
|
||||
rows.push(row);
|
||||
assistantRows.push({ rowIndex: rows.length - 1, comparableText, sourceEvent: event });
|
||||
return rows.length - 1;
|
||||
}
|
||||
|
||||
for (let index = assistantRows.length - 1; index >= 0; index -= 1) {
|
||||
const state = assistantRows[index];
|
||||
if (!state) continue;
|
||||
const existingRow = rows[state.rowIndex];
|
||||
if (!existingRow) continue;
|
||||
if (state.comparableText === comparableText) {
|
||||
rows[state.rowIndex] = mergeAssistantRows(existingRow, row);
|
||||
state.sourceEvent = event;
|
||||
state.derivedSnapshotSuffix = false;
|
||||
return state.rowIndex;
|
||||
}
|
||||
if (comparableText.includes(state.comparableText)) {
|
||||
const previousIndex = comparableText.indexOf(state.comparableText);
|
||||
if (state.derivedSnapshotSuffix === true && previousIndex === 0) {
|
||||
rows[state.rowIndex] = mergeAssistantRows(row, existingRow);
|
||||
state.comparableText = comparableText;
|
||||
state.sourceEvent = event;
|
||||
state.derivedSnapshotSuffix = false;
|
||||
return state.rowIndex;
|
||||
}
|
||||
const snapshotDecision = assistantSnapshotDecision(comparableText, state.comparableText, event, state.sourceEvent);
|
||||
if (snapshotDecision.action === "append-suffix" && snapshotDecision.suffixText) {
|
||||
const suffixRow = { ...row, body: snapshotDecision.suffixText };
|
||||
rows.push(suffixRow);
|
||||
assistantRows.push({ rowIndex: rows.length - 1, comparableText: snapshotDecision.suffixText, sourceEvent: event, derivedSnapshotSuffix: true });
|
||||
return rows.length - 1;
|
||||
}
|
||||
if (snapshotDecision.action === "keep-existing") return state.rowIndex;
|
||||
rows[state.rowIndex] = mergeAssistantRows(row, existingRow);
|
||||
state.comparableText = comparableText;
|
||||
state.sourceEvent = event;
|
||||
state.derivedSnapshotSuffix = false;
|
||||
return state.rowIndex;
|
||||
}
|
||||
if (state.comparableText.includes(comparableText)) {
|
||||
rows[state.rowIndex] = mergeAssistantRows(existingRow, row);
|
||||
return state.rowIndex;
|
||||
}
|
||||
}
|
||||
|
||||
rows.push(row);
|
||||
assistantRows.push({ rowIndex: rows.length - 1, comparableText, sourceEvent: event });
|
||||
return rows.length - 1;
|
||||
}
|
||||
|
||||
function upsertAuthoritativeFinalResponseRow(rows: TraceEventRow[], assistantRows: AssistantRowState[], completionEvent: TraceEvent, finalText: string): number {
|
||||
const comparableText = comparableAssistantText(finalText);
|
||||
let matchedState: AssistantRowState | null = null;
|
||||
for (const state of assistantRows) {
|
||||
const row = rows[state.rowIndex];
|
||||
if (!row || row.bodyFormat !== "markdown") continue;
|
||||
row.terminal = undefined;
|
||||
if (!matchedState && state.comparableText === comparableText) matchedState = state;
|
||||
}
|
||||
if (matchedState) {
|
||||
const row = rows[matchedState.rowIndex];
|
||||
if (row) {
|
||||
rows[matchedState.rowIndex] = {
|
||||
...row,
|
||||
body: finalText,
|
||||
terminal: true,
|
||||
tone: "ok",
|
||||
header: `${traceEventClock(completionEvent)} 助手最终消息`
|
||||
};
|
||||
matchedState.comparableText = comparableText;
|
||||
matchedState.derivedSnapshotSuffix = false;
|
||||
return matchedState.rowIndex;
|
||||
}
|
||||
}
|
||||
rows.push({
|
||||
rowId: `trace-final-response:${completionEvent.seq ?? "completed"}`,
|
||||
seq: numberOrNull(completionEvent.seq),
|
||||
function traceFinalResponseRow(event: TraceEvent, finalText: string): TraceEventRow {
|
||||
return {
|
||||
rowId: `trace-final-response:${event.seq ?? "completed"}`,
|
||||
seq: numberOrNull(event.seq),
|
||||
tone: "ok",
|
||||
header: `${traceEventClock(completionEvent)} 助手最终消息`,
|
||||
header: `${traceEventClock(event)} 助手最终消息`,
|
||||
terminal: true,
|
||||
body: finalText,
|
||||
bodyFormat: "markdown"
|
||||
});
|
||||
assistantRows.push({ rowIndex: rows.length - 1, comparableText, sourceEvent: completionEvent });
|
||||
return rows.length - 1;
|
||||
}
|
||||
|
||||
function assistantSnapshotDecision(nextText: string, previousText: string, nextEvent: TraceEvent, previousEvent: TraceEvent): AssistantSnapshotDecision {
|
||||
if (!sameAssistantSnapshotIdentity(nextEvent, previousEvent)) return { action: "replace" };
|
||||
if (isAuthoritativeAssistantEvent(nextEvent) || isAuthoritativeAssistantEvent(previousEvent)) return { action: "replace" };
|
||||
const previousIndex = nextText.indexOf(previousText);
|
||||
if (previousIndex < 0) return { action: "replace" };
|
||||
const suffixText = nextText.slice(previousIndex + previousText.length).trim();
|
||||
if (suffixText && suffixText.length >= Math.max(80, previousText.length / 3)) return { action: "append-suffix", suffixText };
|
||||
if (previousIndex > 0 || !suffixText) return { action: "keep-existing" };
|
||||
return { action: "replace" };
|
||||
}
|
||||
|
||||
function sameAssistantSnapshotIdentity(left: TraceEvent, right: TraceEvent): boolean {
|
||||
const leftItem = nonEmptyString(left.itemId ?? left.messageId ?? left.id);
|
||||
const rightItem = nonEmptyString(right.itemId ?? right.messageId ?? right.id);
|
||||
if (leftItem && rightItem) return leftItem === rightItem;
|
||||
return false;
|
||||
}
|
||||
|
||||
function isAuthoritativeAssistantEvent(event: TraceEvent): boolean {
|
||||
return event.replyAuthority === true || event.final === true || event.terminal === true || String(event.status ?? "") === "completed";
|
||||
}
|
||||
|
||||
function mergeAssistantRows(preferred: TraceEventRow, metadata: TraceEventRow): TraceEventRow {
|
||||
return {
|
||||
...preferred,
|
||||
terminal: preferred.terminal || metadata.terminal ? true : undefined,
|
||||
header: metadata.terminal && !preferred.terminal ? metadata.header : preferred.header,
|
||||
tone: preferred.tone === "ok" || metadata.tone === "ok" ? "ok" : preferred.tone
|
||||
};
|
||||
}
|
||||
|
||||
function comparableAssistantText(value: unknown): string {
|
||||
return cleanTraceText(value).replace(/\s+/gu, " ");
|
||||
}
|
||||
|
||||
function traceFinalResponseText(trace: Record<string, unknown>): string | null {
|
||||
const response = trace.finalResponse && typeof trace.finalResponse === "object" ? trace.finalResponse as Record<string, unknown> : null;
|
||||
return nonEmptyString(response?.text ?? response?.content ?? response?.message);
|
||||
}
|
||||
|
||||
function markAssistantRowTerminal(trace: Record<string, unknown>, row: TraceEventRow, completionEvent: TraceEvent): TraceEventRow {
|
||||
return {
|
||||
...row,
|
||||
tone: "ok",
|
||||
terminal: true,
|
||||
header: `${row.header}，轮次完成（总耗时 ${formatTraceDuration(traceRelativeMs(trace, completionEvent))}）`
|
||||
};
|
||||
}
|
||||
|
||||
function traceCompletionSummaryRow(trace: Record<string, unknown>, event: TraceEvent): TraceEventRow {
|
||||
return {
|
||||
rowId: `trace-completion:${event.seq ?? "turn"}`,
|
||||
|
||||
@@ -78,6 +78,32 @@ test("R1 shared trace renderer suppresses AgentRun reuse diagnostics", () => {
|
||||
assert.doesNotMatch(rows[0]?.header ?? "", /reused|ensured|runner-job/u);
|
||||
});
|
||||
|
||||
test("R1 shared trace renderer keeps terminal assistant at its own event position", () => {
|
||||
const rows = traceDisplayRows({ traceId: "trc_terminal_position", status: "completed", startedAt: "2026-06-20T12:00:00.000Z" }, [
|
||||
{ seq: 1, label: "agentrun:assistant:message", type: "assistant", status: "running", message: "FINAL", createdAt: "2026-06-20T12:00:01.000Z" },
|
||||
{ seq: 2, label: "item/commandExecution:completed", type: "tool_call", toolName: "commandExecution", status: "completed", command: "make test", createdAt: "2026-06-20T12:00:02.000Z" },
|
||||
{ seq: 3, label: "agentrun:assistant:message", type: "assistant", status: "completed", message: "FINAL", final: true, replyAuthority: true, terminal: true, createdAt: "2026-06-20T12:00:03.000Z" }
|
||||
]);
|
||||
const toolIndex = rows.findIndex((row) => row.seq === 2);
|
||||
const finalIndex = rows.findIndex((row) => row.seq === 3 && row.terminal === true && row.body === "FINAL");
|
||||
assert.ok(toolIndex >= 0);
|
||||
assert.ok(finalIndex > toolIndex);
|
||||
assert.equal(rows.some((row) => row.seq === 1 && row.terminal === true), false);
|
||||
});
|
||||
|
||||
test("R1 shared trace renderer keeps completion final response at completion event position", () => {
|
||||
const rows = traceDisplayRows({ traceId: "trc_completion_position", status: "completed", startedAt: "2026-06-20T12:10:00.000Z", finalResponse: { text: "FINAL" } }, [
|
||||
{ seq: 1, label: "agentrun:assistant:message", type: "assistant", status: "running", message: "FINAL", createdAt: "2026-06-20T12:10:01.000Z" },
|
||||
{ seq: 2, label: "item/commandExecution:completed", type: "tool_call", toolName: "commandExecution", status: "completed", command: "make test", createdAt: "2026-06-20T12:10:02.000Z" },
|
||||
{ seq: 3, label: "agentrun:result:completed", type: "result", status: "completed", terminal: true, createdAt: "2026-06-20T12:10:03.000Z" }
|
||||
]);
|
||||
const toolIndex = rows.findIndex((row) => row.seq === 2);
|
||||
const finalIndex = rows.findIndex((row) => row.seq === 3 && row.terminal === true && row.body === "FINAL");
|
||||
assert.ok(toolIndex >= 0);
|
||||
assert.ok(finalIndex > toolIndex);
|
||||
assert.equal(rows.some((row) => row.seq === 1 && row.terminal === true), false);
|
||||
});
|
||||
|
||||
test("R1 trace API snapshots stay authoritative over compact result traces", () => {
|
||||
const previous = {
|
||||
traceId: "trc_merge",
|
||||
|
||||
Reference in New Issue
Block a user