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

Merge pull request #2232 from pikasTech/issue/2229-workbench-no-refresh

Browse Source 
fix(web): recover Workbench in-place refresh
This commit is contained in:
Lyon
2026-06-27 16:12:12 +08:00
committed by GitHub
parent 3fced0ba71 4b4a7c78ba
commit 6120f7fdfe
2 changed files with 122 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/hwlab-cloud-web/scripts/workbench-e2e-server.ts
+6 -1
View File
@@ -109,6 +109,11 @@ const sseClients = new Set<ServerResponse>();
const server = createServer((request, response) => {
void handleRequest(request, response).catch((error) => {
if (response.headersSent) {
console.error("workbench-e2e-server stream error", error);
response.destroy();
return;
}
json(response, 500, { ok: false, status: 500, error: { code: "e2e_server_error", message: error instanceof Error ? error.message : String(error) } });
});
});
@@ -2372,7 +2377,7 @@ function authPayload(): JsonRecord {
return { authenticated: false, mode: "server", authMethod: null, identityAuthority: null, sessionKind: null, actor: null, user: null, capabilities: deniedCapabilities, expiresAt: null, valuesRedacted: true, error: { code: "auth_required", message: "Authentication is required", status: 401 } };
}
const actor = { id: "usr_e2e_admin", username: "e2e-admin", role: "admin", status: "active" };
return { authenticated: true, mode: "server", authMethod: "web-session", identityAuthority: "hwlab-local", sessionKind: "browser", actor, user: actor, capabilities: { web: "available", workbench: "available", codeAgent: "available", billing: "unlinked", apiKeys: "available", admin: "available" }, expiresAt: null, valuesRedacted: true };
return { authenticated: true, mode: "server", authMethod: "web-session", identityAuthority: "hwlab-local", sessionKind: "browser", actor, user: actor, capabilities: { web: "available", workbench: "available", codeAgent: "available", billing: "unlinked", apiKeys: "available", admin: "available" }, access: { nav: { profileId: "e2e-admin", allowedIds: ["*"], valuesRedacted: true } }, expiresAt: null, valuesRedacted: true };
}
function authLoginResponse(response: ServerResponse): void {
web/hwlab-cloud-web/src/stores/workbench.ts
+116 -21
View File
@@ -35,6 +35,9 @@ const WORKBENCH_TRACE_EVENTS_TIMEOUT_MS = 5_000;
const SESSION_LIST_REALTIME_REFRESH_DELAY_MS = 5_000;
const SESSION_LIST_TERMINAL_REFRESH_DELAY_MS = 1_500;
const SESSION_LIST_MIN_REFRESH_INTERVAL_MS = 15_000;
const WORKBENCH_REALTIME_ERROR_GAP_FILL_MIN_MS = 2_000;
const WORKBENCH_ACTIVE_TRACE_REST_GAP_FILL_INITIAL_MS = 2_500;
const WORKBENCH_ACTIVE_TRACE_REST_GAP_FILL_REPEAT_MS = 5_000;
interface HydrateOptions {
sessionId?: string | null;
@@ -82,6 +85,8 @@ export const useWorkbenchStore = defineStore("workbench", () => {
const realtimeSessionMessagesInFlight = new Set<string>();
const realtimeSessionMessagesLastRefreshAtById = new Map<string, number>();
const realtimeSessionRefreshInFlight = new Set<string>();
const realtimeErrorGapFillLastAtByKey = new Map<string, number>();
const activeTraceRestGapFillTimers = new Map<string, number>();
let realtimeStream: WorkbenchEventStream | null = null;
let realtimeKey = "";
const realtimeOutboxSeqByKey = new Map<string, number>();
@@ -420,6 +425,12 @@ export const useWorkbenchStore = defineStore("workbench", () => {
updateSessionMessages(ownerSessionId, (source) => source.map((message) => messageMatchesTraceAuthority(message, traceId, authoritySessionId, ownerSessionId) ? project(message) : message));
}
function latestMessageForTrace(traceId: string | null | undefined, source: ChatMessage[] = messages.value): ChatMessage | null {
const id = firstNonEmptyString(traceId);
if (!id) return null;
return [...source].reverse().find((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === id) ?? null;
}
function appendActiveMessages(...items: ChatMessage[]): void {
replaceActiveMessages([...messages.value, ...items]);
}
@@ -564,21 +575,21 @@ export const useWorkbenchStore = defineStore("workbench", () => {
return false;
}
function fetchWorkbenchTurnStatus(traceId: string, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId)): Promise<ApiResult<AgentChatResultResponse>> {
function fetchWorkbenchTurnStatus(traceId: string, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId), options: { force?: boolean } = {}): Promise<ApiResult<AgentChatResultResponse>> {
const activitySource = useActivityTimeout ? () => activityRef.value : null;
return runWorkbenchReadHydration(
() => api.workbench.turn(traceId, 8000, activitySource),
workbenchReadCooldownKey("turn", traceId),
{ minIntervalMs: WORKBENCH_TURN_STATUS_MIN_REFRESH_MS },
{ minIntervalMs: WORKBENCH_TURN_STATUS_MIN_REFRESH_MS, force: options.force },
);
}
function fetchWorkbenchTraceEvents(traceId: string, afterProjectedSeq: number, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId)): Promise<ApiResult<AgentChatResultResponse>> {
function fetchWorkbenchTraceEvents(traceId: string, afterProjectedSeq: number, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId), options: { force?: boolean } = {}): Promise<ApiResult<AgentChatResultResponse>> {
const activitySource = useActivityTimeout ? () => activityRef.value : null;
return runWorkbenchReadHydration(
() => api.workbench.traceEvents(traceId, WORKBENCH_TRACE_EVENTS_TIMEOUT_MS, activitySource, { afterProjectedSeq, limit: TRACE_HYDRATION_PAGE_LIMIT }),
workbenchReadCooldownKey("trace-events", traceId),
{ minIntervalMs: WORKBENCH_TRACE_EVENTS_MIN_REFRESH_MS },
{ minIntervalMs: WORKBENCH_TRACE_EVENTS_MIN_REFRESH_MS, force: options.force },
);
}
@@ -621,13 +632,13 @@ export const useWorkbenchStore = defineStore("workbench", () => {
}
}
async function refreshMessageProjectionForTrace(sessionId: string | null | undefined, traceId: string): Promise<void> {
async function refreshMessageProjectionForTrace(sessionId: string | null | undefined, traceId: string, options: { force?: boolean } = {}): Promise<void> {
const id = normalizeWorkbenchSessionId(sessionId);
if (!id) return;
const response = await runWorkbenchReadHydration(
() => api.workbench.sessionMessages(id, { limit: 100 }),
workbenchReadCooldownKey("session-messages", id),
{ minIntervalMs: WORKBENCH_SESSION_MESSAGES_MIN_REFRESH_MS },
{ minIntervalMs: WORKBENCH_SESSION_MESSAGES_MIN_REFRESH_MS, force: options.force },
);
if (!response.ok || !response.data) {
if (shouldSuppressTransientWorkbenchReadFailure(response)) return;
@@ -700,13 +711,13 @@ export const useWorkbenchStore = defineStore("workbench", () => {
return true;
}
async function refreshTurnStatusByTraceId(traceId: string | null | undefined): Promise<void> {
async function refreshTurnStatusByTraceId(traceId: string | null | undefined, options: { force?: boolean } = {}): Promise<void> {
const id = firstNonEmptyString(traceId);
if (!id) return;
const response = await fetchWorkbenchTurnStatus(id);
const response = await fetchWorkbenchTurnStatus(id, shouldUseActivityTimeoutForTrace(id), { force: options.force });
if (response.ok && response.data) {
applyTurnStatusSnapshot(id, response.data);
if (response.data.terminal === true || isTerminalMessageStatus(response.data.status)) completeTrace(id, response.data);
if (response.data.terminal === true || isTerminalMessageStatus(response.data.status)) completeTrace(id, response.data, { forceRead: options.force });
return;
}
if (shouldSuppressTransientWorkbenchReadFailure(response)) return;
@@ -862,6 +873,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
completeTrace(canonicalTraceId, response.data as AgentChatResultResponse);
return true;
}
scheduleActiveTraceRestGapFill(canonicalTraceId, "submit-rest-gap");
restartRealtime("submit");
return true;
}
@@ -878,6 +890,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
recordActivity("cancel:accepted");
if (message.status === "running") chatPending.value = false;
currentRequest.value = null;
clearActiveTraceRestGapFill(traceId);
void clearActiveTrace(traceId, "cancel-agent-message");
scheduleSessionListRefresh(sessionId, SESSION_LIST_TERMINAL_REFRESH_DELAY_MS);
}
@@ -900,7 +913,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
await submitMessage(retryInput);
}
async function hydrateTraceEventsForMessage(message: ChatMessage): Promise<void> {
async function hydrateTraceEventsForMessage(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
const traceId = message.traceId ?? message.runnerTrace?.traceId;
if (!traceId) return;
if (traceHydrationInFlight.has(traceId)) return;
@@ -908,7 +921,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
try {
let afterProjectedSeq = traceHydrationProjectedSeq(message.runnerTrace);
for (let page = 0; page < TRACE_HYDRATION_MAX_PAGES; page += 1) {
const result = await fetchTraceHydrationPage(traceId, afterProjectedSeq);
const result = await fetchTraceHydrationPage(traceId, afterProjectedSeq, { force: options.force });
if (!result.ok || !result.data) {
if (shouldSuppressTransientWorkbenchReadFailure(result)) return;
if (messageHasCompletedFinalResponse(message)) return;
@@ -925,10 +938,10 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
}
}
async function fetchTraceHydrationPage(traceId: string, afterProjectedSeq: number): Promise<ApiResult<AgentChatResultResponse>> {
async function fetchTraceHydrationPage(traceId: string, afterProjectedSeq: number, options: { force?: boolean } = {}): Promise<ApiResult<AgentChatResultResponse>> {
let lastResult: ApiResult<AgentChatResultResponse> | null = null;
for (let attempt = 0; attempt < TRACE_HYDRATION_MAX_ATTEMPTS; attempt += 1) {
const result = await fetchWorkbenchTraceEvents(traceId, afterProjectedSeq);
const result = await fetchWorkbenchTraceEvents(traceId, afterProjectedSeq, shouldUseActivityTimeoutForTrace(traceId), { force: options.force });
if (result.ok && result.data) return result;
lastResult = result;
if (attempt < TRACE_HYDRATION_MAX_ATTEMPTS - 1) await delayTraceHydrationRetry(TRACE_HYDRATION_RETRY_DELAY_MS * (attempt + 1));
@@ -1103,6 +1116,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
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");
restartRealtime("reattach");
}
@@ -1134,11 +1148,70 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
traceId,
afterSeq,
onOpen: () => undefined,
onError: () => undefined,
onError: () => handleRealtimeStreamError(sessionId, traceId),
onEvent: (event, eventName) => applyRealtimeEvent(event, eventName)
});
}
function handleRealtimeStreamError(sessionId: string | null, traceId: string | null): void {
const activeId = normalizeWorkbenchSessionId(sessionId ?? selectedSessionId.value);
const activeTraceId = firstNonEmptyString(traceId, realtimeTraceId());
const key = [activeId ?? "", activeTraceId ?? ""].join("|");
if (!key.trim()) return;
const lastAt = realtimeErrorGapFillLastAtByKey.get(key) ?? 0;
if (lastAt > 0 && Date.now() - lastAt < WORKBENCH_REALTIME_ERROR_GAP_FILL_MIN_MS) return;
realtimeErrorGapFillLastAtByKey.set(key, Date.now());
if (activeId === activeSessionId.value) void refreshRealtimeSessionMessages(activeId, "realtime-error:messages", { force: true });
if (activeId) void refreshSessions(activeId, { force: true });
if (!activeTraceId || !shouldApplyActiveTraceAuthority(activeTraceId, activeId)) return;
void refreshTurnStatusByTraceId(activeTraceId, { force: true });
const message = latestMessageForTrace(activeTraceId);
if (message) void hydrateTraceEventsForMessage(message, { force: true });
}
function scheduleActiveTraceRestGapFill(traceId: string | null | undefined, reason: string, delayMs = WORKBENCH_ACTIVE_TRACE_REST_GAP_FILL_INITIAL_MS): void {
const id = firstNonEmptyString(traceId);
if (!id || typeof window === "undefined") return;
clearActiveTraceRestGapFill(id);
const timer = window.setTimeout(() => {
activeTraceRestGapFillTimers.delete(id);
void refreshActiveTraceFromRest(id, reason);
}, Math.max(0, Math.trunc(delayMs)));
activeTraceRestGapFillTimers.set(id, timer);
}
function clearActiveTraceRestGapFill(traceId: string | null | undefined): void {
const id = firstNonEmptyString(traceId);
if (!id || typeof window === "undefined") return;
const timer = activeTraceRestGapFillTimers.get(id);
if (timer) window.clearTimeout(timer);
activeTraceRestGapFillTimers.delete(id);
}
async function refreshActiveTraceFromRest(traceId: string, reason: string): Promise<void> {
const id = firstNonEmptyString(traceId);
if (!id) return;
const ownerBefore = traceOwnerSessionId(id, null);
if (ownerBefore === activeSessionId.value) recordActivity(reason);
await refreshTurnStatusByTraceId(id, { force: true });
const ownerSessionId = traceOwnerSessionId(id, turnStatusAuthority.value[id]?.sessionId ?? null) ?? ownerBefore;
if (ownerSessionId) {
await refreshSessions(ownerSessionId, { force: true });
await refreshMessageProjectionForTrace(ownerSessionId, id, { force: true });
}
const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value;
const message = latestMessageForTrace(id, ownerMessages);
if (message) await hydrateTraceEventsForMessage(message, { force: true });
const turn = turnStatusAuthority.value[id];
const terminal = turn?.terminal === true || isTerminalMessageStatus(turn?.status) || isTerminalMessageStatus(message?.status);
if (terminal) {
clearActiveTraceRestGapFill(id);
return;
}
const stillActive = currentRequest.value?.traceId === id || isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status);
if (stillActive) scheduleActiveTraceRestGapFill(id, "active-rest-gap:repeat", WORKBENCH_ACTIVE_TRACE_REST_GAP_FILL_REPEAT_MS);
}
function stopRealtime(): void {
realtimeStream?.close();
realtimeStream = null;
@@ -1300,10 +1373,26 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
}
async function refreshTerminalTraceFromRest(traceId: string, reason: string): Promise<void> {
if (terminalRealtimeRefreshInFlight.has(traceId)) return;
terminalRealtimeRefreshInFlight.add(traceId);
terminalRealtimeRefreshInFlight.delete(traceId);
void reason;
const id = firstNonEmptyString(traceId);
if (!id || terminalRealtimeRefreshInFlight.has(id)) return;
terminalRealtimeRefreshInFlight.add(id);
try {
const ownerBefore = traceOwnerSessionId(id, null);
if (ownerBefore === activeSessionId.value) recordActivity(reason);
await refreshTurnStatusByTraceId(id, { force: true });
const ownerSessionId = traceOwnerSessionId(id, turnStatusAuthority.value[id]?.sessionId ?? null) ?? ownerBefore;
if (ownerSessionId) await refreshSessions(ownerSessionId, { force: true });
if (ownerSessionId) await refreshMessageProjectionForTrace(ownerSessionId, id, { force: true });
const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value;
const message = latestMessageForTrace(id, ownerMessages);
if (message) {
await hydrateTraceEventsForMessage(message, { force: true });
} else if (ownerSessionId) {
await refreshRealtimeSessionMessages(ownerSessionId, `${reason}:message-gap`, { force: true });
}
} finally {
terminalRealtimeRefreshInFlight.delete(id);
}
}
function installRealtimeVisibilityHandler(): void {
@@ -1393,7 +1482,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
}
}
function completeTrace(traceId: string, result: AgentChatResultResponse): void {
function completeTrace(traceId: string, result: AgentChatResultResponse, options: { forceRead?: boolean } = {}): void {
const authoritySessionId = traceResultSessionId(result);
const ownerSessionId = traceOwnerSessionId(traceId, authoritySessionId);
if (!ownerSessionId) return;
@@ -1415,8 +1504,13 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
}));
rememberTurnStatus(traceId, result);
markWorkbenchTraceProjected(traceId);
void refreshMessageProjectionForTrace(ownerSessionId, traceId);
void hydrateTraceEvents(serverState.value.messagesBySessionId[ownerSessionId] ?? []);
clearActiveTraceRestGapFill(traceId);
void refreshMessageProjectionForTrace(ownerSessionId, traceId, { force: options.forceRead });
const ownerMessages = serverState.value.messagesBySessionId[ownerSessionId] ?? [];
const terminalMessage = latestMessageForTrace(traceId, ownerMessages);
if (options.forceRead && terminalMessage) void hydrateTraceEventsForMessage(terminalMessage, { force: true });
else void hydrateTraceEvents(ownerMessages);
if (options.forceRead) void refreshSessions(ownerSessionId, { force: true });
if (ownerSessionId === activeSessionId.value) {
chatPending.value = false;
currentRequest.value = null;
@@ -1596,6 +1690,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
}
async function clearActiveTrace(traceId: string, reason: string): Promise<void> {
clearActiveTraceRestGapFill(traceId);
if (currentRequest.value?.traceId === traceId) currentRequest.value = null;
}