fix: seal terminal turn directly in workbench

2026-07-02 13:42:29 +00:00
parent c2629b7b08
commit 3c2cd90608
4 changed files with 150 additions and 38 deletions
@@ -5,7 +5,7 @@ import assert from "node:assert/strict";
 import test from "node:test";

 import type { AgentChatResultResponse, ChatMessage } from "../src/types/index.ts";
-import { clearRunnerTraceTransientDiagnostics, mergeTerminalResultTrace, messageStatusPatchForTerminalMerge, messageTimingPatchForMerge, messageTimingSealPatchForProjectionMerge, nonBlockingProjection, normalizeAgentError, projectionFromResult, shouldClearCompletedTurnDiagnostics, terminalMessagePatchFromTurnResult, terminalMessageTimingPatchForNormalize } from "../src/stores/workbench-message-projection-runtime.ts";
+import { clearRunnerTraceTransientDiagnostics, mergeTerminalResultTrace, messageHasTerminalResponse, messageStatusPatchForTerminalMerge, messageTimingPatchForMerge, messageTimingSealPatchForProjectionMerge, nonBlockingProjection, normalizeAgentError, projectionFromResult, shouldClearCompletedTurnDiagnostics, terminalAuthorityMessageFromTurnResult, terminalMessagePatchFromTurnResult, terminalMessageTimingPatchForNormalize } from "../src/stores/workbench-message-projection-runtime.ts";

 test("terminal result merge preserves runner trace evidence and timing", () => {
  const previous: ChatMessage["runnerTrace"] = {
@@ -109,6 +109,37 @@ test("terminal turn result seals completed final response body", () => {
  assert.equal((terminal?.finalResponse as { text?: string } | undefined)?.text, "final answer");
 });

+test("terminal turn authority can project final response without a session message page row", () => {
+  const terminal = terminalAuthorityMessageFromTurnResult({
+    traceId: "trc_direct_terminal",
+    sessionId: "ses_direct_terminal",
+    threadId: "thr_direct_terminal",
+    status: "completed",
+    terminal: true,
+    finalResponse: { text: "direct final body" },
+    startedAt: "2026-07-02T13:18:40.000Z",
+    finishedAt: "2026-07-02T13:18:47.000Z"
+  } as AgentChatResultResponse);
+
+  assert.equal(terminal?.id, "msg_trc_direct_terminal_terminal_authority");
+  assert.equal(terminal?.status, "completed");
+  assert.equal(terminal?.traceAutoLifecycle, "terminal");
+  assert.equal(terminal?.text, "direct final body");
+  assert.equal(messageHasTerminalResponse(terminal), true);
+});
+
+test("terminal turn authority does not synthesize an empty completed message", () => {
+  const terminal = terminalAuthorityMessageFromTurnResult({
+    traceId: "trc_direct_terminal_empty",
+    status: "completed",
+    terminal: true,
+    terminalSealBlocked: true,
+    waitingFor: "final_response"
+  } as AgentChatResultResponse);
+
+  assert.equal(terminal, null);
+});
+
 test("terminal turn result accepts authoritative assistant trace final", () => {
  const restored = {
    role: "agent",
@@ -8,6 +8,7 @@ import type { AgentChatResultResponse, ChatMessage } from "../types";
 import {
  messageHasTerminalResponse,
  messageNeedsTerminalDiagnostics,
+  terminalAuthorityMessageFromTurnResult,
  terminalMessagePatchFromTurnResult,
  traceHasTerminalResponse,
  turnResultIsTerminalForMerge,
@@ -76,6 +77,28 @@ test("turn result merge seals completed result with final response", () => {
  assert.equal(patch?.text, "final answer");
 });

+test("turn authority creates terminal agent message when projection row is missing", () => {
+  const traceId = "trc_frontend_terminal_missing_row";
+  const message = terminalAuthorityMessageFromTurnResult(turnResult({
+    traceId,
+    sessionId: "ses_terminal_missing_row",
+    threadId: "thr_terminal_missing_row",
+    status: "completed",
+    terminal: true,
+    finalResponse: { text: "final answer without session message row" },
+    startedAt: "2026-07-02T13:18:40.000Z",
+    finishedAt: "2026-07-02T13:18:47.000Z"
+  }));
+
+  assert.equal(message?.role, "agent");
+  assert.equal(message?.traceId, traceId);
+  assert.equal(message?.sessionId, "ses_terminal_missing_row");
+  assert.equal(message?.status, "completed");
+  assert.equal(message?.text, "final answer without session message row");
+  assert.equal(record(message?.finalResponse)?.source, "turn-result");
+  assert.equal(messageHasTerminalResponse(message), true);
+});
+
 test("turn result merge seals failed result with failure final response", () => {
  const traceId = "trc_frontend_terminal_failed_with_body";
  const message = agentMessage({ id: "msg_frontend_terminal_failed_with_body", status: "running", traceId });
@@ -111,11 +134,14 @@ test("terminal response helpers treat failed body as sealed authority", () => {

 test("workbench active terminal paths seal final response from turn authority", () => {
  const source = fs.readFileSync(path.join(storeDir, "workbench.ts"), "utf8");
-  const terminalPatchUses = source.match(/terminalMessagePatchFromTurnResult\(message, result(?: as AgentChatResultResponse)?\) \?\? \{\}/gu) ?? [];
-  const syncBlock = source.slice(source.indexOf("function syncTurnStatusToMessage"), source.indexOf("async function submitMessage"));
+  const projectBlock = source.slice(source.indexOf("function projectTurnAuthorityToMessages"), source.indexOf("async function submitMessage"));
+  const realtimeTurnBlock = source.slice(source.indexOf("function applyRealtimeTurnSnapshot"), source.indexOf("async function refreshTerminalTraceFromRest"));
+  const terminalRestBlock = source.slice(source.indexOf("async function refreshTerminalTraceFromRest"), source.indexOf("function installRealtimeVisibilityHandler"));
  const completeBlock = source.slice(source.indexOf("function completeTrace"), source.indexOf("async function hydrateTerminalMessageDiagnostics"));

-  assert.equal(terminalPatchUses.length, 2);
-  assert.match(syncBlock, /\.\.\.terminalPatch/u);
-  assert.match(completeBlock, /\.\.\.terminalPatch/u);
+  assert.match(projectBlock, /terminalAuthorityMessageFromTurnResult\(result\)/u);
+  assert.match(projectBlock, /\.\.\.terminalPatch/u);
+  assert.match(realtimeTurnBlock, /!terminalTurn[\s\S]*refreshRealtimeSessionMessages/u);
+  assert.match(terminalRestBlock, /traceProjectionIsTerminalSealed\(id, ownerMessagesBeforeProjection\)[\s\S]*return/u);
+  assert.match(completeBlock, /projectTurnAuthorityToMessages\(traceId, result, "complete-trace"\)/u);
 });
@@ -324,6 +324,41 @@ export function terminalMessagePatchFromTurnResult(message: ChatMessage, result:
  };
 }

+export function terminalAuthorityMessageFromTurnResult(result: AgentChatResultResponse): ChatMessage | null {
+  const traceId = firstNonEmptyString(result.traceId);
+  if (!traceId) return null;
+  const now = new Date().toISOString();
+  const sessionId = firstNonEmptyString(result.sessionId) ?? null;
+  const threadId = firstNonEmptyString(result.threadId) ?? null;
+  const createdAt = firstNonEmptyString(result.startedAt, result.createdAt, result.updatedAt, now) ?? now;
+  const messageId = `msg_${traceId}_terminal_authority`;
+  const seed: ChatMessage = {
+    id: messageId,
+    messageId,
+    role: "agent",
+    title: "Code Agent",
+    text: "",
+    status: "running",
+    traceId,
+    sessionId,
+    threadId,
+    createdAt,
+    updatedAt: now,
+    traceAutoLifecycle: "running",
+    runnerTrace: {
+      traceId,
+      sessionId: sessionId ?? undefined,
+      threadId: threadId ?? undefined,
+      status: "running",
+      updatedAt: now
+    }
+  };
+  const patch = terminalMessagePatchFromTurnResult(seed, result);
+  if (!patch) return null;
+  const message = { ...seed, ...patch };
+  return messageHasTerminalResponse(message) ? message : null;
+}
+
 function terminalMessageBodyPatchFromTurnResult(message: ChatMessage, result: AgentChatResultResponse, resultStatus: string | null): Partial<ChatMessage> {
  const status = normalizedStatusText(resultStatus);
  if (!isTerminalMessageStatus(status)) return {};
@@ -36,6 +36,7 @@ import {
  messageNeedsTerminalDiagnostics,
  messageNeedsTraceHydration,
  messageStatusPatchForTerminalMerge,
+  terminalAuthorityMessageFromTurnResult,
  terminalMessagePatchFromTurnResult,
  messageText,
  messageTimingPatch,
@@ -616,22 +617,50 @@ export const useWorkbenchStore = defineStore("workbench", () => {
  }

  function syncTurnStatusToMessage(traceId: string, result: AgentChatResultResponse | TraceSnapshot): void {
+    projectTurnAuthorityToMessages(traceId, result as AgentChatResultResponse, "turn-status");
+  }
+
+  function projectTurnAuthorityToMessages(traceId: string, result: AgentChatResultResponse, reason: string): boolean {
    const authoritySessionId = traceResultSessionId(result);
-    updateTraceMessages(traceId, authoritySessionId, (message) => {
-      const resultStatus = turnResultStatusForMerge(result as AgentChatResultResponse);
-      const terminal = turnResultIsTerminalForMerge(result as AgentChatResultResponse);
-      const resultError = normalizeAgentError((result as AgentChatResultResponse).error ?? null);
-      const resultProjection = projectionFromResult(result as AgentChatResultResponse);
-      const mergedRunnerTrace = mergeTerminalResultTrace(message.runnerTrace, result as AgentChatResultResponse);
-      const clearCompletedDiagnostics = shouldClearCompletedTurnDiagnostics(resultStatus, resultError);
-      const runnerTrace = clearCompletedDiagnostics ? clearRunnerTraceTransientDiagnostics(mergedRunnerTrace) : mergedRunnerTrace;
-      rememberTraceAuthority(runnerTrace);
-      const error = resultError ?? (terminal || clearCompletedDiagnostics ? null : normalizeAgentError(runnerTrace?.error ?? message.error));
-      const agentRun = agentRunFromResult(result as AgentChatResultResponse, runnerTrace) ?? agentRunFromMessage(message);
-      const projection = clearCompletedDiagnostics ? nonBlockingProjection(resultProjection) : resultProjection ?? (terminal ? null : runnerTrace.projection ?? message.projection ?? null);
-      const terminalPatch = terminalMessagePatchFromTurnResult(message, result as AgentChatResultResponse) ?? {};
-      return { ...message, ...messageTimingPatchForMerge(message, result), ...messageStatusPatchForTerminalMerge(message, resultStatus, terminal), runnerTrace, error, projection, projectionStatus: projection?.projectionStatus ?? null, projectionHealth: projection?.projectionHealth ?? null, blocker: projection?.blocker ?? null, agentRun: agentRun ?? undefined, ...terminalPatch, updatedAt: new Date().toISOString() };
+    const ownerSessionId = traceOwnerSessionId(traceId, authoritySessionId);
+    if (!ownerSessionId) return false;
+    let matched = false;
+    let sealed = false;
+    updateSessionMessages(ownerSessionId, (source) => {
+      const projected = source.map((message) => {
+        if (!messageMatchesTraceAuthority(message, traceId, authoritySessionId, ownerSessionId)) return message;
+        matched = true;
+        const wasSealed = messageHasSealedTerminalResult(message);
+        const next = projectTurnAuthorityMessage(message, result);
+        if (!wasSealed && messageHasSealedTerminalResult(next)) sealed = true;
+        return next;
+      });
+      if (matched) return projected;
+      const terminalMessage = terminalAuthorityMessageFromTurnResult(result);
+      if (!terminalMessage) return projected;
+      sealed = true;
+      return [...projected, terminalMessage];
    });
+    if (sealed) {
+      recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-authority", sessionId: ownerSessionId, traceId, outcome: "ok", diagnostic: { code: "terminal_direct_seal", reason, source: "turn-authority", valuesRedacted: true } });
+    }
+    return sealed;
+  }
+
+  function projectTurnAuthorityMessage(message: ChatMessage, result: AgentChatResultResponse): ChatMessage {
+    const resultStatus = turnResultStatusForMerge(result);
+    const terminal = turnResultIsTerminalForMerge(result);
+    const resultError = normalizeAgentError(result.error ?? null);
+    const resultProjection = projectionFromResult(result);
+    const mergedRunnerTrace = mergeTerminalResultTrace(message.runnerTrace, result);
+    const clearCompletedDiagnostics = shouldClearCompletedTurnDiagnostics(resultStatus, resultError);
+    const runnerTrace = clearCompletedDiagnostics ? clearRunnerTraceTransientDiagnostics(mergedRunnerTrace) : mergedRunnerTrace;
+    rememberTraceAuthority(runnerTrace);
+    const error = resultError ?? (terminal || clearCompletedDiagnostics ? null : normalizeAgentError(runnerTrace?.error ?? message.error));
+    const agentRun = agentRunFromResult(result, runnerTrace) ?? agentRunFromMessage(message);
+    const projection = clearCompletedDiagnostics ? nonBlockingProjection(resultProjection) : resultProjection ?? (terminal ? null : runnerTrace.projection ?? message.projection ?? null);
+    const terminalPatch = terminalMessagePatchFromTurnResult(message, result) ?? {};
+    return { ...message, ...messageTimingPatchForMerge(message, result), ...messageStatusPatchForTerminalMerge(message, resultStatus, terminal), runnerTrace, error, projection, projectionStatus: projection?.projectionStatus ?? null, projectionHealth: projection?.projectionHealth ?? null, blocker: projection?.blocker ?? null, agentRun: agentRun ?? undefined, ...terminalPatch, updatedAt: new Date().toISOString() };
  }

  async function submitMessage(text: string): Promise<boolean> {
@@ -1175,10 +1204,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
    if (!traceId) return;
    if (!shouldApplyActiveTraceAuthority(traceId, traceResultSessionId(turn))) return;
    const activeId = activeSessionId.value;
-    if (activeId && !messages.value.some((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId)) void refreshRealtimeSessionMessages(activeId, `realtime-turn-gap:${traceId}`);
    const status = firstNonEmptyString(turn.status) ?? undefined;
+    const terminalTurn = turn.terminal === true || isTerminalMessageStatus(status);
+    if (activeId && !terminalTurn && !messages.value.some((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId)) void refreshRealtimeSessionMessages(activeId, `realtime-turn-gap:${traceId}`);
    applyTurnStatusSnapshot(traceId, { ...turn, traceId, status, running: turn.running === true, terminal: turn.terminal === true, sessionId: firstNonEmptyString(turn.sessionId) ?? undefined, threadId: firstNonEmptyString(turn.threadId) ?? undefined, agentRun: turn.agentRun as AgentRunProvenance | undefined } as AgentChatResultResponse);
-    if (turn.terminal === true || isTerminalMessageStatus(status)) void refreshTerminalTraceFromRest(traceId, "realtime-turn-snapshot");
+    if (terminalTurn) void refreshTerminalTraceFromRest(traceId, "realtime-turn-snapshot");
  }

  async function refreshTerminalTraceFromRest(traceId: string, reason: string): Promise<void> {
@@ -1188,6 +1218,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
    if (ownerBefore === activeSessionId.value) recordActivity(reason);
    await refreshTurnStatusByTraceId(id);
    const ownerSessionId = traceOwnerSessionId(id, turnStatusAuthority.value[id]?.sessionId ?? null) ?? ownerBefore;
+    const ownerMessagesBeforeProjection = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value;
+    if (traceProjectionIsTerminalSealed(id, ownerMessagesBeforeProjection)) {
+      clearActiveTraceRestGapFill(id);
+      return;
+    }
    if (ownerSessionId) {
      scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs);
      await refreshMessageProjectionForTrace(ownerSessionId, id);
@@ -1287,22 +1322,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
    const ownerSessionId = traceOwnerSessionId(traceId, authoritySessionId);
    if (!ownerSessionId) return;
    if (ownerSessionId === activeSessionId.value) recordActivity(`trace:terminal:${firstNonEmptyString(result.lastEventLabel, result.status, "completed") ?? "completed"}`);
-    updateSessionMessages(ownerSessionId, (source) => source.map((message) => {
-      if (!messageMatchesTraceAuthority(message, traceId, authoritySessionId, ownerSessionId)) return message;
-      const resultStatus = turnResultStatusForMerge(result);
-      const terminal = turnResultIsTerminalForMerge(result);
-      const resultError = normalizeAgentError(result.error ?? null);
-      const resultProjection = projectionFromResult(result);
-      const mergedRunnerTrace = mergeTerminalResultTrace(message.runnerTrace, result);
-      const clearCompletedDiagnostics = shouldClearCompletedTurnDiagnostics(resultStatus, resultError);
-      const runnerTrace = clearCompletedDiagnostics ? clearRunnerTraceTransientDiagnostics(mergedRunnerTrace) : mergedRunnerTrace;
-      rememberTraceAuthority(runnerTrace);
-      const error = resultError ?? (terminal || clearCompletedDiagnostics ? null : normalizeAgentError(runnerTrace?.error ?? message.error));
-      const projection = clearCompletedDiagnostics ? nonBlockingProjection(resultProjection) : resultProjection ?? (terminal ? null : runnerTrace.projection ?? message.projection ?? null);
-      const agentRun = agentRunFromResult(result, runnerTrace) ?? agentRunFromMessage(message);
-      const terminalPatch = terminalMessagePatchFromTurnResult(message, result) ?? {};
-      return { ...message, ...messageTimingPatchForMerge(message, result), ...messageStatusPatchForTerminalMerge(message, resultStatus, terminal), runnerTrace, error, projection, projectionStatus: projection?.projectionStatus ?? null, projectionHealth: projection?.projectionHealth ?? null, blocker: projection?.blocker ?? null, agentRun: agentRun ?? undefined, ...terminalPatch, updatedAt: new Date().toISOString() };
-    }));
+    projectTurnAuthorityToMessages(traceId, result, "complete-trace");
    rememberTurnStatus(traceId, result);
    markWorkbenchTraceProjected(traceId);
    clearActiveTraceRestGapFill(traceId);