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

Merge pull request #2343 from pikasTech/fix/2338-passive-turn-authority

Browse Source 
fix: hydrate passive workbench turn authority
This commit is contained in:
Lyon
2026-07-02 12:18:47 +08:00
committed by GitHub
parent 0eec1877d5 9b7da7c9eb
commit a2d5cff29e
3 changed files with 80 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts
+37
View File
@@ -21,6 +21,7 @@ import { reduceWorkbenchRealtimeEvent } from "../src/stores/workbench-event-redu
import { planWorkbenchRealtimeApply, planWorkbenchRealtimeRecovery } from "../src/stores/workbench-realtime-plan.ts";
import { cleanupWorkbenchServerStateDroppedSessions, cleanupWorkbenchServerStateSessions, createWorkbenchServerState, reduceWorkbenchServerState } from "../src/stores/workbench-server-state.ts";
import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "../src/stores/workbench-session-cache.ts";
import { selectActiveTurnStatusRefreshTraceIds } from "../src/stores/workbench-session.ts";
test("Workbench scoped keys encode delimiter characters", () => {
assert.deepEqual(splitWorkbenchScopedKey(workbenchRealtimeScopeKey("ses|one", "trc/two")), ["workbench.realtime", "ses|one", "trc/two"]);
@@ -285,6 +286,42 @@ test("server-state cleanup removes dropped session trace and turn authority", ()
assert.deepEqual(Object.keys(cleanedById.traceById), []);
});
test("passive message projection selects active turn status refresh without local request", () => {
const traceIds = selectActiveTurnStatusRefreshTraceIds({
messages: [
agentMessage({ id: "m_old", traceId: "trc_old", status: "completed", text: "done" }),
agentMessage({ id: "m_new", traceId: "trc_new", status: "running", text: "thinking" })
],
currentRequestTraceId: null,
turnStatusAuthority: {}
});
assert.deepEqual(traceIds, ["trc_new"]);
});
test("passive message projection skips sealed terminal turn status refresh", () => {
const traceIds = selectActiveTurnStatusRefreshTraceIds({
messages: [agentMessage({ id: "m_done", traceId: "trc_done", status: "completed", text: "done" })],
turnStatusAuthority: { trc_done: { traceId: "trc_done", status: "completed", running: false, terminal: true, sessionId: "ses_1" } }
});
assert.deepEqual(traceIds, []);
});
test("turn status refresh keeps local request trace priority", () => {
const traceIds = selectActiveTurnStatusRefreshTraceIds({
messages: [
agentMessage({ id: "m_request", traceId: "trc_request", status: "pending", text: "" }),
agentMessage({ id: "m_observed", traceId: "trc_observed", status: "running", text: "thinking" })
],
currentRequestTraceId: "trc_request",
turnStatusAuthority: {},
limit: 2
});
assert.deepEqual(traceIds, ["trc_request", "trc_observed"]);
});
test("realtime event reducer classifies SSE payloads before store side effects", () => {
const trace = reduceWorkbenchRealtimeEvent({ type: "trace.event", traceId: "trc_1", event: { traceId: "trc_1", label: "delta" }, snapshot: { traceId: "trc_1", status: "running" } }, "workbench.trace.event");
assert.equal(trace.activityLabel, "realtime:trace.event");
web/hwlab-cloud-web/src/stores/workbench-session.ts
+39
View File
@@ -112,6 +112,21 @@ export function resolveCancelableAgentMessage(input: { messages: ChatMessage[];
return null;
}
export function selectActiveTurnStatusRefreshTraceIds(input: { messages: ChatMessage[]; currentRequestTraceId?: string | null; turnStatusAuthority?: TurnStatusAuthorityMap; limit?: number }): string[] {
const traceIds = new Set<string>();
const limit = Math.max(1, Math.trunc(input.limit ?? 3));
const currentRequestTraceId = firstNonEmptyString(input.currentRequestTraceId);
if (currentRequestTraceId && traceNeedsTurnStatusRefresh(currentRequestTraceId, input.messages, input.turnStatusAuthority)) traceIds.add(currentRequestTraceId);
for (const message of [...input.messages].reverse()) {
const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);
if (!traceId || traceIds.has(traceId)) continue;
if (!messageHasActiveTurnStatusEvidence(message, traceId, currentRequestTraceId)) continue;
if (traceNeedsTurnStatusRefresh(traceId, input.messages, input.turnStatusAuthority)) traceIds.add(traceId);
if (traceIds.size >= limit) break;
}
return [...traceIds];
}
export function shouldShowSessionListLoading(input: { loading: boolean; sessionsReady: boolean }): boolean {
return input.loading && !input.sessionsReady;
}
@@ -282,6 +297,30 @@ function latestAgentMessage(messages: ChatMessage[] | undefined): ChatMessage |
return [...(messages ?? [])].reverse().find((message) => message.role === "agent") ?? null;
}
function latestMessageForTrace(messages: ChatMessage[], traceId: string): ChatMessage | null {
return [...messages].reverse().find((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId) ?? null;
}
function messageHasActiveTurnStatusEvidence(message: ChatMessage, traceId: string, currentRequestTraceId: string | null): boolean {
if (currentRequestTraceId === traceId) return true;
return isActiveStatus(message.status) || message.traceAutoLifecycle === "running";
}
function traceNeedsTurnStatusRefresh(traceId: string | null | undefined, messages: ChatMessage[], turnStatusAuthority: TurnStatusAuthorityMap | undefined): boolean {
const id = firstNonEmptyString(traceId);
if (!id) return false;
const turn = turnStatusAuthority?.[id];
const message = latestMessageForTrace(messages, id);
if (turnStatusAuthorityIsSealed(turn, message)) return false;
if (messageIsTerminalForSession(message)) return false;
return true;
}
function turnStatusAuthorityIsSealed(turn: TurnStatusAuthority | undefined, message: ChatMessage | null): boolean {
if (!(turn?.terminal === true || isTerminalStatus(turn?.status))) return false;
return message ? messageHasTerminalResponse(message) : false;
}
function messageIsTerminalForSession(message: ChatMessage | null | undefined): boolean {
if (!message || message.role !== "agent") return false;
const status = normalizeSessionStatus(message.status);
web/hwlab-cloud-web/src/stores/workbench.ts
+4 -38
View File
@@ -15,7 +15,7 @@ import type { AgentChatResponse, AgentChatResultResponse, AgentRunProvenance, Ap
import { firstNonEmptyString, nextProtocolId, normalizeWorkbenchSessionId, normalizeWorkbenchSessionRouteId } from "@/utils";
import { composeWorkbenchScopedKey } from "@/utils/workbench-key";
import { failWorkbenchSessionSwitch, failWorkbenchSubmitJourney, finishWorkbenchSessionSwitchFullLoad, markWorkbenchSubmitApiAccepted, markWorkbenchTraceEventsReceived, markWorkbenchTraceProjected, recordWorkbenchLoadingState, recordWorkbenchRuntimeDiagnostic, startWorkbenchSessionSwitch, startWorkbenchSubmitJourney } from "@/utils/workbench-performance";
import { RECENT_DRAFTS_STORAGE_KEY, appendSessionPage, defaultProviderProfileOptions, isArchivedSession, mergeSessionIntoList, normalizeChatMessageStatus, normalizeRecentDrafts, normalizeWorkbenchMessageTitle, providerProfileOptionsFromPayload, recordRecentDraft, resolveCancelableAgentMessage, resolveComposerState, shouldShowSessionListLoading, sortSessionTabs, stableSessionList, type DraftEntry, type ProviderProfileOption, type TurnStatusAuthority } from "./workbench-session";
import { RECENT_DRAFTS_STORAGE_KEY, appendSessionPage, defaultProviderProfileOptions, isArchivedSession, mergeSessionIntoList, normalizeChatMessageStatus, normalizeRecentDrafts, normalizeWorkbenchMessageTitle, providerProfileOptionsFromPayload, recordRecentDraft, resolveCancelableAgentMessage, resolveComposerState, selectActiveTurnStatusRefreshTraceIds, shouldShowSessionListLoading, sortSessionTabs, stableSessionList, type DraftEntry, type ProviderProfileOption, type TurnStatusAuthority } from "./workbench-session";
import { initialWorkbenchSessionIdFromLocation } from "./workbench-projection";
import { cleanupWorkbenchServerStateSessions, createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, selectActiveSession, selectSessionList, selectSessionStatusAuthority, selectTraceAuthorityById, selectTurnStatusAuthority, type WorkbenchServerAction } from "./workbench-server-state";
import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "./workbench-session-cache";
@@ -538,6 +538,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
const pageMessages = Array.isArray(response.data.messages) ? response.data.messages.map((message) => normalizeChatMessage(message as ChatMessage)) : [];
const merged = mergeMessageProjectionPage(id, pageMessages);
rememberSessionMessages(id, merged);
await hydrateTurnStatusAuthority(merged);
hydrateTerminalTraceGaps(merged, "session-message-page");
}
@@ -567,6 +568,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
const pageMessages = Array.isArray(response.data.messages) ? response.data.messages.map((message) => normalizeChatMessage(message as ChatMessage)) : [];
const merged = mergeMessageProjectionPage(id, pageMessages);
rememberSessionMessages(id, merged);
await hydrateTurnStatusAuthority(merged);
hydrateTerminalTraceGaps(merged, `trace-message-page:${traceId}`);
}
@@ -602,43 +604,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
}
function activeTurnStatusRefreshTraceIds(source: ChatMessage[] = messages.value): string[] {
const traceIds = new Set<string>();
const requestTraceId = firstNonEmptyString(currentRequest.value?.traceId);
if (requestTraceId && traceNeedsTurnStatusRefresh(requestTraceId, source)) traceIds.add(requestTraceId);
for (const message of [...source].reverse()) {
const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);
if (!traceId || traceIds.has(traceId)) continue;
if (traceId !== requestTraceId) continue;
if (!messageHasActiveTurnStatusEvidence(message, traceId)) continue;
if (traceNeedsTurnStatusRefresh(traceId, source)) traceIds.add(traceId);
if (traceIds.size >= 3) break;
}
return [...traceIds];
}
function messageHasActiveTurnStatusEvidence(message: ChatMessage, traceId: string): boolean {
if (currentRequest.value?.traceId === traceId) return true;
return isTraceActiveStatus(message.status);
}
function traceNeedsTurnStatusRefresh(traceId: string | null | undefined, source: ChatMessage[] = messages.value): boolean {
const id = firstNonEmptyString(traceId);
if (!id) return false;
const turn = turnStatusAuthority.value[id];
const message = [...source].reverse().find((item) => firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === id) ?? null;
if (traceAuthorityIsSealed(turn, message)) return false;
if (messageIsSealedTerminal(message)) return false;
return true;
}
function traceAuthorityIsSealed(turn: TurnStatusAuthority | undefined, message: ChatMessage | null): boolean {
const status = normalizedStatusText(turn?.status);
if (!(turn?.terminal === true || isTerminalMessageStatus(status))) return false;
return message ? messageHasTerminalResponse(message) : false;
}
function messageIsSealedTerminal(message: ChatMessage | null | undefined): boolean {
return messageHasTerminalResponse(message);
return selectActiveTurnStatusRefreshTraceIds({ messages: source, currentRequestTraceId: currentRequest.value?.traceId, turnStatusAuthority: turnStatusAuthority.value, limit: 3 });
}
async function refreshTurnStatusByTraceId(traceId: string | null | undefined, options: { force?: boolean } = {}): Promise<void> {