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

fix(workbench): route automatic recovery through sync replay

Browse Source
This commit is contained in:
root
2026-07-08 20:27:59 +02:00
parent 3134747ddb
commit 91b60711e4
2 changed files with 123 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts
+36 -18
View File
@@ -136,37 +136,38 @@ test("workbench active terminal paths seal final response from turn authority",
const source = fs.readFileSync(path.join(storeDir, "workbench.ts"), "utf8");
const projectBlock = source.slice(source.indexOf("function projectTurnAuthorityToMessages"), source.indexOf("async function submitMessage"));
const realtimeTurnBlock = source.slice(source.indexOf("function applyRealtimeTurnSnapshot"), source.indexOf("function scheduleRealtimeTurnProjection"));
const realtimeTurnProjectionBlock = source.slice(source.indexOf("function flushRealtimeTurnProjection"), source.indexOf("async function refreshTerminalTraceFromRest"));
const terminalRestBlock = source.slice(source.indexOf("async function refreshTerminalTraceFromRest"), source.indexOf("function installRealtimeVisibilityHandler"));
const realtimeTurnProjectionBlock = source.slice(source.indexOf("function flushRealtimeTurnProjection"), source.indexOf("async function refreshTerminalTraceFromSyncReplay"));
const terminalSyncBlock = source.slice(source.indexOf("async function refreshTerminalTraceFromSyncReplay"), source.indexOf("function installRealtimeVisibilityHandler"));
const completeBlock = source.slice(source.indexOf("function completeTrace"), source.indexOf("async function hydrateTerminalMessageDiagnostics"));
const realtimeSessionDetailBlock = source.slice(source.indexOf("async function refreshRealtimeSessionFromRest"), source.indexOf("function completeTrace"));
const realtimeSessionSyncBlock = source.slice(source.indexOf("async function refreshRealtimeSessionFromSyncReplay"), source.indexOf("function completeTrace"));
const sessionDetailReadBlock = source.slice(source.indexOf("function fetchSessionDetailPage"), source.indexOf("function sessionMessageProjectionWindowLimit"));
const traceHydrationBlock = source.slice(source.indexOf("async function hydrateTraceEventsForMessagePages"), source.indexOf("async function fetchTraceHydrationPage"));
const traceHydrationBlock = source.slice(source.indexOf("async function hydrateTraceEventsForExplicitDetailPages"), source.indexOf("async function fetchTraceHydrationPage"));
const loadBlock = source.slice(source.indexOf("async function loadWorkbenchSession"), source.indexOf("async function sealRestoredActiveTurnMessages"));
const restoreSealBlock = source.slice(source.indexOf("async function sealRestoredActiveTurnMessages"), source.indexOf("function reattachRestoredActiveTrace"));
const activeSyncReplayBlock = source.slice(source.indexOf("async function refreshActiveTraceFromSyncReplay"), source.indexOf("function stopRealtime"));
assert.match(projectBlock, /terminalAuthorityMessageFromTurnResult\(result\)/u);
assert.match(projectBlock, /type:\s*"message\.upsert"/u);
assert.match(projectBlock, /\.\.\.terminalPatch/u);
assert.match(source, /createKeyedSingleflight<ApiResult<AgentChatResultResponse>>/u);
assert.match(source, /traceHydrationSingleflight\.run/u);
assert.match(realtimeTurnBlock, /!terminalTurn[\s\S]*refreshRealtimeSessionMessages/u);
assert.doesNotMatch(source, new RegExp(["refresh", "RealtimeSessionMessages"].join(""), "u"));
assert.match(realtimeTurnBlock, /rememberTurnStatus\(traceId, result\)[\s\S]*scheduleRealtimeTurnProjection\(\{ traceId, result, terminalTurn \}\)/u);
assert.doesNotMatch(realtimeTurnBlock, /applyTurnStatusSnapshot\(|refreshTerminalTraceFromRest\(/u);
assert.doesNotMatch(realtimeTurnBlock, new RegExp(`applyTurnStatusSnapshot\\(|${["refresh", "TerminalTraceFromRest"].join("")}\\(`, "u"));
assert.match(realtimeTurnProjectionBlock, /syncTurnStatusToMessage\(next\.traceId, next\.result\)/u);
assert.match(realtimeTurnProjectionBlock, /next\.terminalTurn[\s\S]*refreshTerminalTraceFromRest\(next\.traceId, "realtime-turn-snapshot"\)/u);
assert.match(realtimeTurnProjectionBlock, /next\.terminalTurn[\s\S]*refreshTerminalTraceFromSyncReplay\(next\.traceId, "realtime-turn-snapshot"\)/u);
assert.match(realtimeTurnProjectionBlock, /workbench_realtime_turn_projection_budget/u);
assert.match(terminalRestBlock, /traceProjectionIsTerminalSealed\(id, ownerMessagesBeforeProjection\)[\s\S]*return/u);
assert.match(terminalSyncBlock, /refreshWorkbenchSyncReplay\(ownerSessionId, id, null, `terminal-sync-replay:\$\{reason\}`\)/u);
assert.doesNotMatch(terminalSyncBlock, /refreshTurnStatusByTraceId|refreshMessageProjectionForTrace|fetchSessionMessagesPage|hydrateTraceEventsForMessage/u);
assert.match(completeBlock, /projectTurnAuthorityToMessages\(traceId, result, "complete-trace"\)/u);
assert.match(completeBlock, /traceProjectionIsTerminalSealed\(traceId, serverState\.value\.messagesBySessionId\[ownerSessionId\] \?\? \[\]\)[\s\S]*scheduleSessionListRefresh/u);
assert.match(realtimeSessionDetailBlock, /traceIdFromRealtimeRefreshReason\(reason\)/u);
assert.match(realtimeSessionDetailBlock, /traceTerminalBodyIsVisible\(traceId, id\)[\s\S]*terminal_low_priority_session_detail_skip[\s\S]*return/u);
assert.match(realtimeSessionDetailBlock, /hydrateTurnStatusAuthority\(activeMessagesBeforeDetail, \{ traceId, limit: 1, reason: "realtime-session-detail:turn-authority" \}\)/u);
assert.match(realtimeSessionDetailBlock, /sessionDetailAutoReadDecision\(\{ traceId,[\s\S]*workbenchSessionDetailMinRefreshMs \}\)/u);
assert.match(realtimeSessionDetailBlock, /session_detail_auto_read_skip/u);
assert.match(realtimeSessionDetailBlock, /traceTerminalBodyIsVisible\(traceId, id\)[\s\S]*terminal_low_priority_session_detail_apply_skip[\s\S]*return/u);
assert.match(realtimeSessionDetailBlock, /fetchSessionDetailPage\(id, \{ reason: `realtime-session-detail:\$\{reason\}` \}\)/u);
assert.doesNotMatch(realtimeSessionDetailBlock, /loadWorkbenchSession|applySelectedSessionDetail/u);
assert.match(completeBlock, /options\.forceRead[\s\S]*refreshMessageProjectionForTrace\(ownerSessionId, traceId, \{ force: true \}\)/u);
assert.match(completeBlock, /else scheduleActiveTraceSyncReplay\(traceId, "complete-trace-sync-replay", 0\)/u);
assert.match(realtimeSessionSyncBlock, /traceIdFromRealtimeRefreshReason\(reason\)/u);
assert.match(realtimeSessionSyncBlock, /refreshWorkbenchSyncReplay\(id, traceId, null, `realtime-session-sync:\$\{reason\}`\)/u);
assert.doesNotMatch(realtimeSessionSyncBlock, /hydrateTurnStatusAuthority|sessionDetailAutoReadDecision|fetchSessionDetailPage|fetchSessionMessagesPage|refreshMessageProjectionForTrace/u);
assert.doesNotMatch(realtimeSessionSyncBlock, /loadWorkbenchSession|applySelectedSessionDetail/u);
assert.match(activeSyncReplayBlock, /refreshWorkbenchSyncReplay\(ownerSessionId, id, null, `active-sync-replay:\$\{reason\}`\)/u);
assert.doesNotMatch(activeSyncReplayBlock, /refreshTurnStatusByTraceId|refreshMessageProjectionForTrace|fetchSessionMessagesPage|hydrateTraceEventsForMessage/u);
assert.match(traceHydrationBlock, /traceAuthorityById\.value\[traceId\] \?\? message\.runnerTrace/u);
assert.match(traceHydrationBlock, /traceEventsHydrationReadDecision\(traceId, afterProjectedSeq, message, options\)/u);
assert.match(traceHydrationBlock, /trace_events_auto_read_skip/u);
@@ -177,6 +178,23 @@ test("workbench active terminal paths seal final response from turn authority",
assert.match(loadBlock, /sessionFromWorkbenchSession\(detail\.data\?\.session, \{ includeMessages: false \}\)/u);
assert.match(loadBlock, /const fallbackMessages = seed\?\.sessionId === id \? seed\.messages \?\? \[\] : \[\];/u);
assert.doesNotMatch(loadBlock, /limit:\s*100/u);
assert.doesNotMatch(restoreSealBlock, /Promise\.all/u);
assert.match(restoreSealBlock, /workbench_restored_turn_auto_read_disabled/u);
assert.doesNotMatch(restoreSealBlock, /fetchWorkbenchTurnStatus|Promise\.all/u);
assert.doesNotMatch(restoreSealBlock, /force:\s*true/u);
});
test("workbench automatic live repair does not call legacy REST fan-out", () => {
const source = fs.readFileSync(path.join(storeDir, "workbench.ts"), "utf8");
const automaticBlocks = [
source.slice(source.indexOf("async function submitMessage"), source.indexOf("async function cancelAgentMessage")),
source.slice(source.indexOf("function reattachTrace"), source.indexOf("function restartRealtime")),
source.slice(source.indexOf("async function refreshActiveTraceFromSyncReplay"), source.indexOf("function stopRealtime")),
source.slice(source.indexOf("function flushRealtimeTurnProjection"), source.indexOf("function installRealtimeVisibilityHandler")),
source.slice(source.indexOf("async function refreshRealtimeSessionFromSyncReplay"), source.indexOf("function completeTrace")),
source.slice(source.indexOf("async function sealRestoredActiveTurnMessages"), source.indexOf("function reattachRestoredActiveTrace"))
].join("\n");
assert.match(automaticBlocks, /refreshWorkbenchSyncReplay/u);
assert.doesNotMatch(automaticBlocks, /fetchWorkbenchTurnStatus|fetchWorkbenchTraceEvents|fetchSessionMessagesPage|fetchSessionDetailPage/u);
assert.doesNotMatch(automaticBlocks, /refreshTurnStatusByTraceId|refreshMessageProjectionForTrace|hydrateTraceEventsForMessage\(/u);
});
web/hwlab-cloud-web/src/stores/workbench.ts
+87 -147
View File
@@ -71,7 +71,7 @@ import { workbenchSyncReplayDiagnostic, workbenchSyncReplayEvents } from "./work
import { useWorkbenchColadaMutations } from "./workbench-colada-mutations";
import { useWorkbenchColadaQueries } from "./workbench-colada-queries";
import { useWorkbenchColadaReducer } from "./workbench-colada-reducer";
import { projectionMergeCommitSummary, sessionDetailAutoReadDecision, traceEventsAutoReadDecision, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey, workbenchTraceEventsReadKey, type TraceEventsReadRangeRecord } from "./workbench-session-messages-read-budget";
import { projectionMergeCommitSummary, traceEventsAutoReadDecision, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey, workbenchTraceEventsReadKey, type TraceEventsReadRangeRecord } from "./workbench-session-messages-read-budget";
import { terminalSealResultWithoutTraceEvents, traceHydrationProjectedSeq, traceNextProjectedSeq } from "./workbench-trace-hydration";
const WORKBENCH_SESSION_PROJECTION_SIGNAL_CHANNEL = "hwlab.workbench.sessionProjection.v1";
@@ -115,7 +115,6 @@ export const useWorkbenchStore = defineStore("workbench", () => {
const sessionMessagesReadSingleflight = createKeyedSingleflight<ApiResult<WorkbenchMessagePageResponse>>();
const sessionDetailReadSingleflight = createKeyedSingleflight<ApiResult<WorkbenchSessionDetailResponse>>();
const traceEventsReadRanges = new Map<string, TraceEventsReadRangeRecord>();
const sessionDetailAutoReadAtMs = new Map<string, number>();
const realtimeTurnProjectionQueue = new Map<string, RealtimeTurnProjectionItem>();
let realtimeTurnProjectionScheduled = false;
const providerProfile = ref<ProviderProfile>(readString("hwlab.workbench.providerProfile.v1", "codex"));
@@ -528,25 +527,6 @@ export const useWorkbenchStore = defineStore("workbench", () => {
return isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status);
}
async function refreshSessionMessageProjectionPage(sessionId: string | null | undefined, options: { force?: boolean } = {}): Promise<void> {
const id = normalizeWorkbenchSessionId(sessionId);
if (!id) return;
const response = await fetchSessionMessagesPage(id, { limit: sessionMessageProjectionWindowLimit(), reason: "session-message-page", force: options.force });
if (!response.ok || !response.data) return;
const pageMessages = Array.isArray(response.data.messages) ? response.data.messages.map((message) => normalizeChatMessage(message as ChatMessage)) : [];
const merged = mergeMessageProjectionPage(id, pageMessages, { limit: sessionMessageProjectionWindowLimit() });
commitMergedSessionMessages(id, pageMessages, merged, { limit: sessionMessageProjectionWindowLimit(), reason: "session-message-page" });
await hydrateTurnStatusAuthority(merged, { limit: 1, reason: "session-message-page" });
hydrateTerminalTraceGaps(merged, "session-message-page");
}
async function refreshRealtimeSessionMessages(sessionId: string | null | undefined, reason: string, options: { force?: boolean } = {}): Promise<void> {
const id = normalizeWorkbenchSessionId(sessionId);
if (!id || id !== activeSessionId.value) return;
recordActivity(reason);
await refreshSessionMessageProjectionPage(id, { force: options.force });
}
async function refreshMessageProjectionForTrace(sessionId: string | null | undefined, traceId: string, options: { force?: boolean } = {}): Promise<void> {
const id = normalizeWorkbenchSessionId(sessionId);
if (!id) return;
@@ -636,13 +616,25 @@ export const useWorkbenchStore = defineStore("workbench", () => {
return existing.find((item) => item.role === message.role && firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === traceId) ?? null;
}
async function hydrateTurnStatusAuthority(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string } = {}): Promise<void> {
for (const traceId of activeTurnStatusRefreshTraceIds(source, options)) {
async function hydrateTurnStatusAuthority(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string; force?: boolean } = {}): Promise<void> {
const traceIds = activeTurnStatusRefreshTraceIds(source, options);
if (options.force !== true) {
if (traceIds.length > 0) {
recordWorkbenchRuntimeDiagnostic({
module: "workbench-turn-status",
traceId: options.traceId ?? traceIds[0] ?? null,
outcome: "ok",
diagnostic: { code: "workbench_turn_status_auto_read_disabled", reason: options.reason ?? "auto-turn-status-hydrate", source: "sync-replay-authority", requestedCount: traceIds.length, valuesRedacted: true }
});
}
return;
}
for (const traceId of traceIds) {
await refreshTurnStatusByTraceId(traceId);
}
}
function activeTurnStatusRefreshTraceIds(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string } = {}): string[] {
function activeTurnStatusRefreshTraceIds(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string; force?: boolean } = {}): string[] {
const requestedTraceId = firstNonEmptyString(options.traceId, currentRequest.value?.traceId);
const traceIds = selectActiveTurnStatusRefreshTraceIds({ messages: source, currentRequestTraceId: requestedTraceId, turnStatusAuthority: turnStatusAuthority.value, limit: options.limit ?? 1 });
recordWorkbenchRuntimeDiagnostic({
@@ -664,6 +656,10 @@ export const useWorkbenchStore = defineStore("workbench", () => {
async function refreshTurnStatusByTraceId(traceId: string | null | undefined, options: { force?: boolean } = {}): Promise<void> {
const id = firstNonEmptyString(traceId);
if (!id) return;
if (options.force !== true) {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-turn-status", traceId: id, outcome: "ok", diagnostic: { code: "workbench_turn_status_auto_read_disabled", source: "sync-replay-authority", valuesRedacted: true } });
return;
}
if (traceTerminalBodyIsVisible(id)) {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_turn_skip", source: "turn-status", valuesRedacted: true } });
return;
@@ -829,7 +825,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
completeTrace(canonicalTraceId, response.data as AgentChatResultResponse);
return true;
}
scheduleActiveTraceRestGapFill(canonicalTraceId, "submit-rest-gap");
scheduleActiveTraceSyncReplay(canonicalTraceId, "submit-sync-replay");
restartRealtime("submit");
return true;
}
@@ -846,7 +842,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
recordActivity("cancel:accepted");
if (message.status === "running") chatPending.value = false;
currentRequest.value = null;
clearActiveTraceRestGapFill(traceId);
clearActiveTraceSyncReplay(traceId);
void clearActiveTrace(traceId, "cancel-agent-message");
scheduleSessionListRefresh(sessionId, runtimePolicy.sessionListTerminalRefreshDelayMs);
}
@@ -872,6 +868,10 @@ export const useWorkbenchStore = defineStore("workbench", () => {
async function hydrateTraceEventsForMessage(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
const traceId = message.traceId ?? message.runnerTrace?.traceId;
if (!traceId) return;
if (options.force !== true) {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-trace-events-read", sessionId: message.sessionId ?? message.runnerTrace?.sessionId ?? null, traceId, outcome: "ok", diagnostic: { code: "trace_events_auto_read_disabled", reason: "sync-replay-authority", source: "trace-hydration", valuesRedacted: true } });
return;
}
await hydrateTraceEventsForMessageNow(message, options);
}
@@ -884,11 +884,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
}
await traceHydrationSingleflight.run(traceId, async () => {
if (traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId)) return;
await hydrateTraceEventsForMessagePages(message, options);
await hydrateTraceEventsForExplicitDetailPages(message, options);
}, { reason: options.force ? "force-trace-hydration" : "trace-hydration" });
}
async function hydrateTraceEventsForMessagePages(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
async function hydrateTraceEventsForExplicitDetailPages(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
const traceId = message.traceId ?? message.runnerTrace?.traceId;
if (!traceId) return;
let afterProjectedSeq = traceHydrationProjectedSeq(traceAuthorityById.value[traceId] ?? message.runnerTrace);
@@ -953,8 +953,16 @@ export const useWorkbenchStore = defineStore("workbench", () => {
void reason;
}
async function hydrateTraceEvents(source: ChatMessage[] = messages.value): Promise<void> {
for (const message of traceHydrationCandidates(source)) void hydrateTraceEventsForMessage(message);
async function hydrateTraceEvents(source: ChatMessage[] = messages.value, options: { force?: boolean } = {}): Promise<void> {
const candidates = traceHydrationCandidates(source);
if (options.force !== true) {
if (candidates.length > 0) {
const traceId = firstNonEmptyString(candidates[0]?.traceId, candidates[0]?.runnerTrace?.traceId);
recordWorkbenchRuntimeDiagnostic({ module: "workbench-trace-events-read", sessionId: candidates[0]?.sessionId ?? candidates[0]?.runnerTrace?.sessionId ?? null, traceId: traceId ?? null, outcome: "ok", diagnostic: { code: "trace_events_auto_read_disabled", reason: "sync-replay-authority", source: "trace-hydration", candidateCount: candidates.length, valuesRedacted: true } });
}
return;
}
for (const message of candidates) void hydrateTraceEventsForMessage(message, { force: true });
}
function traceHydrationCandidates(source: ChatMessage[]): ChatMessage[] {
@@ -1042,7 +1050,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
}));
markWorkbenchTraceProjected(traceId);
if (traceResultHasTerminalEvidence(result) && !turnResultIsTerminalForMerge(result)) {
void refreshTerminalTraceFromRest(traceId, "trace-hydration-terminal-evidence");
void refreshTerminalTraceFromSyncReplay(traceId, "trace-hydration-terminal-evidence");
}
if (turnResultIsTerminalForMerge(result)) {
rememberTurnStatus(traceId, result);
@@ -1092,24 +1100,10 @@ export const useWorkbenchStore = defineStore("workbench", () => {
const initial: AgentChatResponse = { status: "running", traceId, turnUrl: `/v1/workbench/turns/${encodeURIComponent(traceId)}` };
if (!messages.value.some((message) => message.traceId === traceId)) appendActiveMessages(makeMessage("agent", "", "running", { traceId, sessionId: selectedSessionId.value ?? undefined, threadId: selectedThreadId.value ?? undefined, title: "Code Agent", traceAutoLifecycle: "running" }));
currentRequest.value = { traceId, sessionId: selectedSessionId.value ?? null, threadId: selectedThreadId.value ?? null, status: initial.status };
scheduleActiveTraceRestGapFill(traceId, "reattach-rest-gap");
scheduleActiveTraceSyncReplay(traceId, "reattach-sync-replay");
restartRealtime("reattach");
}
async function validateAndReattachTrace(traceId: string): Promise<void> {
const result = await fetchWorkbenchTurnStatus(traceId);
if (!result.ok || !result.data) {
await clearActiveTrace(traceId, result.status === 404 ? "reattach-result-not-found" : "reattach-result-unavailable");
return;
}
applyTurnStatusSnapshot(traceId, result.data);
if (result.data.running === true || isTraceActiveStatus(result.data.status)) {
reattachTrace(traceId);
return;
}
await clearActiveTrace(traceId, "reattach-terminal-result");
}
function restartRealtime(reason: string): void {
const traceId = realtimeTraceId();
const sessionId = selectedSessionId.value ?? null;
@@ -1160,58 +1154,45 @@ export const useWorkbenchStore = defineStore("workbench", () => {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-sync-replay", sessionId, traceId, outcome: "ok", diagnostic: workbenchSyncReplayDiagnostic(result.data, events, { reason, sinceOutboxSeq }) });
}
function scheduleActiveTraceRestGapFill(traceId: string | null | undefined, reason: string, delayMs = runtimePolicy.workbenchActiveTraceRestGapFillInitialMs): void {
function scheduleActiveTraceSyncReplay(traceId: string | null | undefined, reason: string, delayMs = runtimePolicy.workbenchActiveTraceRestGapFillInitialMs): void {
const id = firstNonEmptyString(traceId);
if (!id) return;
if (typeof window === "undefined") {
void refreshActiveTraceFromRest(id, reason);
void refreshActiveTraceFromSyncReplay(id, reason);
return;
}
window.setTimeout(() => { void refreshActiveTraceFromRest(id, reason); }, delayMs);
window.setTimeout(() => { void refreshActiveTraceFromSyncReplay(id, reason); }, delayMs);
}
function clearActiveTraceRestGapFill(traceId: string | null | undefined): void {
function clearActiveTraceSyncReplay(traceId: string | null | undefined): void {
void traceId;
}
async function refreshActiveTraceFromRest(traceId: string, reason: string): Promise<void> {
async function refreshActiveTraceFromSyncReplay(traceId: string, reason: string): Promise<void> {
const id = firstNonEmptyString(traceId);
if (!id) return;
if (traceTerminalBodyIsVisible(id)) {
clearActiveTraceRestGapFill(id);
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_rest_gap_skip", reason, source: "active-rest-gap", valuesRedacted: true } });
clearActiveTraceSyncReplay(id);
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_sync_replay_skip", reason, source: "active-sync-replay", valuesRedacted: true } });
return;
}
const ownerBefore = traceOwnerSessionId(id, null);
const ownerMessagesBeforeRest = ownerBefore ? serverState.value.messagesBySessionId[ownerBefore] ?? [] : messages.value;
if (!shouldHydrateTerminalTraceFromRest({ traceId: id, messages: ownerMessagesBeforeRest, turnStatusAuthority: turnStatusAuthority.value })) {
clearActiveTraceRestGapFill(id);
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: ownerBefore, traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_rest_gap_skip", reason, source: "active-rest-gap-preflight", valuesRedacted: true } });
return;
}
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.sessionListRealtimeRefreshDelayMs);
await refreshMessageProjectionForTrace(ownerSessionId, id);
}
const ownerSessionId = traceOwnerSessionId(id, currentRequest.value?.sessionId ?? null);
if (ownerSessionId === activeSessionId.value) recordActivity(reason);
await refreshWorkbenchSyncReplay(ownerSessionId, id, null, `active-sync-replay:${reason}`);
const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value;
if (traceProjectionIsTerminalSealed(id, ownerMessages)) {
clearActiveTraceSyncReplay(id);
return;
}
const message = latestMessageForTrace(id, ownerMessages);
const turn = turnStatusAuthority.value[id];
const terminal = messageHasTerminalResponse(message);
if (terminal) {
clearActiveTraceRestGapFill(id);
clearActiveTraceSyncReplay(id);
return;
}
if (message) await hydrateTraceEventsForMessage(message);
const stillActive = currentRequest.value?.traceId === id || isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status);
if (stillActive) scheduleActiveTraceRestGapFill(id, "active-rest-gap:repeat", runtimePolicy.workbenchActiveTraceRestGapFillRepeatMs);
if (stillActive) scheduleActiveTraceSyncReplay(id, "active-sync-replay:repeat", runtimePolicy.workbenchActiveTraceRestGapFillRepeatMs);
}
function stopRealtime(): void {
@@ -1417,7 +1398,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
const startedAt = performanceNowMs();
if (shouldApplyActiveTraceAuthority(next.traceId, traceResultSessionId(next.result))) {
syncTurnStatusToMessage(next.traceId, next.result);
if (next.terminalTurn) void refreshTerminalTraceFromRest(next.traceId, "realtime-turn-snapshot");
if (next.terminalTurn) void refreshTerminalTraceFromSyncReplay(next.traceId, "realtime-turn-snapshot");
}
recordWorkbenchRuntimeDiagnostic({
module: "workbench-turn-status",
@@ -1437,38 +1418,23 @@ export const useWorkbenchStore = defineStore("workbench", () => {
if (realtimeTurnProjectionQueue.size > 0) scheduleRealtimeTurnProjectionFlush();
}
async function refreshTerminalTraceFromRest(traceId: string, reason: string): Promise<void> {
async function refreshTerminalTraceFromSyncReplay(traceId: string, reason: string): Promise<void> {
const id = firstNonEmptyString(traceId);
if (!id) return;
if (traceTerminalBodyIsVisible(id)) {
clearActiveTraceRestGapFill(id);
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_terminal_rest_skip", reason, source: "terminal-rest", valuesRedacted: true } });
clearActiveTraceSyncReplay(id);
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_terminal_sync_skip", reason, source: "terminal-sync-replay", valuesRedacted: true } });
return;
}
const ownerBefore = traceOwnerSessionId(id, null);
const ownerMessagesBeforeRest = ownerBefore ? serverState.value.messagesBySessionId[ownerBefore] ?? [] : messages.value;
if (!shouldHydrateTerminalTraceFromRest({ traceId: id, messages: ownerMessagesBeforeRest, turnStatusAuthority: turnStatusAuthority.value })) {
clearActiveTraceRestGapFill(id);
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: ownerBefore, traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_terminal_rest_skip", reason, source: "terminal-rest-preflight", valuesRedacted: true } });
return;
}
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);
}
const ownerSessionId = traceOwnerSessionId(id, null);
if (ownerSessionId === activeSessionId.value) recordActivity(reason);
await refreshWorkbenchSyncReplay(ownerSessionId, id, null, `terminal-sync-replay:${reason}`);
const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value;
const message = latestMessageForTrace(id, ownerMessages);
if (!message && ownerSessionId) {
await refreshRealtimeSessionMessages(ownerSessionId, `${reason}:message-gap`);
if (traceProjectionIsTerminalSealed(id, ownerMessages)) {
clearActiveTraceSyncReplay(id);
return;
}
if (ownerSessionId) scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs);
}
function installRealtimeVisibilityHandler(): void {
@@ -1512,7 +1478,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
if (!sessionId || sessionId !== activeSessionId.value) return;
const traceId = firstNonEmptyString(record.traceId);
if (traceId && messages.value.some((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId)) return;
void refreshRealtimeSessionMessages(sessionId, `cross-tab-session-projection:${firstNonEmptyString(record.reason, traceId, "session") ?? "session"}`);
void refreshRealtimeSessionFromSyncReplay(sessionId, `cross-tab-session-projection:${firstNonEmptyString(record.reason, traceId, "session") ?? "session"}`);
}
function applyTraceSnapshot(traceId: string, snapshot: TraceSnapshot): void {
@@ -1538,48 +1504,22 @@ export const useWorkbenchStore = defineStore("workbench", () => {
return { ...message, ...messageTimingPatchForMerge(message, trace), ...messageStatusPatchForTerminalMerge(message, traceStatus, terminal), runnerTrace, error: clearCompletedDiagnostics ? null : error ?? message.error ?? null, projection, projectionStatus: projection?.projectionStatus ?? null, projectionHealth: projection?.projectionHealth ?? null, blocker: projection?.blocker ?? null, updatedAt: new Date().toISOString() };
}));
if (!matchedMessage && ownerSessionId === activeSessionId.value) {
void refreshRealtimeSessionMessages(ownerSessionId, `realtime-trace-gap:${traceId}`);
void refreshRealtimeSessionFromRest(ownerSessionId, `realtime-trace-snapshot:${traceId}`);
void refreshRealtimeSessionFromSyncReplay(ownerSessionId, `realtime-trace-snapshot:${traceId}`);
}
markWorkbenchTraceProjected(traceId);
if (!traceProjectionIsTerminalSealed(traceId, serverState.value.messagesBySessionId[ownerSessionId] ?? [])) scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListRealtimeRefreshDelayMs);
}
async function refreshRealtimeSessionFromRest(sessionId: string, reason: string): Promise<void> {
async function refreshRealtimeSessionFromSyncReplay(sessionId: string, reason: string): Promise<void> {
const id = normalizeWorkbenchSessionId(sessionId);
if (!id || id !== activeSessionId.value) return;
const traceId = traceIdFromRealtimeRefreshReason(reason);
if (traceTerminalBodyIsVisible(traceId, id)) {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_session_detail_skip", reason, source: "realtime-session-detail", valuesRedacted: true } });
return;
}
const activeMessagesBeforeDetail = serverState.value.messagesBySessionId[id] ?? messages.value;
await hydrateTurnStatusAuthority(activeMessagesBeforeDetail, { traceId, limit: 1, reason: "realtime-session-detail:turn-authority" });
if (traceTerminalBodyIsVisible(traceId, id)) {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_session_detail_turn_skip", reason, source: "realtime-session-detail", valuesRedacted: true } });
return;
}
const detailDecision = sessionDetailAutoReadDecision({ traceId, terminalBodyVisible: false, lastReadAtMs: sessionDetailAutoReadAtMs.get(id) ?? null, nowMs: Date.now(), minIntervalMs: runtimePolicy.workbenchSessionDetailMinRefreshMs });
if (!detailDecision.read) {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-session-detail-read", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "session_detail_auto_read_skip", reason: detailDecision.reason, source: "realtime-session-detail", valuesRedacted: true } });
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_session_sync_skip", reason, source: "realtime-session-sync", valuesRedacted: true } });
return;
}
recordActivity(reason);
sessionDetailAutoReadAtMs.set(id, Date.now());
const detail = await fetchSessionDetailPage(id, { reason: `realtime-session-detail:${reason}` });
if (!detail.ok || activeSessionId.value !== id) return;
const detailSession = sessionFromWorkbenchSession(detail.data?.session, { includeMessages: false });
if (detailSession) {
const existingMessages = serverState.value.messagesBySessionId[id] ?? [];
rememberSessionDetail({ ...detailSession, messages: existingMessages, messageCount: detailSession.messageCount ?? existingMessages.length });
if (traceTerminalBodyIsVisible(traceId, id)) {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_session_detail_apply_skip", reason, source: "realtime-session-detail", valuesRedacted: true } });
return;
}
}
const activeMessages = serverState.value.messagesBySessionId[id] ?? messages.value;
await hydrateTurnStatusAuthority(activeMessages, { traceId, limit: 1, reason: "realtime-session-detail" });
if (traceId && !traceProjectionIsTerminalSealed(traceId, activeMessages)) hydrateTerminalTraceGaps(selectProjectionMessageWindow(activeMessages, { traceId, limit: traceMessageProjectionWindowLimit() }), `realtime-session-detail:${reason}`);
await refreshWorkbenchSyncReplay(id, traceId, null, `realtime-session-sync:${reason}`);
}
function completeTrace(traceId: string, result: AgentChatResultResponse, options: { forceRead?: boolean } = {}): void {
@@ -1590,10 +1530,13 @@ export const useWorkbenchStore = defineStore("workbench", () => {
projectTurnAuthorityToMessages(traceId, result, "complete-trace");
rememberTurnStatus(traceId, result);
markWorkbenchTraceProjected(traceId);
clearActiveTraceRestGapFill(traceId);
clearActiveTraceSyncReplay(traceId);
const ownerMessages = serverState.value.messagesBySessionId[ownerSessionId] ?? [];
const terminalMessage = latestMessageForTrace(traceId, ownerMessages);
if (!traceProjectionIsTerminalSealed(traceId, ownerMessages)) void refreshMessageProjectionForTrace(ownerSessionId, traceId, { force: options.forceRead });
if (!traceProjectionIsTerminalSealed(traceId, ownerMessages)) {
if (options.forceRead) void refreshMessageProjectionForTrace(ownerSessionId, traceId, { force: true });
else scheduleActiveTraceSyncReplay(traceId, "complete-trace-sync-replay", 0);
}
if (options.forceRead && terminalMessage && !traceProjectionIsTerminalSealed(traceId, ownerMessages)) void hydrateTraceEventsForMessage(terminalMessage, { force: true });
if (ownerSessionId === activeSessionId.value) {
chatPending.value = false;
@@ -1853,7 +1796,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
}
async function clearActiveTrace(traceId: string, reason: string): Promise<void> {
clearActiveTraceRestGapFill(traceId);
clearActiveTraceSyncReplay(traceId);
if (currentRequest.value?.traceId === traceId) currentRequest.value = null;
}
@@ -1933,27 +1876,24 @@ export const useWorkbenchStore = defineStore("workbench", () => {
async function sealRestoredActiveTurnMessages(source: ChatMessage[]): Promise<ChatMessage[]> {
const targets = source.filter(messageNeedsRestoredTurnSeal).slice(-1);
if (targets.length === 0) return source;
const patches = new Map<string, Partial<ChatMessage>>();
for (const message of targets) {
const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);
if (!traceId || traceProjectionIsTerminalSealed(traceId, source)) continue;
const response = await fetchWorkbenchTurnStatus(traceId, false);
if (!response.ok || !response.data) continue;
const patch = terminalMessagePatchFromTurnResult(message, response.data);
if (patch) patches.set(traceId, patch);
recordWorkbenchRuntimeDiagnostic({
module: "workbench-turn-status",
sessionId: message.sessionId ?? message.runnerTrace?.sessionId ?? null,
traceId,
outcome: "ok",
diagnostic: { code: "workbench_restored_turn_auto_read_disabled", reason: "load-session:restored-turn", source: "sync-replay-authority", valuesRedacted: true }
});
}
if (patches.size === 0) return source;
return source.map((message) => {
const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);
const patch = traceId ? patches.get(traceId) : null;
return patch ? { ...message, ...patch } : message;
});
return source;
}
function reattachRestoredActiveTrace(): void {
void hydrateTraceEvents(messages.value);
const traceId = activeTraceIdFromMessages(messages.value, turnStatusAuthority.value);
if (traceId) void validateAndReattachTrace(traceId);
if (traceId) reattachTrace(traceId);
}
function performanceNowMs(): number {