Merge pull request #2362 from pikasTech/fix/2356-response-json-budget
fix(workbench): budget realtime session detail refresh
This commit is contained in:
Lyon
@@ -139,6 +139,7 @@ test("workbench active terminal paths seal final response from turn authority",
|
||||
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 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 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"));
|
||||
|
||||
@@ -154,8 +155,12 @@ test("workbench active terminal paths seal final response from turn authority",
|
||||
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, /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(sessionDetailReadBlock, /workbenchSessionDetailReadKey\(\{ sessionId, force: options\.force \}\)/u);
|
||||
assert.match(sessionDetailReadBlock, /fetchSession\(sessionId, \{ includeMessages: false,/u);
|
||||
assert.match(loadBlock, /const messageLimit = sessionMessageProjectionWindowLimit\(\);/u);
|
||||
assert.match(loadBlock, /fetchSession\(requestId, \{ includeMessages: false,/u);
|
||||
assert.match(loadBlock, /fetchSessionDetailPage\(requestId, \{ reason: "load-session:detail", force: true \}\)/u);
|
||||
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);
|
||||
|
||||
@@ -46,6 +46,19 @@ test("server state reducer keeps completed message without final response unseal
|
||||
assert.notEqual(selectSessionStatusAuthority(state)[sessionId]?.status, "completed");
|
||||
});
|
||||
|
||||
test("server state reducer keeps session detail reference stable when detail has no semantic change", () => {
|
||||
const sessionId = "ses_state_stable_detail";
|
||||
const traceId = "trc_state_stable_detail";
|
||||
let state = createWorkbenchServerState();
|
||||
const messages = [agentMessage({ id: "msg_state_stable_detail", status: "running", traceId, sessionId })];
|
||||
state = reduceWorkbenchServerState(state, { type: "session.detail", session: sessionRecord({ sessionId, status: "running", lastTraceId: traceId, updatedAt: "2026-07-01T00:00:00.000Z", messageCount: 1, messages }) });
|
||||
const first = state;
|
||||
state = reduceWorkbenchServerState(state, { type: "session.detail", session: sessionRecord({ sessionId, status: "running", lastTraceId: traceId, updatedAt: "2026-07-01T00:00:00.000Z", messageCount: 1, messages: [...messages] }) });
|
||||
|
||||
assert.equal(state, first);
|
||||
assert.equal(selectActiveMessages(state, sessionId), selectActiveMessages(first, sessionId));
|
||||
});
|
||||
|
||||
test("server state reducer keeps failed message without terminal body unsealed", () => {
|
||||
const sessionId = "ses_state_failed_without_body";
|
||||
const traceId = "trc_state_failed_without_body";
|
||||
|
||||
@@ -116,12 +116,14 @@ function reduceSessionDetail(state: WorkbenchServerState, session: WorkbenchSess
|
||||
const messages = Array.isArray(session.messages) ? mergeMessageList(existingMessages, session.messages) : existingMessages;
|
||||
const merged = mergeSessionRecord(existing, { ...session, messages });
|
||||
const sessionStatus = sessionStatusAuthorityFromDetail(session);
|
||||
const nextSessionStatus = sessionStatus ? mergeSessionStatusAuthority(state.sessionStatusById[sessionId], sessionStatus) : null;
|
||||
if (existing && state.sessionOrder.includes(sessionId) && sessionRecordsEquivalent(existing, merged) && messageListsEquivalent(existingMessages, messages) && sessionStatusAuthorityEquivalent(state.sessionStatusById[sessionId], nextSessionStatus)) return state;
|
||||
return {
|
||||
...state,
|
||||
sessionOrder: state.sessionOrder.includes(sessionId) ? state.sessionOrder : [sessionId, ...state.sessionOrder],
|
||||
sessionsById: { ...state.sessionsById, [sessionId]: merged },
|
||||
sessionStatusById: sessionStatus
|
||||
? { ...state.sessionStatusById, [sessionId]: mergeSessionStatusAuthority(state.sessionStatusById[sessionId], sessionStatus) }
|
||||
? { ...state.sessionStatusById, [sessionId]: nextSessionStatus ?? sessionStatus }
|
||||
: state.sessionStatusById,
|
||||
messagesBySessionId: {
|
||||
...state.messagesBySessionId,
|
||||
@@ -415,6 +417,58 @@ function mergeSessionRecord(existing: WorkbenchSessionRecord | undefined, incomi
|
||||
};
|
||||
}
|
||||
|
||||
function sessionRecordsEquivalent(left: WorkbenchSessionRecord | null | undefined, right: WorkbenchSessionRecord | null | undefined): boolean {
|
||||
if (left === right) return true;
|
||||
if (!left || !right) return false;
|
||||
const leftMessages = Array.isArray(left.messages) ? left.messages : [];
|
||||
const rightMessages = Array.isArray(right.messages) ? right.messages : [];
|
||||
const leftRecord = { ...left, messages: undefined };
|
||||
const rightRecord = { ...right, messages: undefined };
|
||||
return valuesEquivalent(leftRecord, rightRecord) && messageListsEquivalent(leftMessages, rightMessages);
|
||||
}
|
||||
|
||||
function sessionStatusAuthorityEquivalent(left: SessionStatusAuthority | null | undefined, right: SessionStatusAuthority | null | undefined): boolean {
|
||||
if (left === right) return true;
|
||||
if (!left || !right) return !left && !right;
|
||||
return left.sessionId === right.sessionId
|
||||
&& left.status === right.status
|
||||
&& (left.updatedAt ?? null) === (right.updatedAt ?? null)
|
||||
&& (left.lastTraceId ?? null) === (right.lastTraceId ?? null)
|
||||
&& valuesEquivalent(left.projection ?? null, right.projection ?? null);
|
||||
}
|
||||
|
||||
function messageListsEquivalent(left: ChatMessage[], right: ChatMessage[]): boolean {
|
||||
if (left === right) return true;
|
||||
if (left.length !== right.length) return false;
|
||||
for (let index = 0; index < left.length; index += 1) {
|
||||
if (!valuesEquivalent(left[index], right[index])) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
function valuesEquivalent(left: unknown, right: unknown): boolean {
|
||||
if (Object.is(left, right)) return true;
|
||||
if (!left || !right || typeof left !== "object" || typeof right !== "object") return false;
|
||||
if (Array.isArray(left) || Array.isArray(right)) {
|
||||
if (!Array.isArray(left) || !Array.isArray(right) || left.length !== right.length) return false;
|
||||
for (let index = 0; index < left.length; index += 1) {
|
||||
if (!valuesEquivalent(left[index], right[index])) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
const leftRecord = left as Record<string, unknown>;
|
||||
const rightRecord = right as Record<string, unknown>;
|
||||
const leftKeys = Object.keys(leftRecord).filter((key) => leftRecord[key] !== undefined).sort();
|
||||
const rightKeys = Object.keys(rightRecord).filter((key) => rightRecord[key] !== undefined).sort();
|
||||
if (leftKeys.length !== rightKeys.length) return false;
|
||||
for (let index = 0; index < leftKeys.length; index += 1) {
|
||||
const key = leftKeys[index];
|
||||
if (key !== rightKeys[index]) return false;
|
||||
if (!valuesEquivalent(leftRecord[key], rightRecord[key])) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
function projectSessionRecord(session: WorkbenchSessionRecord | null | undefined, authority: SessionStatusAuthority | null | undefined): WorkbenchSessionRecord | null {
|
||||
if (!session) return null;
|
||||
if (!authority?.status || normalizedMessageStatus(session.status) === "archived") return session;
|
||||
|
||||
@@ -2,7 +2,7 @@ import assert from "node:assert/strict";
|
||||
import { test } from "bun:test";
|
||||
|
||||
import type { ChatMessage } from "@/types";
|
||||
import { projectionMergeCommitSummary, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget";
|
||||
import { projectionMergeCommitSummary, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget";
|
||||
import { selectActiveTurnStatusRefreshTraceIds } from "./workbench-session";
|
||||
|
||||
function message(id: string): ChatMessage {
|
||||
@@ -19,6 +19,14 @@ test("session messages singleflight key separates limit and force class", () =>
|
||||
assert.equal(limit20, workbenchSessionMessagesReadKey({ sessionId: "ses_key", limit: 20, force: false }));
|
||||
});
|
||||
|
||||
test("session detail singleflight key separates force freshness class", () => {
|
||||
const fresh = workbenchSessionDetailReadKey({ sessionId: "ses_key", force: false });
|
||||
const forced = workbenchSessionDetailReadKey({ sessionId: "ses_key", force: true });
|
||||
|
||||
assert.notEqual(fresh, forced);
|
||||
assert.equal(fresh, workbenchSessionDetailReadKey({ sessionId: "ses_key" }));
|
||||
});
|
||||
|
||||
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);
|
||||
|
||||
@@ -11,11 +11,25 @@ export interface SessionMessagesReadKeyInput {
|
||||
force?: boolean | null;
|
||||
}
|
||||
|
||||
export interface SessionDetailReadKeyInput {
|
||||
sessionId: string | null | undefined;
|
||||
force?: boolean | null;
|
||||
}
|
||||
|
||||
export interface ProjectionMergeCommitSummary {
|
||||
changed: boolean;
|
||||
changedCount: number;
|
||||
}
|
||||
|
||||
export function workbenchSessionDetailReadKey(input: SessionDetailReadKeyInput): string {
|
||||
return composeWorkbenchScopedKey(
|
||||
"workbench.session-detail.read",
|
||||
firstNonEmptyString(input.sessionId) ?? "unknown-session",
|
||||
"messages:false",
|
||||
input.force === true ? "force" : "fresh"
|
||||
);
|
||||
}
|
||||
|
||||
export function workbenchSessionMessagesReadKey(input: SessionMessagesReadKeyInput): string {
|
||||
return composeWorkbenchScopedKey(
|
||||
"workbench.session-messages.read",
|
||||
|
||||
@@ -11,7 +11,7 @@ import { agentErrorFromProjection, normalizeApiErrorRecord, normalizeErrorDiagno
|
||||
import { readWorkbenchJson, readWorkbenchNumber, readWorkbenchString, removeWorkbenchStorageKey, writeWorkbenchJson, writeWorkbenchString } from "@/utils/workbench-storage-runtime";
|
||||
import { createWorkbenchStreamTransportRuntime, type WorkbenchRealtimeEvent, type WorkbenchStreamTransportRecovery } from "@/utils/workbench-realtime-runtime";
|
||||
import { mergeRunnerTrace, snapshotToRunnerTrace, type TraceSnapshot } from "@/composables/useTraceSubscription";
|
||||
import type { WorkbenchMessagePageResponse } from "@/api/workbench";
|
||||
import type { WorkbenchMessagePageResponse, WorkbenchSessionDetailResponse } from "@/api/workbench";
|
||||
import type { AgentChatResponse, AgentChatResultResponse, AgentRunProvenance, ApiError, ApiResult, ChatMessage, ErrorDiagnostic, LiveSurface, ProjectionBlocker, ProjectionDiagnostic, ProviderProfile, TraceEvent, WorkbenchSessionRecord, WorkbenchTurnTimingProjection } from "@/types";
|
||||
import { firstNonEmptyString, nextProtocolId, normalizeWorkbenchSessionId, normalizeWorkbenchSessionRouteId } from "@/utils";
|
||||
import { composeWorkbenchScopedKey } from "@/utils/workbench-key";
|
||||
@@ -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, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget";
|
||||
import { projectionMergeCommitSummary, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget";
|
||||
|
||||
const WORKBENCH_SESSION_PROJECTION_SIGNAL_CHANNEL = "hwlab.workbench.sessionProjection.v1";
|
||||
const WORKBENCH_SESSION_PROJECTION_SIGNAL_KEY = "hwlab.workbench.sessionProjectionSignal.v1";
|
||||
@@ -92,6 +92,11 @@ interface SessionMessagesReadOptions {
|
||||
force?: boolean;
|
||||
}
|
||||
|
||||
interface SessionDetailReadOptions {
|
||||
reason: string;
|
||||
force?: boolean;
|
||||
}
|
||||
|
||||
export const useWorkbenchStore = defineStore("workbench", () => {
|
||||
const runtimePolicy = workbenchRuntimePolicy();
|
||||
const workbenchColadaReducer = useWorkbenchColadaReducer();
|
||||
@@ -100,6 +105,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
|
||||
const turnStatusReadSingleflight = createKeyedSingleflight<ApiResult<AgentChatResultResponse>>();
|
||||
const traceHydrationSingleflight = createKeyedSingleflight<void>();
|
||||
const sessionMessagesReadSingleflight = createKeyedSingleflight<ApiResult<WorkbenchMessagePageResponse>>();
|
||||
const sessionDetailReadSingleflight = createKeyedSingleflight<ApiResult<WorkbenchSessionDetailResponse>>();
|
||||
const providerProfile = ref<ProviderProfile>(readString("hwlab.workbench.providerProfile.v1", "codex"));
|
||||
const providerOptions = ref<ProviderProfileOption[]>(defaultProviderProfileOptions(providerProfile.value));
|
||||
const recentDrafts = ref<DraftEntry[]>(readRecentDrafts());
|
||||
@@ -553,6 +559,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
|
||||
return sessionMessagesReadSingleflight.run(key, () => workbenchColadaQueries.fetchSessionMessages(sessionId, { limit: options.limit, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs, force: options.force }), { reason: options.reason });
|
||||
}
|
||||
|
||||
function fetchSessionDetailPage(sessionId: string, options: SessionDetailReadOptions): Promise<ApiResult<WorkbenchSessionDetailResponse>> {
|
||||
const key = workbenchSessionDetailReadKey({ sessionId, force: options.force });
|
||||
return sessionDetailReadSingleflight.run(key, () => workbenchColadaQueries.fetchSession(sessionId, { includeMessages: false, minIntervalMs: runtimePolicy.workbenchSessionDetailMinRefreshMs, force: options.force }), { reason: options.reason });
|
||||
}
|
||||
|
||||
function sessionMessageProjectionWindowLimit(): number {
|
||||
return boundedProjectionMessageLimit(runtimePolicy.workbenchSessionMessagesWindowLimit, runtimePolicy.sessionListPageLimit);
|
||||
}
|
||||
@@ -1413,15 +1424,20 @@ export const useWorkbenchStore = defineStore("workbench", () => {
|
||||
return;
|
||||
}
|
||||
recordActivity(reason);
|
||||
const existing = sessions.value.find((item) => item.sessionId === id) ?? null;
|
||||
const selected = await loadWorkbenchSession(id, existing);
|
||||
if (selected && activeSessionId.value === id) {
|
||||
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;
|
||||
}
|
||||
applySelectedSessionDetail(selected, "system");
|
||||
}
|
||||
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}`);
|
||||
}
|
||||
|
||||
function completeTrace(traceId: string, result: AgentChatResultResponse, options: { forceRead?: boolean } = {}): void {
|
||||
@@ -1758,7 +1774,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
|
||||
const normalizedRequestId = normalizeWorkbenchSessionId(requestId);
|
||||
const messageLimit = sessionMessageProjectionWindowLimit();
|
||||
const eagerMessages = normalizedRequestId ? fetchSessionMessagesPage(normalizedRequestId, { limit: messageLimit, reason: "load-session:eager", force: true }) : null;
|
||||
const detail = await workbenchColadaQueries.fetchSession(requestId, { includeMessages: false, force: true, minIntervalMs: runtimePolicy.workbenchSessionDetailMinRefreshMs });
|
||||
const detail = await fetchSessionDetailPage(requestId, { reason: "load-session:detail", force: true });
|
||||
if (!detail.ok) return null;
|
||||
const detailSession = sessionFromWorkbenchSession(detail.data?.session, { includeMessages: false });
|
||||
const id = detailSession?.sessionId ?? normalizedRequestId ?? seed?.sessionId;
|
||||
|
||||
Reference in New Issue
Block a user