From d0eb94ae6ad54819db2e2528a7fe0c1afd515aa0 Mon Sep 17 00:00:00 2001 From: lyon Date: Thu, 18 Jun 2026 23:17:58 +0800 Subject: [PATCH] fix: add Workbench projection read model --- internal/cloud/server-code-agent-http.ts | 128 ++++---------- internal/cloud/server-workbench-http.test.ts | 13 ++ internal/cloud/server-workbench-http.ts | 119 +++++-------- internal/cloud/workbench-facts-store.ts | 165 ++++++++++++++++++ .../cloud/workbench-projection-finalizer.ts | 81 +++++++++ internal/cloud/workbench-projection-writer.ts | 46 +++++ internal/cloud/workbench-read-model.ts | 62 +++++++ .../scripts/workbench-server-state.test.ts | 27 ++- .../src/stores/workbench-server-state.ts | 18 +- web/hwlab-cloud-web/src/stores/workbench.ts | 10 +- 10 files changed, 486 insertions(+), 183 deletions(-) create mode 100644 internal/cloud/workbench-facts-store.ts create mode 100644 internal/cloud/workbench-projection-finalizer.ts create mode 100644 internal/cloud/workbench-projection-writer.ts create mode 100644 internal/cloud/workbench-read-model.ts diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index 1a629278..909b6c43 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -1,5 +1,5 @@ -// SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. -// Responsibility: Code Agent HTTP turn admission, lifecycle compatibility wrappers, and session evidence projection. +// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. +// Responsibility: Code Agent HTTP turn admission, lifecycle compatibility wrappers, and Workbench projection writer/finalizer integration. import { createHash, randomUUID } from "node:crypto"; @@ -18,6 +18,8 @@ import { import { createCodeAgentErrorPayload, handleCodeAgentChat } from "./code-agent-chat.ts"; import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { codeAgentSessionLifecycleSummary } from "./code-agent-session-lifecycle.ts"; +import { scheduleWorkbenchProjectionFinalizer } from "./workbench-projection-finalizer.ts"; +import { writeWorkbenchProjectionSession } from "./workbench-projection-writer.ts"; import { firstHeaderValue, getHeader, @@ -794,76 +796,20 @@ function scheduleAgentRunProjectionSync({ traceId, params = {}, options = {}, tr const noResponseTimeoutMs = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_NO_RESPONSE_TIMEOUT_MS, null); const pollIntervalMs = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_SYNC_POLL_INTERVAL_MS, 1_000); const refreshOptions = codeAgentTurnStatusRefreshOptions(options); - let lastActivityAt = Date.now(); - let lastActivitySignature = agentRunProjectionActivitySignature(currentResult, traceStore.snapshot(traceId)); - let idleDegraded = false; - - setImmediate(() => { - void (async () => { - let result = currentResult; - let lastError = null; - while (true) { - const cached = options.codeAgentChatResults?.get?.(traceId); - if (isCodeAgentResultCanceled(cached)) return; - if (isTraceCommandTerminalStatus(cached?.status)) { - scheduleCodeAgentTerminalTurnStatusEffects({ payload: cached, params, options, preserveLastTraceId: true }); - return; - } - try { - const synced = await syncAgentRunChatResult({ traceId, currentResult: result, options: refreshOptions, traceStore }); - result = synced.result ?? result; - if (result && isTraceCommandTerminalStatus(result.status)) { - scheduleCodeAgentTerminalTurnStatusEffects({ payload: result, params, options, preserveLastTraceId: true }); - return; - } - const signature = agentRunProjectionActivitySignature(result, synced.runnerTrace ?? traceStore.snapshot(traceId)); - if (signature && signature !== lastActivitySignature) { - lastActivitySignature = signature; - lastActivityAt = Date.now(); - idleDegraded = false; - } - lastError = null; - } catch (error) { - lastError = error; - } - const idleMs = Date.now() - lastActivityAt; - if (noResponseTimeoutMs && idleMs >= noResponseTimeoutMs && !idleDegraded) { - traceStore.append(traceId, { - type: "turn-status", - status: "degraded", - label: "agentrun:projection:no-response-idle-timeout", - errorCode: lastError?.code ?? "no_response_idle_timeout", - degradedReason: lastError ? "projection_activity_stale" : "no_response_idle_timeout", - message: lastError?.message ?? `AgentRun projection has no new activity for ${idleMs}ms; idle threshold ${noResponseTimeoutMs}ms.`, - runId: result?.agentRun?.runId ?? currentResult.agentRun.runId, - commandId: result?.agentRun?.commandId ?? currentResult.agentRun.commandId, - timeoutMs: noResponseTimeoutMs, - idleMs, - lastActivityAt: new Date(lastActivityAt).toISOString(), - waitingFor: "agentrun-result-activity", - terminal: false, - valuesPrinted: false - }); - idleDegraded = true; - } - await sleepAgentRunProjectionSync(pollIntervalMs); - } - })().catch((error) => { - traceStore.append(traceId, { - type: "turn-status", - status: "degraded", - label: "agentrun:projection:sync-crashed", - errorCode: error?.code ?? "agentrun_projection_sync_crashed", - message: error?.message ?? "AgentRun projection sync crashed.", - runId: currentResult.agentRun.runId, - commandId: currentResult.agentRun.commandId, - waitingFor: "agentrun-result", - terminal: false, - valuesPrinted: false - }); - }); + return scheduleWorkbenchProjectionFinalizer({ + traceId, + currentResult, + traceStore, + noResponseTimeoutMs, + pollIntervalMs, + getCachedResult: () => options.codeAgentChatResults?.get?.(traceId), + isCanceled: isCodeAgentResultCanceled, + isTerminal: isTraceCommandTerminalStatus, + activitySignature: agentRunProjectionActivitySignature, + sleep: sleepAgentRunProjectionSync, + syncResult: ({ result }) => syncAgentRunChatResult({ traceId, currentResult: result, options: refreshOptions, traceStore }), + onTerminal: (payload, finalizerOptions = {}) => scheduleCodeAgentTerminalTurnStatusEffects({ payload, params, options, preserveLastTraceId: finalizerOptions.preserveLastTraceId === true }) }); - return { scheduled: true, traceId, noResponseTimeoutMs, pollIntervalMs, valuesPrinted: false }; } function agentRunProjectionActivitySignature(result, runnerTrace) { @@ -2096,32 +2042,20 @@ async function recordCodeAgentSessionOwner({ payload = {}, params = {}, options const sessionId = safeSessionId(payload.sessionId ?? session?.sessionId ?? sessionReuse?.sessionId ?? params.sessionId) || null; const conversationId = safeConversationId(payload.conversationId ?? session?.conversationId ?? sessionReuse?.conversationId ?? params.conversationId) || null; const threadId = safeOpaqueId(session?.threadId ?? sessionReuse?.threadId ?? payload.threadId ?? params.threadId) || null; - try { - return await options.accessController.recordAgentSessionOwner({ - ownerUserId, - ownerRole: options.actor?.role ?? params.ownerRole ?? null, - sessionId, - projectId: params.projectId ?? payload.projectId ?? null, - conversationId, - threadId, - traceId: preserveLastTraceId ? undefined : traceId, - status, - session: codeAgentSessionOwnerEvidence(payload, params) - }); - } catch (error) { - const traceStore = options.traceStore ?? defaultCodeAgentTraceStore; - if (traceId) { - traceStore.append(traceId, { - type: "session-owner", - status: "degraded", - label: "session-owner:persist-failed", - errorCode: "agent_session_owner_persist_failed", - message: error?.message ?? "Code Agent session owner binding persistence failed.", - valuesPrinted: false - }); - } - return null; - } + return writeWorkbenchProjectionSession({ + accessController: options.accessController, + traceStore: options.traceStore ?? defaultCodeAgentTraceStore, + traceId, + ownerUserId, + ownerRole: options.actor?.role ?? params.ownerRole ?? null, + sessionId, + projectId: params.projectId ?? payload.projectId ?? null, + conversationId, + threadId, + status, + preserveLastTraceId, + session: codeAgentSessionOwnerEvidence(payload, params) + }); } function codeAgentOwnerStatusForResult(result = {}) { diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index d34d7dfb..e0ebb998 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -1,3 +1,6 @@ +// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1. +// Responsibility: Workbench read model, durable projection diagnostics, and pure-GET regression tests. + import assert from "node:assert/strict"; import { test } from "bun:test"; @@ -438,6 +441,16 @@ test("workbench read model projects completed current turn for idle session summ assert.equal(turn.body.turn.running, false); assert.equal(turn.body.turn.terminal, true); assert.equal(turn.body.turn.assistantText, finalText); + assert.equal(turn.body.projectionStatus, "caught-up"); + assert.equal(turn.body.sourceRunId, "run_workbench_idle_completed"); + assert.equal(turn.body.sourceCommandId, "cmd_workbench_idle_completed"); + + const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`); + assert.equal(trace.status, 200); + assert.equal(trace.body.projectionStatus, "caught-up"); + assert.equal(trace.body.sourceRunId, "run_workbench_idle_completed"); + assert.equal(trace.body.sourceCommandId, "cmd_workbench_idle_completed"); + assert.equal(trace.body.lastProjectedSeq, 2); } finally { await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); } diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 094d049d..3f2e5185 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -1,5 +1,5 @@ /* - * SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 * 职责: Workbench REST read model。只读投影 session/message/turn/trace facts，不执行 AgentRun sync、billing finalize 或 workspace repair。 */ import { createHash } from "node:crypto"; @@ -13,6 +13,7 @@ import { safeTraceId, sendJson } from "./server-http-utils.ts"; +import { createWorkbenchReadModel } from "./workbench-read-model.ts"; const DEFAULT_PAGE_LIMIT = 50; const MAX_PAGE_LIMIT = 100; @@ -77,6 +78,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option const requestedTraceId = safeTraceId(url.searchParams.get("traceId")); const heartbeatMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_HEARTBEAT_MS, DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS); const traceStore = options.traceStore ?? defaultCodeAgentTraceStore; + const readModel = createWorkbenchReadModel(options, auth.actor); let closed = false; const cleanup = []; @@ -108,7 +110,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option } }); - const initialSession = requestedSessionId ? await visibleSessionById(options.accessController?.store, requestedSessionId, auth.actor) : null; + const initialSession = requestedSessionId ? await readModel.getSessionById(requestedSessionId) : null; const initialSessionSnapshot = objectValue(initialSession?.session); const activeTraceId = requestedTraceId ?? safeTraceId(initialSession?.lastTraceId ?? initialSessionSnapshot.lastTraceId ?? initialSessionSnapshot.currentTraceId) @@ -202,19 +204,8 @@ async function handleWorkbenchSessionList(response, url, options, actor) { const limit = boundedLimit(url.searchParams.get("limit")); const includeSessionId = safeSessionId(url.searchParams.get("includeSessionId")); const includeRouteId = includeSessionId ?? safeConversationId(url.searchParams.get("includeSessionId")); - const store = options.accessController?.store; - const listInput = { - ownerUserId: actor.id, - actorRole: actor.role, - ownerScoped: true, - limit, - includeArchived: false - }; - let sessions = await store?.listAgentSessionsForUser?.(listInput) ?? []; - if (includeRouteId && !sessions.some((session) => session.id === includeSessionId)) { - const included = await visibleSessionByRouteId(store, includeRouteId, actor); - if (included) sessions = [included, ...sessions]; - } + const readModel = createWorkbenchReadModel(options, actor); + const sessions = await readModel.listSessions({ limit, includeRouteId }); const summaries = sessions.map((session) => sessionSummary(session, options)).filter(Boolean); sendJson(response, 200, { ok: true, @@ -229,7 +220,8 @@ async function handleWorkbenchSessionList(response, url, options, actor) { } async function handleWorkbenchSessionDetail(response, options, actor, sessionId) { - const session = await visibleSessionByRouteId(options.accessController?.store, sessionId, actor); + const readModel = createWorkbenchReadModel(options, actor); + const session = await readModel.getSessionByRouteId(sessionId); if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId })); sendJson(response, 200, { ok: true, @@ -242,7 +234,8 @@ async function handleWorkbenchSessionDetail(response, options, actor, sessionId) } async function handleWorkbenchMessagePage(response, url, options, actor, sessionId) { - const session = await visibleSessionByRouteId(options.accessController?.store, sessionId, actor); + const readModel = createWorkbenchReadModel(options, actor); + const session = await readModel.getSessionByRouteId(sessionId); if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId })); const messages = sessionMessages(session, options); const limit = boundedLimit(url.searchParams.get("limit")); @@ -270,21 +263,29 @@ async function handleWorkbenchTurnSnapshot(response, url, options, actor, rawTur const traceId = safeTraceId(rawTurnId) ?? queryTraceId; const turnId = safeTurnId(rawTurnId) || traceId || rawTurnId; if (!traceId) return sendJson(response, 400, workbenchError("invalid_turn_id", "turnId must currently be a trace id or include traceId query.", { turnId })); - const result = options.codeAgentChatResults?.get?.(traceId) ?? null; - const session = await visibleSessionByTrace(options.accessController?.store, traceId, actor); - if (result?.ownerUserId && !canActorReadOwner(result.ownerUserId, actor)) { + const readModel = createWorkbenchReadModel(options, actor); + const result = readModel.resultForTrace(traceId); + const session = await readModel.getSessionByTraceId(traceId); + if (result?.ownerUserId && !readModel.canReadOwner(result.ownerUserId)) { return sendJson(response, 403, workbenchError("agent_session_owner_required", "Only the session owner or admin can read this Workbench turn.", { traceId })); } const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId)); - const trace = await traceSnapshot(options, traceId); + const trace = await readModel.traceSnapshot(traceId); const status = workbenchTurnProjectionStatus(trace); const found = Boolean(resultVisible || session); if (!found) return sendJson(response, 404, workbenchError("workbench_turn_not_found", "Workbench turn is not visible to the current actor.", { turnId, traceId })); + const projection = readModel.projectionDiagnostics({ traceId, result, trace }); sendJson(response, 200, { ok: true, status, contractVersion: "workbench-turn-snapshot-v1", turn: turnSnapshot({ turnId, traceId, status, result, session, trace }), + projection, + projectionStatus: projection.projectionStatus, + lastProjectedSeq: projection.lastProjectedSeq, + sourceRunId: projection.sourceRunId, + sourceCommandId: projection.sourceCommandId, + blocker: projection.blocker, valuesRedacted: true, secretMaterialStored: false }); @@ -293,23 +294,31 @@ async function handleWorkbenchTurnSnapshot(response, url, options, actor, rawTur async function handleWorkbenchTraceEventPage(response, url, options, actor, rawTraceId) { const traceId = safeTraceId(rawTraceId); if (!traceId) return sendJson(response, 400, workbenchError("invalid_trace_id", "traceId must start with trc_.", { traceId: rawTraceId })); - const result = options.codeAgentChatResults?.get?.(traceId) ?? null; - const session = await visibleSessionByTrace(options.accessController?.store, traceId, actor); - if (result?.ownerUserId && !canActorReadOwner(result.ownerUserId, actor)) { + const readModel = createWorkbenchReadModel(options, actor); + const result = readModel.resultForTrace(traceId); + const session = await readModel.getSessionByTraceId(traceId); + if (result?.ownerUserId && !readModel.canReadOwner(result.ownerUserId)) { return sendJson(response, 403, workbenchError("agent_session_owner_required", "Only the session owner or admin can read this Workbench trace.", { traceId })); } const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId)); if (!resultVisible && !session) { return sendJson(response, 404, workbenchError("workbench_trace_not_found", "Workbench trace is not visible to the current actor.", { traceId })); } - const trace = await traceSnapshot(options, traceId); + const trace = await readModel.traceSnapshot(traceId); const page = traceEventPage(trace, tracePageOptions(url)); + const projection = readModel.projectionDiagnostics({ traceId, result, trace }); sendJson(response, 200, { ok: true, status: "succeeded", contractVersion: "workbench-trace-events-v1", traceId, ...page, + projection, + projectionStatus: projection.projectionStatus, + lastProjectedSeq: projection.lastProjectedSeq, + sourceRunId: projection.sourceRunId, + sourceCommandId: projection.sourceCommandId, + blocker: projection.blocker, valuesRedacted: true, secretMaterialStored: false }); @@ -567,61 +576,10 @@ function tracePageOptions(url) { return { afterSeq, limit: boundedLimit(url.searchParams.get("limit")) }; } -async function traceSnapshot(options, traceId) { - const memory = traceSnapshotSync(options, traceId); - if (memory?.status !== "missing" && (memory.eventCount > 0 || memory.lastEvent)) return memory; - const durable = await durableTraceSnapshot(options, traceId); - return durable ?? memory; -} - function traceSnapshotSync(options, traceId) { return (options.traceStore ?? defaultCodeAgentTraceStore).snapshot(traceId); } -async function durableTraceSnapshot(options, traceId) { - const store = options.runtimeStore; - if (!store || typeof store.queryAgentTraceEvents !== "function") return null; - let result; - try { - result = await store.queryAgentTraceEvents({ traceId }); - } catch { - return null; - } - const events = Array.isArray(result?.events) ? result.events : []; - if (events.length === 0) return null; - const normalizedEvents = events.map((event, index) => ({ ...event, traceId, seq: eventSeq(event, index) })); - const firstEvent = normalizedEvents[0] ?? null; - const lastEvent = normalizedEvents.at(-1) ?? null; - const status = durableTraceStatus(normalizedEvents); - return { - traceId, - status, - createdAt: firstEvent?.createdAt ?? null, - updatedAt: lastEvent?.createdAt ?? firstEvent?.createdAt ?? null, - startedAt: firstEvent?.createdAt ?? null, - finishedAt: TERMINAL_STATUSES.has(status) ? lastEvent?.createdAt ?? null : null, - eventCount: normalizedEvents.length, - events: normalizedEvents, - assistantStreams: [], - eventLabels: normalizedEvents.map((event) => event.label).filter(Boolean), - lastEvent, - outputTruncated: normalizedEvents.some((event) => event.outputTruncated === true), - valuesPrinted: false - }; -} - -function durableTraceStatus(events) { - const lastEvent = events.at(-1) ?? null; - const normalized = normalizeStatus(lastEvent?.status ?? lastEvent?.type); - if (lastEvent?.terminal === true) { - if (normalized === "observed" || normalized === "event" || normalized === "unknown") return "completed"; - return normalized; - } - if (TERMINAL_STATUSES.has(normalized) || RUNNING_STATUSES.has(normalized)) return normalized; - if (["failed", "error", "timeout"].includes(normalized)) return normalized; - return "running"; -} - async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId, reason }) { const context = await visibleTraceContext(options, actor, traceId); if (!context.visible) { @@ -650,12 +608,13 @@ async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId, } async function visibleTraceContext(options, actor, traceId) { - const result = options.codeAgentChatResults?.get?.(traceId) ?? null; - const session = await visibleSessionByTrace(options.accessController?.store, traceId, actor); - if (result?.ownerUserId && !canActorReadOwner(result.ownerUserId, actor)) return { visible: false }; + const readModel = createWorkbenchReadModel(options, actor); + const result = readModel.resultForTrace(traceId); + const session = await readModel.getSessionByTraceId(traceId); + if (result?.ownerUserId && !readModel.canReadOwner(result.ownerUserId)) return { visible: false }; const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId)); if (!resultVisible && !session) return { visible: false }; - const trace = await traceSnapshot(options, traceId); + const trace = await readModel.traceSnapshot(traceId); const status = workbenchTurnProjectionStatus(trace); return { visible: true, turnId: traceId, traceId, status, result, session, trace }; } diff --git a/internal/cloud/workbench-facts-store.ts b/internal/cloud/workbench-facts-store.ts new file mode 100644 index 00000000..747350f8 --- /dev/null +++ b/internal/cloud/workbench-facts-store.ts @@ -0,0 +1,165 @@ +/* + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1. + * 职责: Workbench durable facts store adapter。统一读取 session/message/turn/trace projection facts，不在 GET 中推进 AgentRun facts。 + */ +import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; +import { safeConversationId, safeSessionId, safeTraceId } from "./server-http-utils.ts"; + +export function createWorkbenchFactsStore(options = {}, actor = null) { + const accessStore = options.accessController?.store ?? null; + const traceStore = options.traceStore ?? defaultCodeAgentTraceStore; + const runtimeStore = options.runtimeStore ?? null; + const results = options.codeAgentChatResults ?? null; + + async function getSessionById(sessionId) { + const safeId = safeSessionId(sessionId); + if (!safeId) return null; + const session = await accessStore?.getAgentSession?.(safeId) ?? null; + return canActorReadSession(session, actor) ? session : null; + } + + async function getSessionByRouteId(routeId) { + const sessionId = safeSessionId(routeId); + if (sessionId) return getSessionById(sessionId); + const conversationId = safeConversationId(routeId); + if (!conversationId) return null; + const listed = await accessStore?.listAgentSessionsForUser?.({ + ownerUserId: actor?.id, + actorRole: actor?.role, + ownerScoped: true, + conversationId, + limit: 1, + includeArchived: false + }) ?? []; + const matched = listed.find((session) => canActorReadSession(session, actor)) ?? null; + if (matched) return matched; + const suffix = conversationId.slice("cnv_".length); + const deterministicSessionId = safeSessionId(`ses_${suffix}`); + return deterministicSessionId ? getSessionById(deterministicSessionId) : null; + } + + async function getSessionByTraceId(traceId) { + const safeId = safeTraceId(traceId); + if (!safeId) return null; + const session = await accessStore?.getAgentSessionByTraceId?.(safeId) ?? null; + return canActorReadSession(session, actor) ? session : null; + } + + async function listSessions({ limit, includeRouteId } = {}) { + let sessions = await accessStore?.listAgentSessionsForUser?.({ + ownerUserId: actor?.id, + actorRole: actor?.role, + ownerScoped: true, + limit, + includeArchived: false + }) ?? []; + const includeSessionId = safeSessionId(includeRouteId); + if (includeRouteId && !sessions.some((session) => session.id === includeSessionId)) { + const included = await getSessionByRouteId(includeRouteId); + if (included) sessions = [included, ...sessions]; + } + return sessions.filter((session) => canActorReadSession(session, actor)); + } + + function resultForTrace(traceId) { + const safeId = safeTraceId(traceId); + return safeId ? results?.get?.(safeId) ?? null : null; + } + + function traceSnapshotSync(traceId) { + return traceStore.snapshot(traceId); + } + + async function traceSnapshot(traceId) { + const memory = traceSnapshotSync(traceId); + if (memory?.status !== "missing" && (memory.eventCount > 0 || memory.lastEvent)) return memory; + const durable = await durableTraceSnapshot(runtimeStore, traceId); + return durable ?? memory; + } + + function subscribeTrace(traceId, listener) { + return traceStore.subscribe(traceId, listener); + } + + return { + getSessionById, + getSessionByRouteId, + getSessionByTraceId, + listSessions, + resultForTrace, + traceSnapshot, + traceSnapshotSync, + subscribeTrace, + canReadOwner: (ownerUserId) => canActorReadOwner(ownerUserId, actor) + }; +} + +async function durableTraceSnapshot(runtimeStore, traceId) { + if (!runtimeStore || typeof runtimeStore.queryAgentTraceEvents !== "function") return null; + let result; + try { + result = await runtimeStore.queryAgentTraceEvents({ traceId }); + } catch { + return null; + } + const events = Array.isArray(result?.events) ? result.events : []; + if (events.length === 0) return null; + const normalizedEvents = events.map((event, index) => ({ ...event, traceId, seq: eventSeq(event, index) })); + const firstEvent = normalizedEvents[0] ?? null; + const lastEvent = normalizedEvents.at(-1) ?? null; + const status = durableTraceStatus(normalizedEvents); + return { + traceId, + status, + createdAt: firstEvent?.createdAt ?? null, + updatedAt: lastEvent?.createdAt ?? firstEvent?.createdAt ?? null, + startedAt: firstEvent?.createdAt ?? null, + finishedAt: TERMINAL_STATUSES.has(status) ? lastEvent?.createdAt ?? null : null, + eventCount: normalizedEvents.length, + events: normalizedEvents, + assistantStreams: [], + eventLabels: normalizedEvents.map((event) => event.label).filter(Boolean), + lastEvent, + outputTruncated: normalizedEvents.some((event) => event.outputTruncated === true), + valuesPrinted: false + }; +} + +const TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled", "idle"]); +const RUNNING_STATUSES = new Set(["running", "pending", "queued", "accepted", "dispatching", "streaming", "active"]); + +function durableTraceStatus(events) { + const lastEvent = events.at(-1) ?? null; + const normalized = normalizeStatus(lastEvent?.status ?? lastEvent?.type); + if (lastEvent?.terminal === true) { + if (normalized === "observed" || normalized === "event" || normalized === "unknown") return "completed"; + return normalized; + } + if (TERMINAL_STATUSES.has(normalized) || RUNNING_STATUSES.has(normalized)) return normalized; + if (["failed", "error", "timeout"].includes(normalized)) return normalized; + return "running"; +} + +function canActorReadSession(session, actor) { + if (!session || !actor) return false; + if (normalizeStatus(session.status) === "archived") return false; + if (actor.role === "admin") return true; + return session.ownerUserId === actor.id; +} + +function canActorReadOwner(ownerUserId, actor) { + if (!ownerUserId || !actor) return true; + if (actor.role === "admin") return true; + return ownerUserId === actor.id; +} + +function eventSeq(event, index) { + const seq = Number(event?.seq); + return Number.isFinite(seq) && seq > 0 ? Math.trunc(seq) : index + 1; +} + +function normalizeStatus(value) { + const text = String(value ?? "").trim().toLowerCase().replace(/_/gu, "-"); + if (text === "cancelled") return "canceled"; + return text || "unknown"; +} diff --git a/internal/cloud/workbench-projection-finalizer.ts b/internal/cloud/workbench-projection-finalizer.ts new file mode 100644 index 00000000..2bcd6d28 --- /dev/null +++ b/internal/cloud/workbench-projection-finalizer.ts @@ -0,0 +1,81 @@ +/* + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0. + * 职责: WorkbenchProjectionFinalizer 组件入口。以 checkpoint/轮询驱动 AgentRun facts 追平，不让 GET path 承担 finalize/repair。 + */ +import { appendProjectionDiagnostic } from "./workbench-projection-writer.ts"; + +export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult = null, traceStore, getCachedResult, isCanceled, isTerminal, syncResult, onTerminal, activitySignature, noResponseTimeoutMs = null, pollIntervalMs = 1000, sleep = defaultSleep } = {}) { + if (!traceId || !currentResult?.agentRun?.runId || !currentResult?.agentRun?.commandId || typeof syncResult !== "function") return null; + let lastActivityAt = Date.now(); + let lastActivitySignature = activitySignature?.(currentResult, traceStore?.snapshot?.(traceId)) ?? null; + let idleDegraded = false; + + setImmediate(() => { + void (async () => { + let result = currentResult; + let lastError = null; + while (true) { + const cached = getCachedResult?.(traceId); + if (isCanceled?.(cached)) return; + if (isTerminal?.(cached?.status)) { + onTerminal?.(cached, { preserveLastTraceId: true }); + return; + } + try { + const synced = await syncResult({ result }); + result = synced?.result ?? result; + if (result && isTerminal?.(result.status)) { + onTerminal?.(result, { preserveLastTraceId: true }); + return; + } + const signature = activitySignature?.(result, synced?.runnerTrace ?? traceStore?.snapshot?.(traceId)); + if (signature && signature !== lastActivitySignature) { + lastActivitySignature = signature; + lastActivityAt = Date.now(); + idleDegraded = false; + } + lastError = null; + } catch (error) { + lastError = error; + } + const idleMs = Date.now() - lastActivityAt; + if (noResponseTimeoutMs && idleMs >= noResponseTimeoutMs && !idleDegraded) { + appendProjectionDiagnostic(traceStore, traceId, { + type: "turn-status", + status: "degraded", + label: "projection-finalizer:no-response-idle-timeout", + errorCode: lastError?.code ?? "no_response_idle_timeout", + degradedReason: lastError ? "projection_activity_stale" : "no_response_idle_timeout", + message: lastError?.message ?? `AgentRun projection has no new activity for ${idleMs}ms; idle threshold ${noResponseTimeoutMs}ms.`, + runId: result?.agentRun?.runId ?? currentResult.agentRun.runId, + commandId: result?.agentRun?.commandId ?? currentResult.agentRun.commandId, + timeoutMs: noResponseTimeoutMs, + idleMs, + lastActivityAt: new Date(lastActivityAt).toISOString(), + waitingFor: "agentrun-result-activity", + terminal: false + }); + idleDegraded = true; + } + await sleep(pollIntervalMs); + } + })().catch((error) => { + appendProjectionDiagnostic(traceStore, traceId, { + type: "turn-status", + status: "degraded", + label: "projection-finalizer:sync-crashed", + errorCode: error?.code ?? "workbench_projection_finalizer_crashed", + message: error?.message ?? "Workbench projection finalizer crashed.", + runId: currentResult.agentRun.runId, + commandId: currentResult.agentRun.commandId, + waitingFor: "agentrun-result", + terminal: false + }); + }); + }); + return { scheduled: true, traceId, noResponseTimeoutMs, pollIntervalMs, valuesPrinted: false }; +} + +function defaultSleep(ms) { + return new Promise((resolve) => setTimeout(resolve, Math.max(0, Number(ms) || 0))); +} diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts new file mode 100644 index 00000000..b7f269ef --- /dev/null +++ b/internal/cloud/workbench-projection-writer.ts @@ -0,0 +1,46 @@ +/* + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0. + * 职责: WorkbenchProjectionWriter 组件入口。唯一封装 Code Agent/AgentRun facts 到 Workbench session projection facts 的持久化写入。 + */ +import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; +import { safeTraceId } from "./server-http-utils.ts"; + +export async function writeWorkbenchProjectionSession({ accessController, traceStore = defaultCodeAgentTraceStore, traceId, ownerUserId, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, preserveLastTraceId = false } = {}) { + if (!ownerUserId || typeof accessController?.recordAgentSessionOwner !== "function") return null; + const safeId = safeTraceId(traceId); + try { + return await accessController.recordAgentSessionOwner({ + ownerUserId, + ownerRole, + sessionId, + projectId, + conversationId, + threadId, + traceId: preserveLastTraceId ? undefined : safeId, + status, + session + }); + } catch (error) { + if (safeId) appendProjectionDiagnostic(traceStore, safeId, { + type: "session-owner", + status: "degraded", + label: "projection-writer:session-owner:persist-failed", + errorCode: "workbench_projection_session_persist_failed", + message: error?.message ?? "Workbench projection session persistence failed.", + valuesPrinted: false + }); + return null; + } +} + +export function appendProjectionDiagnostic(traceStore = defaultCodeAgentTraceStore, traceId, event = {}) { + const safeId = safeTraceId(traceId); + if (!safeId || typeof traceStore?.append !== "function") return null; + return traceStore.append(safeId, { + type: "projection-diagnostic", + status: "degraded", + label: "projection:diagnostic", + ...event, + valuesPrinted: false + }); +} diff --git a/internal/cloud/workbench-read-model.ts b/internal/cloud/workbench-read-model.ts new file mode 100644 index 00000000..fc453ebe --- /dev/null +++ b/internal/cloud/workbench-read-model.ts @@ -0,0 +1,62 @@ +/* + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1. + * 职责: WorkbenchReadModel 组件入口。只通过 WorkbenchFactsStore 读取 durable projection facts，不执行 AgentRun sync/finalize/repair。 + */ +import { createWorkbenchFactsStore } from "./workbench-facts-store.ts"; + +export function createWorkbenchReadModel(options = {}, actor = null) { + const facts = createWorkbenchFactsStore(options, actor); + return { + facts, + listSessions: (input = {}) => facts.listSessions(input), + getSessionById: (sessionId) => facts.getSessionById(sessionId), + getSessionByRouteId: (routeId) => facts.getSessionByRouteId(routeId), + getSessionByTraceId: (traceId) => facts.getSessionByTraceId(traceId), + resultForTrace: (traceId) => facts.resultForTrace(traceId), + traceSnapshot: (traceId) => facts.traceSnapshot(traceId), + traceSnapshotSync: (traceId) => facts.traceSnapshotSync(traceId), + subscribeTrace: (traceId, listener) => facts.subscribeTrace(traceId, listener), + canReadOwner: (ownerUserId) => facts.canReadOwner(ownerUserId), + projectionDiagnostics: ({ traceId, result = null, trace = null } = {}) => projectionDiagnostics({ traceId, result, trace }) + }; +} + +export function projectionDiagnostics({ traceId, result = null, trace = null } = {}) { + const agentRun = result?.agentRun && typeof result.agentRun === "object" ? result.agentRun : null; + const events = Array.isArray(trace?.events) ? trace.events : []; + const lastEvent = trace?.lastEvent ?? events.at(-1) ?? null; + const lastProjectedSeq = Number(lastEvent?.seq ?? trace?.lastProjectedSeq ?? 0); + const sourceRunId = agentRun?.runId ?? lastEvent?.runId ?? null; + const sourceCommandId = agentRun?.commandId ?? lastEvent?.commandId ?? null; + const blocker = result?.blocker ?? result?.error ?? null; + const status = blocker + ? "blocked" + : isTerminalStatus(result?.status ?? trace?.status) + ? "caught-up" + : result || trace?.status !== "missing" + ? "projecting" + : "unknown"; + return { + projectionStatus: status, + lastProjectedSeq: Number.isFinite(lastProjectedSeq) && lastProjectedSeq > 0 ? Math.trunc(lastProjectedSeq) : null, + sourceRunId, + sourceCommandId, + blocker: blocker ? diagnosticBlocker(blocker) : null, + updatedAt: trace?.updatedAt ?? result?.updatedAt ?? null, + valuesRedacted: true + }; +} + +function diagnosticBlocker(value = {}) { + if (!value || typeof value !== "object") return { code: "projection_blocked", summary: String(value), valuesRedacted: true }; + return { + code: String(value.code ?? value.errorCode ?? "projection_blocked"), + summary: String(value.summary ?? value.userMessage ?? value.message ?? value.code ?? "Projection is blocked."), + retryable: typeof value.retryable === "boolean" ? value.retryable : null, + valuesRedacted: true + }; +} + +function isTerminalStatus(value) { + return ["completed", "failed", "blocked", "timeout", "cancelled", "canceled", "idle"].includes(String(value ?? "").trim().toLowerCase()); +} diff --git a/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts b/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts index 6ed5024e..1fdf5c1b 100644 --- a/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts @@ -1,9 +1,12 @@ +// SPEC: PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1. +// Responsibility: Regression tests for Workbench route parsing, server-state projection, and session tab status authority. + import assert from "node:assert/strict"; import test from "node:test"; import type { WorkbenchSessionRecord } from "../src/types/index.ts"; import { initialWorkbenchSessionIdFromLocation } from "../src/stores/workbench-projection.ts"; -import { createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, selectActiveSession, selectTraceAuthorityById } from "../src/stores/workbench-server-state.ts"; +import { createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, selectActiveSession, selectSessionStatusAuthority, selectTraceAuthorityById } from "../src/stores/workbench-server-state.ts"; import { sessionToSessionTab, stableSessionList } from "../src/stores/workbench-session.ts"; test("Workbench route parsing accepts only session routes", () => { @@ -75,3 +78,25 @@ test("Workbench session tab status is read from server projection", () => { assert.equal(tab.status, "running"); assert.equal(tab.running, true); }); + +test("Workbench server-state session detail updates session status authority", () => { + let state = createWorkbenchServerState(); + state = reduceWorkbenchServerState(state, { + type: "session.status", + session: { sessionId: "ses_backfill", status: "running" } + }); + state = reduceWorkbenchServerState(state, { + type: "session.detail", + session: { + sessionId: "ses_backfill", + threadId: "thr_backfill", + status: "completed", + lastTraceId: "trc_backfill", + updatedAt: "2026-06-18T00:00:03.000Z" + } + }); + + const authority = selectSessionStatusAuthority(state).ses_backfill; + assert.equal(authority?.status, "completed"); + assert.equal(authority?.lastTraceId, "trc_backfill"); +}); diff --git a/web/hwlab-cloud-web/src/stores/workbench-server-state.ts b/web/hwlab-cloud-web/src/stores/workbench-server-state.ts index ccaf98be..0420dd31 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-server-state.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-server-state.ts @@ -1,4 +1,4 @@ -// SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1. +// SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection. // Responsibility: Normalized Workbench server-state reducer for session, turn, and trace caches. import type { ChatMessage, WorkbenchSessionRecord } from "@/types"; @@ -73,9 +73,13 @@ export function selectActiveMessages(state: WorkbenchServerState, sessionId: str function reduceSessionDetail(state: WorkbenchServerState, session: WorkbenchSessionRecord | null | undefined): WorkbenchServerState { const sessionId = session?.sessionId; if (!sessionId || !session) return state; + const sessionStatus = sessionStatusAuthorityFromDetail(session); return { ...state, sessionsById: { ...state.sessionsById, [sessionId]: session }, + sessionStatusById: sessionStatus + ? { ...state.sessionStatusById, [sessionId]: sessionStatus } + : state.sessionStatusById, messagesBySessionId: { ...state.messagesBySessionId, [sessionId]: Array.isArray(session.messages) ? session.messages : state.messagesBySessionId[sessionId] ?? [] @@ -83,6 +87,18 @@ function reduceSessionDetail(state: WorkbenchServerState, session: WorkbenchSess }; } +function sessionStatusAuthorityFromDetail(session: WorkbenchSessionRecord): SessionStatusAuthority | null { + const sessionId = session.sessionId; + if (!sessionId || typeof session.status !== "string") return null; + return { + sessionId, + status: session.status, + updatedAt: typeof session.updatedAt === "string" ? session.updatedAt : null, + lastTraceId: typeof session.lastTraceId === "string" ? session.lastTraceId : null, + loadedAt: new Date().toISOString() + }; +} + function withoutKey(source: Record, key: string): Record { if (!(key in source)) return source; const next = { ...source }; diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 5e2fe67b..b0c391e7 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -95,9 +95,10 @@ export const useWorkbenchStore = defineStore("workbench", () => { if (targetSessionId) sessionDetailLoadingId.value = targetSessionId; const selected = targetSessionId ? await loadWorkbenchSession(targetSessionId, listedSessions.find((item) => item.sessionId === targetSessionId) ?? null) : null; clearSessionDetailLoading(targetSessionId); - if (selected && isCurrentSessionSelection(requestEpoch, selected.sessionId)) applySelectedSessionDetail(selected, messages.value); - if (selected && routeRequestId && !isCurrentSessionSelection(requestEpoch, selected.sessionId)) applySelectedSessionDetail(selected, messages.value); - if (options.invalidRouteId || (targetSessionId && !selected)) isolateSessionLoadFailure(targetSessionId ?? options.invalidRouteId ?? "invalid-session", null, [], "invalid session URL", { restorePrevious: false }); + if (selected && !isArchivedSession(selected) && isCurrentSessionSelection(requestEpoch, selected.sessionId)) applySelectedSessionDetail(selected, messages.value); + if (selected && !isArchivedSession(selected) && routeRequestId && !isCurrentSessionSelection(requestEpoch, selected.sessionId)) applySelectedSessionDetail(selected, messages.value); + if (selected && isArchivedSession(selected)) isolateSessionLoadFailure(selected.sessionId, null, [], "session archived", { restorePrevious: false }); + if (options.invalidRouteId || (targetSessionId && !selected)) isolateSessionLoadFailure(targetSessionId ?? options.invalidRouteId ?? "invalid-session", null, [], "session URL not found", { restorePrevious: false }); const nextSessions = stableSessionList(sessions.value, listedSessions, selected?.sessionId ?? routeSessionId ?? includeSessionId, selected); sessions.value = nextSessions; rememberSessionList(nextSessions); @@ -935,7 +936,8 @@ async function loadWorkbenchSession(sessionId: string, seed: WorkbenchSessionRec const requestId = normalizeWorkbenchSessionRouteId(sessionId); if (!requestId) return null; const detail = await api.workbench.session(requestId); - const detailSession = detail.ok ? sessionFromWorkbenchSession(detail.data?.session) : null; + if (!detail.ok) return null; + const detailSession = sessionFromWorkbenchSession(detail.data?.session); const id = detailSession?.sessionId ?? normalizeWorkbenchSessionId(requestId) ?? seed?.sessionId; if (!id) return null; const messages = await api.workbench.sessionMessages(id, { limit: 100 });