fix: seal workbench failed terminal body
This commit is contained in:
UniDesk Codex
@@ -1317,6 +1317,67 @@ test("workbench read model seals turn from terminal message projection when chec
|
||||
}
|
||||
});
|
||||
|
||||
test("workbench read model exposes failed terminal body when final response part is missing", async () => {
|
||||
const sessionId = "ses_workbench_failed_terminal_missing_part";
|
||||
const traceId = "trc_failed_terminal_missing_part";
|
||||
const startedAt = "2026-06-29T22:19:32.000Z";
|
||||
const finishedAt = "2026-06-29T22:19:47.000Z";
|
||||
const facts = emptyFacts();
|
||||
facts.sessions.push({
|
||||
sessionId,
|
||||
ownerUserId: ACTOR.id,
|
||||
ownerRole: ACTOR.role,
|
||||
agentId: "hwlab-code-agent",
|
||||
status: "failed",
|
||||
lastTraceId: traceId,
|
||||
projectedSeq: 40,
|
||||
sourceSeq: 40,
|
||||
terminal: true,
|
||||
sealed: true,
|
||||
createdAt: startedAt,
|
||||
updatedAt: finishedAt,
|
||||
valuesRedacted: true
|
||||
});
|
||||
facts.messages.push(
|
||||
{ messageId: "msg_failed_missing_part_user", sessionId, turnId: traceId, traceId, role: "user", status: "sent", text: "hi", projectedSeq: 39, sourceSeq: 39, createdAt: startedAt, updatedAt: startedAt, valuesRedacted: true },
|
||||
{ messageId: "msg_failed_missing_part_agent", sessionId, turnId: traceId, traceId, role: "agent", status: "failed", text: "", projectedSeq: 40, sourceSeq: 40, terminal: true, sealed: true, timing: { startedAt, lastEventAt: finishedAt, finishedAt, durationMs: 15000, valuesRedacted: true }, startedAt, lastEventAt: finishedAt, finishedAt, durationMs: 15000, createdAt: startedAt, updatedAt: finishedAt, valuesRedacted: true }
|
||||
);
|
||||
facts.turns.push({ turnId: traceId, sessionId, traceId, messageId: "msg_failed_missing_part_agent", status: "failed", projectedSeq: 40, sourceSeq: 40, terminal: true, sealed: true, finalResponse: null, diagnostic: { blocker: { code: "provider-stream-disconnected", message: "provider stream disconnected" }, projectionStatus: "caught_up", projectionHealth: "healthy", valuesRedacted: true }, startedAt, lastEventAt: finishedAt, finishedAt, durationMs: 15000, updatedAt: finishedAt, valuesRedacted: true });
|
||||
facts.checkpoints.push({ traceId, sessionId, turnId: traceId, runId: "run_failed_missing_part", commandId: "cmd_failed_missing_part", status: "failed", projectionStatus: "caught_up", projectionHealth: "healthy", projectedSeq: 40, sourceSeq: 40, terminal: true, sealed: true, diagnostic: { blocker: { code: "provider-stream-disconnected", message: "provider stream disconnected" }, valuesRedacted: true }, startedAt, lastEventAt: finishedAt, finishedAt, durationMs: 15000, updatedAt: finishedAt, valuesRedacted: true });
|
||||
const runtimeStore = createRuntimeStoreFromFacts(facts);
|
||||
const accessController = {
|
||||
async ensureBootstrap() {},
|
||||
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
|
||||
};
|
||||
const server = createCloudApiServer({ accessController, runtimeStore });
|
||||
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
|
||||
try {
|
||||
const { port } = server.address();
|
||||
const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(sessionId)}`);
|
||||
assert.equal(sessions.status, 200);
|
||||
assert.equal(sessions.body.sessions[0].status, "failed");
|
||||
assert.equal(sessions.body.sessions[0].terminal, true);
|
||||
|
||||
const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(sessionId)}/messages?limit=10`);
|
||||
assert.equal(messages.status, 200);
|
||||
const assistant = messages.body.messages.find((message) => message.role === "agent");
|
||||
assert.equal(assistant.status, "failed");
|
||||
assert.match(assistant.text, /provider-stream-disconnected/u);
|
||||
assert.equal(assistant.parts[0].type, "final_response");
|
||||
assert.match(assistant.parts[0].text, /provider-stream-disconnected/u);
|
||||
|
||||
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
|
||||
assert.equal(turn.status, 200);
|
||||
assert.equal(turn.body.turn.status, "failed");
|
||||
assert.equal(turn.body.turn.running, false);
|
||||
assert.equal(turn.body.turn.terminal, true);
|
||||
assert.match(turn.body.turn.finalResponse.text, /provider-stream-disconnected/u);
|
||||
} finally {
|
||||
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
|
||||
}
|
||||
});
|
||||
|
||||
test("workbench read model does not expose trace-only memory without visible session or result owner", async () => {
|
||||
const traceStore = createCodeAgentTraceStore();
|
||||
const traceId = "trc_workbench_trace_only";
|
||||
|
||||
@@ -15,7 +15,7 @@ import {
|
||||
} from "./server-http-utils.ts";
|
||||
import { createWorkbenchReadModel } from "./workbench-read-model.ts";
|
||||
import { createWorkbenchRuntimeClient } from "./workbench-runtime-client.ts";
|
||||
import { durableTraceStatus, RUNNING_STATUSES, TERMINAL_STATUSES } from "./workbench-turn-projection.ts";
|
||||
import { durableTraceStatus, RUNNING_STATUSES, terminalFinalResponse, TERMINAL_STATUSES } from "./workbench-turn-projection.ts";
|
||||
import { emitCodeAgentOtelSpan, emitHttpServerRequestSpan } from "./otel-trace.ts";
|
||||
|
||||
const DEFAULT_PAGE_LIMIT = 50;
|
||||
@@ -1189,10 +1189,17 @@ function factMessageDto(message, parts = [], facts = {}) {
|
||||
const messageStatus = normalizeStatus(message?.status);
|
||||
const assistantLike = isAssistantLikeRole(role);
|
||||
const legacyText = projectionText(message?.text, message?.content, message?.message, message?.finalResponse);
|
||||
const normalizedParts = parts.length > 0
|
||||
const baseParts = parts.length > 0
|
||||
? [...parts].sort(compareFactPartsAsc).map((part) => factPartDto(part, messageId, traceId)).filter(Boolean)
|
||||
: !assistantLike && legacyText ? [partFact({ type: "text", text: legacyText, status: message?.status }, 0, messageId, traceId)] : [];
|
||||
const checkpointStatus = assistantLike ? normalizeTerminalStatus(checkpoint?.status) : null;
|
||||
const preliminaryStatus = normalizeStatus(checkpointStatus ?? (assistantLike ? turn?.status : null) ?? messageStatus);
|
||||
const syntheticTerminalFinalResponse = assistantLike && !factPartsHaveFinalResponse(baseParts)
|
||||
? factSyntheticTerminalFinalResponse(preliminaryStatus, traceId, message, checkpoint, turn)
|
||||
: null;
|
||||
const normalizedParts = syntheticTerminalFinalResponse
|
||||
? [...baseParts, partFact({ type: "final_response", text: syntheticTerminalFinalResponse.text, status: syntheticTerminalFinalResponse.status }, baseParts.length, messageId, traceId)]
|
||||
: baseParts;
|
||||
const messageTerminalStatus = assistantLike ? factTerminalStatusFromMessageProjection(message, normalizedParts) : null;
|
||||
const status = normalizeStatus(checkpointStatus ?? messageTerminalStatus ?? (assistantLike ? turn?.status : null) ?? messageStatus);
|
||||
const timing = assistantLike
|
||||
@@ -1228,6 +1235,16 @@ function factMessageAuthorityText({ role, status, parts = [], legacyText = null
|
||||
return firstFactPartText(parts, "final_response") ?? "";
|
||||
}
|
||||
|
||||
function factPartsHaveFinalResponse(parts = []) {
|
||||
return parts.some((part) => part?.type === "final_response" && Boolean(projectionText(part?.text)));
|
||||
}
|
||||
|
||||
function factSyntheticTerminalFinalResponse(status, traceId = null, ...records) {
|
||||
const terminalStatus = normalizeTerminalStatus(status);
|
||||
if (!terminalStatus || terminalStatus === "completed") return null;
|
||||
return terminalFinalResponse(terminalStatus, { traceId, status: terminalStatus, records }, { evidence: records });
|
||||
}
|
||||
|
||||
function isTerminalProjectionStatus(status) {
|
||||
return TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status);
|
||||
}
|
||||
|
||||
@@ -273,6 +273,92 @@ test("workbench projection writer keeps completed AgentRun without final respons
|
||||
assert.equal(facts.checkpoints[0].sealed, false);
|
||||
});
|
||||
|
||||
test("workbench projection writer seals failed AgentRun turns with failure final response", async () => {
|
||||
const factWrites = [];
|
||||
const runtimeStore = {
|
||||
async writeWorkbenchFacts(params, requestMeta) {
|
||||
factWrites.push({ params, requestMeta });
|
||||
return { written: true, facts: params.facts };
|
||||
}
|
||||
};
|
||||
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:03:00.000Z"
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
await writeWorkbenchProjectionSession({
|
||||
accessController,
|
||||
runtimeStore,
|
||||
traceId: "trc_writer_terminal_failed",
|
||||
ownerUserId: "usr_writer",
|
||||
ownerRole: "user",
|
||||
sessionId: "ses_writer_terminal_failed",
|
||||
projectId: "prj_writer",
|
||||
conversationId: "cnv_writer_terminal_failed",
|
||||
threadId: "thread-writer-terminal-failed",
|
||||
status: "failed",
|
||||
payload: {
|
||||
traceId: "trc_writer_terminal_failed",
|
||||
status: "failed",
|
||||
error: { code: "provider-stream-disconnected", message: "provider stream disconnected" },
|
||||
runnerTrace: {
|
||||
traceId: "trc_writer_terminal_failed",
|
||||
status: "failed",
|
||||
startedAt: "2026-06-20T11:02:30.000Z",
|
||||
lastEventAt: "2026-06-20T11:03:00.000Z",
|
||||
finishedAt: "2026-06-20T11:03:00.000Z",
|
||||
events: [
|
||||
{ seq: 1, type: "assistant_message", status: "running", message: "progress only" },
|
||||
{ seq: 2, type: "result", status: "failed", terminal: true, label: "agentrun:terminal:failed", errorCode: "provider-stream-disconnected" }
|
||||
],
|
||||
eventCount: 2
|
||||
},
|
||||
agentRun: { runId: "run_writer_terminal_failed", commandId: "cmd_writer_terminal_failed", status: "failed", terminalStatus: "failed", failureKind: "provider-stream-disconnected", lastSeq: 2 },
|
||||
updatedAt: "2026-06-20T11:03:00.000Z"
|
||||
},
|
||||
session: {
|
||||
sessionStatus: "failed",
|
||||
messages: [
|
||||
{ messageId: "msg_writer_failed_user", role: "user", text: "question", status: "sent", turnId: "trc_writer_terminal_failed", traceId: "trc_writer_terminal_failed" },
|
||||
{ messageId: "msg_writer_failed_agent", role: "agent", text: "", status: "failed", turnId: "trc_writer_terminal_failed", traceId: "trc_writer_terminal_failed" }
|
||||
]
|
||||
}
|
||||
});
|
||||
|
||||
assert.equal(factWrites.length, 1);
|
||||
const facts = factWrites[0].params.facts;
|
||||
const agentMessage = facts.messages.find((message) => message.messageId === "msg_writer_failed_agent");
|
||||
const finalPart = facts.parts.find((part) => part.messageId === "msg_writer_failed_agent" && part.partType === "final_response");
|
||||
assert.equal(facts.sessions[0].status, "failed");
|
||||
assert.equal(facts.sessions[0].terminal, true);
|
||||
assert.equal(facts.sessions[0].sealed, true);
|
||||
assert.equal(agentMessage.status, "failed");
|
||||
assert.equal(agentMessage.terminal, true);
|
||||
assert.equal(agentMessage.sealed, true);
|
||||
assert.match(agentMessage.text, /provider-stream-disconnected/u);
|
||||
assert.equal(finalPart?.status, "failed");
|
||||
assert.match(finalPart?.text ?? "", /provider-stream-disconnected/u);
|
||||
assert.equal(facts.turns[0].status, "failed");
|
||||
assert.equal(facts.turns[0].terminal, true);
|
||||
assert.equal(facts.turns[0].sealed, true);
|
||||
assert.match(facts.turns[0].finalResponse.text, /provider-stream-disconnected/u);
|
||||
assert.equal(facts.checkpoints[0].projectionStatus, "caught_up");
|
||||
assert.equal(facts.checkpoints[0].terminal, true);
|
||||
assert.equal(facts.checkpoints[0].sealed, true);
|
||||
assert.match(facts.checkpoints[0].finalResponse.text, /provider-stream-disconnected/u);
|
||||
});
|
||||
|
||||
test("workbench projection writer does not synthesize terminal duration from updatedAt", async () => {
|
||||
const factWrites = [];
|
||||
const runtimeStore = {
|
||||
|
||||
@@ -7,7 +7,7 @@ import { createHash } from "node:crypto";
|
||||
import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
|
||||
import { emitCodeAgentOtelSpan } from "./otel-trace.ts";
|
||||
import { safeTraceId } from "./server-http-utils.ts";
|
||||
import { createWorkbenchTurnProjection, normalizeWorkbenchStatus, projectionDiagnostics, TERMINAL_STATUSES } from "./workbench-turn-projection.ts";
|
||||
import { createWorkbenchTurnProjection, normalizeWorkbenchStatus, projectionDiagnostics, terminalFinalResponse, 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;
|
||||
@@ -178,7 +178,8 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
|
||||
if (projectedSeq <= 0) throw new Error("Workbench projection allocator returned an invalid projectedSeq.");
|
||||
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 eventTerminalFinalResponse = terminal ? terminalFinalResponse(status, event, { evidence: event, finalResponse: event.finalResponse }) : null;
|
||||
const messageFinalText = terminal ? finalResponseTextValue(event.finalResponse, event.assistantText, event.reply, eventTerminalFinalResponse, 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;
|
||||
const messageFact = sessionId && !suppressedAfterSeal ? normalizeMessageFact({
|
||||
messageId: eventProjectionAssistantMessageId(traceId, event),
|
||||
|
||||
@@ -197,20 +197,22 @@ function terminalTurnEvidence({ result = null, traceTerminal = null } = {}) {
|
||||
return traceTerminal;
|
||||
}
|
||||
|
||||
function terminalFinalResponse(status, result = null, traceTerminal = null) {
|
||||
if (normalizeWorkbenchStatus(status) === "canceled") return canonicalCancelFinalResponse(result, traceTerminal);
|
||||
export function terminalFinalResponse(status, result = null, traceTerminal = null) {
|
||||
const normalizedStatus = normalizeWorkbenchStatus(status);
|
||||
if (normalizedStatus === "canceled") return canonicalCancelFinalResponse(result, traceTerminal);
|
||||
const direct = traceTerminal?.finalResponse ?? result?.finalResponse;
|
||||
if (direct) return direct;
|
||||
const resultText = projectionText(result?.assistantText, result?.finalText, result?.reply);
|
||||
if (resultText) {
|
||||
return {
|
||||
text: resultText,
|
||||
status: normalizeWorkbenchStatus(status),
|
||||
status: normalizedStatus,
|
||||
traceId: textValue(result?.traceId ?? traceTerminal?.finalResponse?.traceId ?? traceTerminal?.evidence?.traceId) || null,
|
||||
source: "result-final-response-evidence",
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
if (normalizedStatus !== "completed") return terminalFailureFinalResponse(normalizedStatus, result, traceTerminal);
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -223,6 +225,42 @@ function canonicalCancelFinalResponse(result = null, traceTerminal = null) {
|
||||
};
|
||||
}
|
||||
|
||||
function terminalFailureFinalResponse(status, result = null, traceTerminal = null) {
|
||||
const normalizedStatus = normalizeWorkbenchStatus(status);
|
||||
if (!normalizedStatus || normalizedStatus === "completed" || normalizedStatus === "unknown") return null;
|
||||
const failure = terminalFailureEvidence(result, traceTerminal?.evidence, traceTerminal);
|
||||
const failureKind = normalizedProjectionFailureKind(failure?.failureKind ?? failure?.errorCode ?? failure?.code ?? failure?.name ?? failure?.status ?? normalizedStatus);
|
||||
const label = failureKind && failureKind !== "unknown" ? failureKind : normalizedStatus;
|
||||
return {
|
||||
text: `Workbench terminal ${normalizedStatus}: ${label}`,
|
||||
status: normalizedStatus,
|
||||
traceId: textValue(result?.traceId ?? traceTerminal?.finalResponse?.traceId ?? traceTerminal?.evidence?.traceId) || null,
|
||||
source: "terminal-failure-diagnostic",
|
||||
failureKind: label,
|
||||
valuesPrinted: false,
|
||||
valuesRedacted: true
|
||||
};
|
||||
}
|
||||
|
||||
function terminalFailureEvidence(...values) {
|
||||
for (const value of values) {
|
||||
if (Array.isArray(value)) {
|
||||
const nested = terminalFailureEvidence(...value);
|
||||
if (nested) return nested;
|
||||
continue;
|
||||
}
|
||||
const record = objectValue(value);
|
||||
if (!record) continue;
|
||||
const kind = normalizedProjectionFailureKind(record.failureKind ?? record.errorCode ?? record.code ?? record.name);
|
||||
if (kind && kind !== "unknown") return record;
|
||||
for (const nested of [record.error, record.blocker, record.diagnostic, record.terminalEvidence, record.providerTrace, record.agentRun, record.payload, record.evidence, record.records]) {
|
||||
const found = terminalFailureEvidence(nested);
|
||||
if (found) return found;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function resultHasTerminalAuthority(result = null, traceTerminal = null) {
|
||||
if (!result || typeof result !== "object") return false;
|
||||
if (traceTerminal) return true;
|
||||
|
||||
@@ -30,3 +30,28 @@ test("server state reducer seals completed message when final response exists",
|
||||
assert.equal(message.text, "final answer");
|
||||
assert.equal(selectSessionStatusAuthority(state)[sessionId]?.status, "completed");
|
||||
});
|
||||
|
||||
test("server state reducer keeps sealed terminal messages when a stale partial snapshot arrives", () => {
|
||||
const sessionId = "ses_state_partial_snapshot";
|
||||
const traceId = "trc_state_partial_snapshot";
|
||||
let state = createWorkbenchServerState();
|
||||
state = reduceWorkbenchServerState(state, { type: "session.detail", session: { sessionId, status: "running", messages: [] } as any });
|
||||
state = reduceWorkbenchServerState(state, { type: "session.messages", sessionId, messages: [
|
||||
{ id: "msg_state_partial_user_1", role: "user", status: "sent", text: "one", sessionId, traceId: "trc_state_partial_1" } as any,
|
||||
{ id: "msg_state_partial_agent_1", role: "agent", status: "completed", text: "one done", sessionId, traceId: "trc_state_partial_1" } as any,
|
||||
{ id: "msg_state_partial_user_2", role: "user", status: "sent", text: "two", sessionId, traceId } as any,
|
||||
{ id: "msg_state_partial_agent_2", role: "agent", status: "completed", text: "final answer", finalResponse: { text: "final answer" }, sessionId, traceId, finishedAt: "2026-07-01T00:00:04.000Z", durationMs: 4000 } as any
|
||||
] });
|
||||
state = reduceWorkbenchServerState(state, { type: "session.messages", sessionId, messages: [
|
||||
{ id: "msg_state_partial_user_1", role: "user", status: "sent", text: "one", sessionId, traceId: "trc_state_partial_1" } as any,
|
||||
{ id: "msg_state_partial_agent_1", role: "agent", status: "completed", text: "one done", sessionId, traceId: "trc_state_partial_1" } as any
|
||||
] });
|
||||
|
||||
const messages = selectActiveMessages(state, sessionId);
|
||||
assert.equal(messages.length, 4);
|
||||
const terminal = messages.find((message) => message.id === "msg_state_partial_agent_2") as any;
|
||||
assert.equal(terminal.status, "completed");
|
||||
assert.equal(terminal.text, "final answer");
|
||||
assert.equal(terminal.finalResponse.text, "final answer");
|
||||
assert.equal(selectSessionStatusAuthority(state)[sessionId]?.status, "completed");
|
||||
});
|
||||
|
||||
@@ -200,11 +200,17 @@ function mergeMessageSnapshot(existing: ChatMessage, incoming: ChatMessage): Cha
|
||||
}
|
||||
|
||||
function mergeMessageList(existing: ChatMessage[], incoming: ChatMessage[]): ChatMessage[] {
|
||||
return incoming.map((rawMessage) => {
|
||||
const merged = existing.map((message) => normalizeUnsealedCompletedMessage(message));
|
||||
for (const rawMessage of incoming) {
|
||||
const message = normalizeUnsealedCompletedMessage(rawMessage);
|
||||
const previous = existing.find((item) => messageMatchesSnapshot(item, message));
|
||||
return previous ? sealExistingMessageTiming(previous, message) : message;
|
||||
});
|
||||
const index = merged.findIndex((item) => messageMatchesSnapshot(item, message));
|
||||
if (index >= 0) {
|
||||
merged[index] = sealExistingMessageTiming(merged[index], message);
|
||||
} else {
|
||||
merged.push(message);
|
||||
}
|
||||
}
|
||||
return merged;
|
||||
}
|
||||
|
||||
function normalizeUnsealedCompletedMessage(message: ChatMessage): ChatMessage {
|
||||
|
||||
Reference in New Issue
Block a user