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

Merge pull request #2355 from pikasTech/fix/2353-terminal-p0-json-cost

Browse Source 
fix(workbench): prioritize terminal final DOM convergence
This commit is contained in:
Lyon
2026-07-02 22:50:01 +08:00
committed by GitHub
parent bc489f87c6 e15756d4a8
commit aadb61dd60
4 changed files with 104 additions and 28 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
+10
View File
@@ -138,10 +138,20 @@ test("workbench active terminal paths seal final response from turn authority",
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"));
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"));
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.match(terminalRestBlock, /traceProjectionIsTerminalSealed\(id, ownerMessagesBeforeProjection\)[\s\S]*return/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(loadBlock, /const messageLimit = sessionMessageProjectionWindowLimit\(\);/u);
assert.doesNotMatch(loadBlock, /limit:\s*100/u);
assert.doesNotMatch(restoreSealBlock, /Promise\.all/u);
assert.doesNotMatch(restoreSealBlock, /force:\s*true/u);
});
web/hwlab-cloud-web/src/stores/workbench-server-state.test.ts
+16
View File
@@ -78,6 +78,22 @@ test("server state reducer seals completed message when final response exists",
assert.equal(selectSessionStatusAuthority(state)[sessionId]?.status, "completed");
});
test("server state reducer upserts terminal authority when the session page has no agent row", () => {
const sessionId = "ses_state_terminal_upsert_missing_agent";
const traceId = "trc_state_terminal_upsert_missing_agent";
let state = createWorkbenchServerState();
state = reduceWorkbenchServerState(state, { type: "session.detail", session: sessionRecord({ sessionId, status: "running", messages: [] }) });
state = reduceWorkbenchServerState(state, { type: "session.messages", sessionId, messages: [userMessage({ id: "msg_state_terminal_upsert_user", text: "run it", sessionId, traceId })] });
state = reduceWorkbenchServerState(state, { type: "message.upsert", sessionId, message: agentMessage({ id: "msg_state_terminal_upsert_agent", status: "completed", text: "final answer from turn authority", finalResponse: { text: "final answer from turn authority" }, sessionId, traceId }) });
const messages = selectActiveMessages(state, sessionId);
assert.equal(messages.length, 2);
const terminal = messages.find((message) => message.role === "agent" && message.traceId === traceId);
assert.ok(terminal);
assert.equal(terminal.status, "completed");
assert.equal(terminal.text, "final answer from turn authority");
});
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";
web/hwlab-cloud-web/src/stores/workbench-server-state.ts
+20
View File
@@ -18,6 +18,7 @@ export type WorkbenchServerAction =
| { type: "session.detail"; session: WorkbenchSessionRecord | null | undefined }
| { type: "session.messages"; sessionId: string | null | undefined; messages: ChatMessage[] }
| { type: "message.snapshot"; sessionId: string | null | undefined; message: ChatMessage | null | undefined }
| { type: "message.upsert"; sessionId: string | null | undefined; message: ChatMessage | null | undefined }
| { type: "session.status"; session: SessionStatusAuthority }
| { type: "session.forget"; sessionId: string }
| { type: "session.cleanupDropped"; keepSessionIds: string[] }
@@ -46,6 +47,8 @@ export function reduceWorkbenchServerState(state: WorkbenchServerState, action:
return reduceSessionMessages(state, action.sessionId, action.messages);
case "message.snapshot":
return reduceMessageSnapshot(state, action.sessionId, action.message);
case "message.upsert":
return reduceMessageUpsert(state, action.sessionId, action.message);
case "session.status":
return { ...state, sessionStatusById: { ...state.sessionStatusById, [action.session.sessionId]: mergeSessionStatusAuthority(state.sessionStatusById[action.session.sessionId], action.session) } };
case "session.forget":
@@ -188,6 +191,23 @@ function reduceMessageSnapshot(state: WorkbenchServerState, sessionId: string |
return reduceSessionMessages(state, id, merged);
}
function reduceMessageUpsert(state: WorkbenchServerState, sessionId: string | null | undefined, message: ChatMessage | null | undefined): WorkbenchServerState {
const id = sessionId || message?.sessionId;
if (!id || !message) return state;
const existingSession = state.sessionsById[id];
const existing = state.messagesBySessionId[id] ?? existingSession?.messages ?? [];
const index = existing.findIndex((item) => messageMatchesSnapshot(item, message));
const mergedMessages = index >= 0
? existing.map((item, itemIndex) => itemIndex === index ? mergeMessageSnapshot(item, message) : item)
: [...existing, message];
const session = existingSession ? { ...existingSession, messages: mergedMessages, messageCount: mergedMessages.length } : null;
return {
...state,
sessionsById: session ? { ...state.sessionsById, [id]: session } : state.sessionsById,
messagesBySessionId: { ...state.messagesBySessionId, [id]: mergedMessages }
};
}
function messageMatchesSnapshot(existing: ChatMessage, incoming: ChatMessage): boolean {
const existingId = existing.messageId ?? existing.id;
const incomingId = incoming.messageId ?? incoming.id;
web/hwlab-cloud-web/src/stores/workbench.ts
+58 -28
View File
@@ -5,6 +5,7 @@ import { computed, nextTick, ref } from "vue";
import { defineStore } from "pinia";
import { api } from "@/api";
import { workbenchRuntimePolicy } from "@/config/workbench-runtime-policy";
import { createKeyedSingleflight } from "@/utils/scheduler/keyed-singleflight";
import { createWorkbenchHealthProbeCache } from "@/utils/workbench-health";
import { agentErrorFromProjection, normalizeApiErrorRecord, normalizeErrorDiagnostic, normalizeProjectionDiagnostic, projectionDiagnosticFromApiFailure, projectionDiagnosticFromFailure } from "@/utils/workbench-error-runtime";
import { readWorkbenchJson, readWorkbenchNumber, readWorkbenchString, removeWorkbenchStorageKey, writeWorkbenchJson, writeWorkbenchString } from "@/utils/workbench-storage-runtime";
@@ -88,6 +89,8 @@ export const useWorkbenchStore = defineStore("workbench", () => {
const workbenchColadaReducer = useWorkbenchColadaReducer();
const workbenchColadaQueries = useWorkbenchColadaQueries();
const workbenchColadaMutations = useWorkbenchColadaMutations();
const turnStatusReadSingleflight = createKeyedSingleflight<ApiResult<AgentChatResultResponse>>();
const traceHydrationSingleflight = createKeyedSingleflight<void>();
const providerProfile = ref<ProviderProfile>(readString("hwlab.workbench.providerProfile.v1", "codex"));
const providerOptions = ref<ProviderProfileOption[]>(defaultProviderProfileOptions(providerProfile.value));
const recentDrafts = ref<DraftEntry[]>(readRecentDrafts());
@@ -445,6 +448,14 @@ export const useWorkbenchStore = defineStore("workbench", () => {
return [...source].reverse().find((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === id) ?? null;
}
function traceTerminalBodyIsVisible(traceId: string | null | undefined, sessionId: string | null | undefined = null): boolean {
const id = firstNonEmptyString(traceId);
if (!id) return false;
const ownerSessionId = normalizeWorkbenchSessionId(sessionId) ?? traceOwnerSessionId(id, null);
const source = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value;
return traceProjectionIsTerminalSealed(id, source);
}
function appendActiveMessages(...items: ChatMessage[]): void {
replaceActiveMessages([...messages.value, ...items]);
}
@@ -471,7 +482,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
function fetchWorkbenchTurnStatus(traceId: string, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId), options: { force?: boolean } = {}): Promise<ApiResult<AgentChatResultResponse>> {
const activitySource = useActivityTimeout ? () => activityRef.value : null;
return workbenchColadaQueries.fetchTurn(traceId, { timeoutMs: 8000, activityRef: activitySource, minIntervalMs: runtimePolicy.workbenchTurnStatusMinRefreshMs, force: options.force });
return turnStatusReadSingleflight.run(traceId, () => workbenchColadaQueries.fetchTurn(traceId, { timeoutMs: 8000, activityRef: activitySource, minIntervalMs: runtimePolicy.workbenchTurnStatusMinRefreshMs, force: options.force }), { reason: options.force ? "force-turn-status" : "turn-status" });
}
function fetchWorkbenchTraceEvents(traceId: string, afterProjectedSeq: number, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId), options: { force?: boolean } = {}): Promise<ApiResult<AgentChatResultResponse>> {
@@ -576,6 +587,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 (traceTerminalBodyIsVisible(id)) {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_turn_skip", source: "turn-status", valuesRedacted: true } });
return;
}
const response = await fetchWorkbenchTurnStatus(id, shouldUseActivityTimeoutForTrace(id), { force: options.force });
if (response.ok && response.data) {
applyTurnStatusSnapshot(id, response.data);
@@ -624,23 +639,13 @@ export const useWorkbenchStore = defineStore("workbench", () => {
const authoritySessionId = traceResultSessionId(result);
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];
});
const source = serverState.value.messagesBySessionId[ownerSessionId] ?? serverState.value.sessionsById[ownerSessionId]?.messages ?? [];
const existing = [...source].reverse().find((message) => messageMatchesTraceAuthority(message, traceId, authoritySessionId, ownerSessionId)) ?? null;
const next = existing ? projectTurnAuthorityMessage(existing, result) : terminalAuthorityMessageFromTurnResult(result);
if (!next) return false;
const message = next.sessionId ? next : { ...next, sessionId: ownerSessionId };
reduceServerState({ type: "message.upsert", sessionId: ownerSessionId, message });
const sealed = !messageHasSealedTerminalResult(existing) && messageHasSealedTerminalResult(message);
if (sealed) {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-authority", sessionId: ownerSessionId, traceId, outcome: "ok", diagnostic: { code: "terminal_direct_seal", reason, source: "turn-authority", valuesRedacted: true } });
}
@@ -793,10 +798,24 @@ export const useWorkbenchStore = defineStore("workbench", () => {
}
async function hydrateTraceEventsForMessageNow(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
const traceId = message.traceId ?? message.runnerTrace?.traceId;
if (!traceId) return;
if (traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId)) {
recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: message.sessionId ?? null, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_trace_skip", source: "trace-hydration", valuesRedacted: true } });
return;
}
await traceHydrationSingleflight.run(traceId, async () => {
if (traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId)) return;
await hydrateTraceEventsForMessagePages(message, options);
}, { reason: options.force ? "force-trace-hydration" : "trace-hydration" });
}
async function hydrateTraceEventsForMessagePages(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
const traceId = message.traceId ?? message.runnerTrace?.traceId;
if (!traceId) return;
let afterProjectedSeq = traceHydrationProjectedSeq(message.runnerTrace);
for (let page = 0; page < runtimePolicy.traceHydrationMaxPages; page += 1) {
if (traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId)) return;
const result = await fetchTraceHydrationPage(traceId, afterProjectedSeq, { force: options.force });
if (!result.ok || !result.data) {
if (shouldSuppressTransientWorkbenchReadFailure(result)) return;
@@ -1037,6 +1056,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
async function refreshActiveTraceFromRest(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 } });
return;
}
const ownerBefore = traceOwnerSessionId(id, null);
if (ownerBefore === activeSessionId.value) recordActivity(reason);
await refreshTurnStatusByTraceId(id);
@@ -1214,6 +1238,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
async function refreshTerminalTraceFromRest(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 } });
return;
}
const ownerBefore = traceOwnerSessionId(id, null);
if (ownerBefore === activeSessionId.value) recordActivity(reason);
await refreshTurnStatusByTraceId(id);
@@ -1305,7 +1334,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
void refreshRealtimeSessionFromRest(ownerSessionId, `realtime-trace-snapshot:${traceId}`);
}
markWorkbenchTraceProjected(traceId);
scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListRealtimeRefreshDelayMs);
if (!traceProjectionIsTerminalSealed(traceId, serverState.value.messagesBySessionId[ownerSessionId] ?? [])) scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListRealtimeRefreshDelayMs);
}
async function refreshRealtimeSessionFromRest(sessionId: string, reason: string): Promise<void> {
@@ -1336,7 +1365,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
void clearActiveTrace(traceId, "trace-terminal");
restartRealtime("trace-terminal");
}
scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs);
if (!traceProjectionIsTerminalSealed(traceId, serverState.value.messagesBySessionId[ownerSessionId] ?? [])) scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs);
}
async function hydrateTerminalMessageDiagnostics(): Promise<void> {
@@ -1649,13 +1678,14 @@ export const useWorkbenchStore = defineStore("workbench", () => {
const requestId = normalizeWorkbenchSessionRouteId(sessionId);
if (!requestId) return null;
const normalizedRequestId = normalizeWorkbenchSessionId(requestId);
const eagerMessages = normalizedRequestId ? workbenchColadaQueries.fetchSessionMessages(normalizedRequestId, { limit: 100, force: true, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs }) : null;
const messageLimit = sessionMessageProjectionWindowLimit();
const eagerMessages = normalizedRequestId ? workbenchColadaQueries.fetchSessionMessages(normalizedRequestId, { limit: messageLimit, force: true, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs }) : null;
const detail = await workbenchColadaQueries.fetchSession(requestId, { force: true, minIntervalMs: runtimePolicy.workbenchSessionDetailMinRefreshMs });
if (!detail.ok) return null;
const detailSession = sessionFromWorkbenchSession(detail.data?.session);
const id = detailSession?.sessionId ?? normalizedRequestId ?? seed?.sessionId;
if (!id) return null;
const messages = eagerMessages && id === normalizedRequestId ? await eagerMessages : await workbenchColadaQueries.fetchSessionMessages(id, { limit: 100, force: true, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs });
const messages = eagerMessages && id === normalizedRequestId ? await eagerMessages : await workbenchColadaQueries.fetchSessionMessages(id, { limit: messageLimit, force: true, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs });
const base = detailSession ?? seed;
if (!base) return null;
const page = messages.ok ? messages.data : null;
@@ -1664,17 +1694,17 @@ export const useWorkbenchStore = defineStore("workbench", () => {
}
async function sealRestoredActiveTurnMessages(source: ChatMessage[]): Promise<ChatMessage[]> {
const targets = source.filter(messageNeedsRestoredTurnSeal).slice(-3);
const targets = source.filter(messageNeedsRestoredTurnSeal).slice(-1);
if (targets.length === 0) return source;
const patches = new Map<string, Partial<ChatMessage>>();
await Promise.all(targets.map(async (message) => {
for (const message of targets) {
const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);
if (!traceId) return;
const response = await fetchWorkbenchTurnStatus(traceId, false, { force: true });
if (!response.ok || !response.data) return;
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);
}));
}
if (patches.size === 0) return source;
return source.map((message) => {
const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);