diff --git a/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts b/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts index 464d9798..c14eb8d0 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts @@ -141,6 +141,7 @@ test("workbench active terminal paths seal final response from turn authority", 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 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 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")); @@ -160,9 +161,15 @@ test("workbench active terminal paths seal final response from turn authority", 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(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); assert.match(sessionDetailReadBlock, /workbenchSessionDetailReadKey\(\{ sessionId, force: options\.force \}\)/u); assert.match(sessionDetailReadBlock, /fetchSession\(sessionId, \{ includeMessages: false,/u); assert.match(loadBlock, /const messageLimit = sessionMessageProjectionWindowLimit\(\);/u); diff --git a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts index 8203b7ef..7a21db54 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts @@ -2,7 +2,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; import type { ChatMessage } from "@/types"; -import { projectionMergeCommitSummary, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget"; +import { projectionMergeCommitSummary, sessionDetailAutoReadDecision, traceEventsAutoReadDecision, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey, workbenchTraceEventsReadKey } from "./workbench-session-messages-read-budget"; import { selectActiveTurnStatusRefreshTraceIds } from "./workbench-session"; function message(id: string): ChatMessage { @@ -27,6 +27,21 @@ test("session detail singleflight key separates force freshness class", () => { assert.equal(fresh, workbenchSessionDetailReadKey({ sessionId: "ses_key" })); }); +test("trace events auto read skips initial and completed equivalent ranges", () => { + const key = workbenchTraceEventsReadKey({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50 }); + + assert.equal(traceEventsAutoReadDecision({ traceId: "trc_range", afterProjectedSeq: 0, limit: 50 }).read, false); + assert.equal(traceEventsAutoReadDecision({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50, cachedRange: { traceId: "trc_range", afterProjectedSeq: 12, limit: 50, nextProjectedSeq: 12, hasMore: false } }).read, false); + assert.equal(traceEventsAutoReadDecision({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50, cachedRange: { traceId: "trc_range", afterProjectedSeq: 12, limit: 50, nextProjectedSeq: 40, hasMore: true } }).read, true); + assert.equal(key, workbenchTraceEventsReadKey({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50 })); +}); + +test("session detail auto read defers trace scoped realtime refresh to turn authority", () => { + assert.deepEqual(sessionDetailAutoReadDecision({ traceId: "trc_turn", nowMs: 10_000, minIntervalMs: 5_000 }), { read: false, reason: "turn-authority-first" }); + assert.deepEqual(sessionDetailAutoReadDecision({ nowMs: 10_000, lastReadAtMs: 9_000, minIntervalMs: 5_000 }), { read: false, reason: "min-interval" }); + assert.deepEqual(sessionDetailAutoReadDecision({ nowMs: 10_000, lastReadAtMs: 1_000, minIntervalMs: 5_000 }), { read: true, reason: "metadata-only" }); +}); + test("projection merge summary reports no write without blocking follow-up hydrate", () => { const existing = [{ ...message("msg_same"), traceId: "trc_current", sessionId: "ses_same" }]; const summary = projectionMergeCommitSummary(existing, existing); diff --git a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts index 80cf261c..5e57ed5e 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts @@ -1,5 +1,5 @@ // SPEC: pikasTech/HWLAB#2356 Workbench bounded request storm follow-up. -// Responsibility: Pure budgeting helpers for session messages read de-duplication and projection merge writes. +// Responsibility: Pure budgeting helpers for Workbench read hydration de-duplication and projection merge writes. import type { ChatMessage } from "@/types"; import { firstNonEmptyString } from "@/utils"; @@ -21,6 +21,37 @@ export interface ProjectionMergeCommitSummary { changedCount: number; } +export interface TraceEventsReadRangeInput { + traceId: string | null | undefined; + afterProjectedSeq: number | null | undefined; + limit: number | null | undefined; +} + +export interface TraceEventsReadRangeRecord extends TraceEventsReadRangeInput { + nextProjectedSeq: number | null; + hasMore: boolean; +} + +export interface TraceEventsAutoReadDecisionInput extends TraceEventsReadRangeInput { + force?: boolean | null; + terminalBodyVisible?: boolean | null; + cachedRange?: TraceEventsReadRangeRecord | null; +} + +export interface SessionDetailAutoReadDecisionInput { + force?: boolean | null; + traceId?: string | null; + terminalBodyVisible?: boolean | null; + lastReadAtMs?: number | null; + nowMs: number; + minIntervalMs: number; +} + +export interface ReadHydrationDecision { + read: boolean; + reason: string; +} + export function workbenchSessionDetailReadKey(input: SessionDetailReadKeyInput): string { return composeWorkbenchScopedKey( "workbench.session-detail.read", @@ -39,6 +70,35 @@ export function workbenchSessionMessagesReadKey(input: SessionMessagesReadKeyInp ); } +export function workbenchTraceEventsReadKey(input: TraceEventsReadRangeInput): string { + return composeWorkbenchScopedKey( + "workbench.trace-events.read", + firstNonEmptyString(input.traceId) ?? "unknown-trace", + `after:${readSeqPart(input.afterProjectedSeq)}`, + `limit:${sessionMessagesReadLimitPart(input.limit)}` + ); +} + +export function traceEventsAutoReadDecision(input: TraceEventsAutoReadDecisionInput): ReadHydrationDecision { + if (input.force === true) return { read: true, reason: "force" }; + if (input.terminalBodyVisible === true) return { read: false, reason: "terminal-visible" }; + const afterProjectedSeq = readSeqPart(input.afterProjectedSeq); + if (afterProjectedSeq <= 0) return { read: false, reason: "initial-range-deferred" }; + const cached = input.cachedRange; + if (cached && readSeqPart(cached.afterProjectedSeq) === afterProjectedSeq && sessionMessagesReadLimitPart(cached.limit) === sessionMessagesReadLimitPart(input.limit) && cached.hasMore !== true) return { read: false, reason: "equivalent-range-complete" }; + return { read: true, reason: "delta-range" }; +} + +export function sessionDetailAutoReadDecision(input: SessionDetailAutoReadDecisionInput): ReadHydrationDecision { + if (input.force === true) return { read: true, reason: "force" }; + if (input.terminalBodyVisible === true) return { read: false, reason: "terminal-visible" }; + if (firstNonEmptyString(input.traceId)) return { read: false, reason: "turn-authority-first" }; + const lastReadAtMs = Number(input.lastReadAtMs); + const minIntervalMs = Math.max(0, Math.trunc(Number(input.minIntervalMs))); + if (Number.isFinite(lastReadAtMs) && input.nowMs - lastReadAtMs < minIntervalMs) return { read: false, reason: "min-interval" }; + return { read: true, reason: "metadata-only" }; +} + export function projectionMergeCommitSummary(existing: ChatMessage[], merged: ChatMessage[]): ProjectionMergeCommitSummary { let changedCount = Math.max(0, merged.length - existing.length); const sharedLength = Math.min(existing.length, merged.length); @@ -52,3 +112,8 @@ function sessionMessagesReadLimitPart(value: number | null | undefined): number const number = Number(value); return Number.isFinite(number) && number > 0 ? Math.trunc(number) : 1; } + +function readSeqPart(value: number | null | undefined): number { + const number = Number(value); + return Number.isFinite(number) && number > 0 ? Math.trunc(number) : 0; +} diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index da04c0b3..eb15f594 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -70,7 +70,7 @@ import { planWorkbenchRealtimeApply, planWorkbenchRealtimeRecovery, type Workben import { useWorkbenchColadaMutations } from "./workbench-colada-mutations"; import { useWorkbenchColadaQueries } from "./workbench-colada-queries"; import { useWorkbenchColadaReducer } from "./workbench-colada-reducer"; -import { projectionMergeCommitSummary, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget"; +import { projectionMergeCommitSummary, sessionDetailAutoReadDecision, 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"; @@ -113,6 +113,8 @@ export const useWorkbenchStore = defineStore("workbench", () => { const traceHydrationSingleflight = createKeyedSingleflight(); const sessionMessagesReadSingleflight = createKeyedSingleflight>(); const sessionDetailReadSingleflight = createKeyedSingleflight>(); + const traceEventsReadRanges = new Map(); + const sessionDetailAutoReadAtMs = new Map(); const realtimeTurnProjectionQueue = new Map(); let realtimeTurnProjectionScheduled = false; const providerProfile = ref(readString("hwlab.workbench.providerProfile.v1", "codex")); @@ -888,9 +890,14 @@ export const useWorkbenchStore = defineStore("workbench", () => { async function hydrateTraceEventsForMessagePages(message: ChatMessage, options: { force?: boolean } = {}): Promise { const traceId = message.traceId ?? message.runnerTrace?.traceId; if (!traceId) return; - let afterProjectedSeq = traceHydrationProjectedSeq(message.runnerTrace); + let afterProjectedSeq = traceHydrationProjectedSeq(traceAuthorityById.value[traceId] ?? message.runnerTrace); for (let page = 0; page < runtimePolicy.traceHydrationMaxPages; page += 1) { if (traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId)) return; + const decision = traceEventsHydrationReadDecision(traceId, afterProjectedSeq, message, options); + if (!decision.read) { + recordTraceEventsHydrationSkip(traceId, message, decision.reason, afterProjectedSeq); + return; + } const result = await fetchTraceHydrationPage(traceId, afterProjectedSeq, { force: options.force }); if (!result.ok || !result.data) { if (shouldSuppressTransientWorkbenchReadFailure(result)) return; @@ -900,11 +907,40 @@ export const useWorkbenchStore = defineStore("workbench", () => { } applyTraceHydrationResult(traceId, result.data); const nextProjectedSeq = traceNextProjectedSeq(result.data, afterProjectedSeq); + rememberTraceEventsHydrationRead(traceId, afterProjectedSeq, nextProjectedSeq, result.data); if (result.data.hasMore !== true || nextProjectedSeq <= afterProjectedSeq) return; afterProjectedSeq = nextProjectedSeq; } } + function traceEventsHydrationReadDecision(traceId: string, afterProjectedSeq: number, message: ChatMessage, options: { force?: boolean }): { read: boolean; reason: string } { + const limit = runtimePolicy.traceHydrationPageLimit; + const key = workbenchTraceEventsReadKey({ traceId, afterProjectedSeq, limit }); + return traceEventsAutoReadDecision({ + traceId, + afterProjectedSeq, + limit, + force: options.force, + terminalBodyVisible: traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId), + cachedRange: traceEventsReadRanges.get(key) ?? null + }); + } + + function rememberTraceEventsHydrationRead(traceId: string, afterProjectedSeq: number, nextProjectedSeq: number, result: AgentChatResultResponse): void { + const limit = runtimePolicy.traceHydrationPageLimit; + traceEventsReadRanges.set(workbenchTraceEventsReadKey({ traceId, afterProjectedSeq, limit }), { traceId, afterProjectedSeq, limit, nextProjectedSeq, hasMore: result.hasMore === true }); + } + + function recordTraceEventsHydrationSkip(traceId: string, message: ChatMessage, reason: string, afterProjectedSeq: number): void { + recordWorkbenchRuntimeDiagnostic({ + module: "workbench-trace-events-read", + sessionId: message.sessionId ?? message.runnerTrace?.sessionId ?? null, + traceId, + outcome: "ok", + diagnostic: { code: "trace_events_auto_read_skip", reason, source: "trace-hydration", afterProjectedSeq, limit: runtimePolicy.traceHydrationPageLimit, valuesRedacted: true } + }); + } + async function fetchTraceHydrationPage(traceId: string, afterProjectedSeq: number, options: { force?: boolean } = {}): Promise> { return fetchWorkbenchTraceEvents(traceId, afterProjectedSeq, shouldUseActivityTimeoutForTrace(traceId), { force: options.force }); } @@ -1495,7 +1531,19 @@ export const useWorkbenchStore = defineStore("workbench", () => { 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 } }); + 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 });