diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index 5ac749c0..2ae60ebb 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -621,9 +621,6 @@ lanes: HWLAB_CLOUD_WEB_DISPLAY_TIME_ZONE: Asia/Shanghai HWLAB_CLOUD_WEB_DISPLAY_TIME_LOCALE: zh-CN HWLAB_CLOUD_WEB_DISPLAY_TIME_LABEL: 北京时间 - HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true" - HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED: "true" - HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false" HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true" HWLAB_CLOUD_WEB_OPENCODE_UPSTREAM_URL: http://opencode-server.hwlab-v03.svc.cluster.local:4096 HWLAB_CLOUD_WEB_OPENCODE_USERNAME: secretRef:hwlab-opencode-server-auth/username @@ -850,11 +847,6 @@ lanes: HWLAB_CODE_AGENT_AGENTRUN_REPO_URL: http://gitea-http.devops-infra.svc.cluster.local:3000/mirrors/pikasTech-HWLAB.git HWLAB_KAFKA_ENABLED: "true" HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED: "true" - HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "true" - HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true" - HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "false" - HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "false" - HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false" HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true" HWLAB_KAFKA_BOOTSTRAP_SERVERS: platform-infra-kafka-kafka-bootstrap.platform-infra.svc.cluster.local:9092 HWLAB_KAFKA_STDIO_TOPIC: codex-stdio.raw.v1 diff --git a/internal/cloud/server-code-agent-admission-http.ts b/internal/cloud/server-code-agent-admission-http.ts index 67f014d4..029837b9 100644 --- a/internal/cloud/server-code-agent-admission-http.ts +++ b/internal/cloud/server-code-agent-admission-http.ts @@ -1042,7 +1042,6 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) { } }); if (adapterEnabled) { - const transactionalAdmission = true; const pendingPromotion = results.get(traceId); if (pendingPromotion?.admissionState === "promotion-pending" && pendingPromotion?.agentRun?.durableDispatch === true) { const retryParams = { ...params, userBillingReservation: pendingPromotion.userBillingReservation ?? null }; @@ -1057,15 +1056,13 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) { } const lifecycle = codeAgentTurnLifecycleFields(traceId, params); - if (transactionalAdmission) { - await persistCodeAgentAdmissionInputState({ - params, - options, - traceId, - status: "admitting", - messageId: lifecycle.userMessageId - }); - } + await persistCodeAgentAdmissionInputState({ + params, + options, + traceId, + status: "admitting", + messageId: lifecycle.userMessageId + }); traceStore.append(traceId, { type: "request", @@ -1086,16 +1083,14 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) { error: billingPreflight?.blocked ? billingPreflight?.payload?.error?.message ?? billingPreflight?.blocker?.summary ?? "billing preflight blocked" : null }); if (billingPreflight?.blocked) { - if (transactionalAdmission) { - await persistCodeAgentAdmissionInputState({ - params, - options, - traceId, - status: "blocked", - messageId: lifecycle.userMessageId, - error: billingPreflight.payload?.error ?? billingPreflight.blocker ?? { code: "billing_preflight_blocked" } - }); - } + await persistCodeAgentAdmissionInputState({ + params, + options, + traceId, + status: "blocked", + messageId: lifecycle.userMessageId, + error: billingPreflight.payload?.error ?? billingPreflight.blocker ?? { code: "billing_preflight_blocked" } + }); traceStore.append(traceId, { type: "billing", status: "blocked", @@ -1121,16 +1116,14 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) { const failure = codeAgentDispatchFailure(error); const rejected = codeAgentSubmissionUnavailablePayload(failure, { params: dispatchParams, options: executionOptions, traceId }); await finalizeCodeAgentBillingUsage({ payload: rejected, params: dispatchParams, options: executionOptions }); - if (transactionalAdmission) { - await persistCodeAgentAdmissionInputState({ - params: dispatchParams, - options: executionOptions, - traceId, - status: "failed", - messageId: lifecycle.userMessageId, - error: failure.payload?.error ?? error - }); - } + await persistCodeAgentAdmissionInputState({ + params: dispatchParams, + options: executionOptions, + traceId, + status: "failed", + messageId: lifecycle.userMessageId, + error: failure.payload?.error ?? error + }); void emitCodeAgentOtelSpan("agentrun_dispatch", traceId, executionOptions.env ?? process.env, { startTimeMs: dispatchStartedAt, status: "error", @@ -1151,17 +1144,15 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) { }, dispatchParams); results.set(traceId, pending); try { - if (transactionalAdmission) { - await persistCodeAgentAdmissionInputState({ - params: dispatchParams, - options: executionOptions, - traceId, - status: "admitting", - messageId: lifecycle.userMessageId, - commandId: payload.agentRun?.commandId ?? null, - agentRun: payload.agentRun ?? null - }); - } + await persistCodeAgentAdmissionInputState({ + params: dispatchParams, + options: executionOptions, + traceId, + status: "admitting", + messageId: lifecycle.userMessageId, + commandId: payload.agentRun?.commandId ?? null, + agentRun: payload.agentRun ?? null + }); await promoteCodeAgentTurnAdmission({ payload: pending, params: dispatchParams, options: executionOptions, traceId }); } catch (error) { throw codeAgentPromotionUnavailableError(error, { payload: pending, params: dispatchParams, options: executionOptions, traceId }); diff --git a/internal/cloud/server-workbench-facts.ts b/internal/cloud/server-workbench-facts.ts index 6b5805e4..a3600a63 100644 --- a/internal/cloud/server-workbench-facts.ts +++ b/internal/cloud/server-workbench-facts.ts @@ -16,7 +16,6 @@ import { import { createWorkbenchReadModel } from "./workbench-read-model.ts"; import { createWorkbenchRuntimeClient } from "./workbench-runtime-client.ts"; import { buildWorkbenchSessionDetail, compactLaunchContext, includeMessagesForSessionDetail } from "./workbench-session-detail-response.ts"; -import { handleWorkbenchSyncHttp } from "./workbench-realtime-authority.ts"; import { durableTraceStatus, RUNNING_STATUSES, terminalFinalResponse, TERMINAL_STATUSES } from "./workbench-turn-projection.ts"; import { emitCodeAgentOtelSpan, emitHttpServerRequestSpan } from "./otel-trace.ts"; diff --git a/internal/cloud/server-workbench-realtime-http.test.ts b/internal/cloud/server-workbench-realtime-http.test.ts index 5b722a88..1de352a2 100644 --- a/internal/cloud/server-workbench-realtime-http.test.ts +++ b/internal/cloud/server-workbench-realtime-http.test.ts @@ -71,22 +71,6 @@ test("projection outbox emits immutable assistant versions in row order", () => assert.deepEqual(events.map((item) => item.payload.cursor.outboxSeq), [10, 11]); }); -test("workbench sync repair endpoint is removed from the product API", async () => { - const server = createCloudApiServer({ - accessController: realtimeAccessController(), - workbenchRuntime: {}, - kafkaEventBridge: projectionRealtimeBridge(), - env: { ...TRANSACTIONAL_REALTIME_ENV } - }); - await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); - try { - const response = await getJson(server.address().port, "/v1/workbench/sync?sessionId=ses_removed_sync_repair"); - assert.equal(response.status, 404); - } finally { - await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); - } -}); - test("workbench realtime initial connection emits current snapshot without replaying historical outbox", async () => { const sessionId = "ses_realtime_snapshot_only"; const traceId = "trc_realtime_snapshot_only"; diff --git a/internal/cloud/server-workbench-realtime-http.ts b/internal/cloud/server-workbench-realtime-http.ts index 3798ddcc..4f6a7115 100644 --- a/internal/cloud/server-workbench-realtime-http.ts +++ b/internal/cloud/server-workbench-realtime-http.ts @@ -15,7 +15,6 @@ import { import { createWorkbenchReadModel } from "./workbench-read-model.ts"; import { createWorkbenchRuntimeClient } from "./workbench-runtime-client.ts"; import { buildWorkbenchSessionDetail, compactLaunchContext, includeMessagesForSessionDetail } from "./workbench-session-detail-response.ts"; -import { handleWorkbenchSyncHttp } from "./workbench-realtime-authority.ts"; import { projectionOutboxRealtimeEvents } from "./workbench-projection-outbox-events.ts"; import { durableTraceStatus, RUNNING_STATUSES, TERMINAL_STATUSES } from "./workbench-turn-projection.ts"; import { emitCodeAgentOtelSpan, emitHttpServerRequestSpan } from "./otel-trace.ts"; diff --git a/scripts/gitops-render.test.ts b/scripts/gitops-render.test.ts index 5902b4f7..eb9d7663 100644 --- a/scripts/gitops-render.test.ts +++ b/scripts/gitops-render.test.ts @@ -687,11 +687,14 @@ test("v03 render keeps node identity as data instead of generated structure", as assert.equal(cloudApiEnv.get("HWLAB_CLOUD_DB_SSL_MODE"), "require"); assert.equal(cloudApiEnv.get("HWLAB_KAFKA_ENABLED"), "true"); assert.equal(cloudApiEnv.get("HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED"), "true"); - assert.equal(cloudApiEnv.get("HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED"), "true"); - assert.equal(cloudApiEnv.get("HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED"), "true"); - assert.equal(cloudApiEnv.get("HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED"), "false"); - assert.equal(cloudApiEnv.get("HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED"), "false"); - assert.equal(cloudApiEnv.get("HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED"), "false"); + for (const name of [ + "HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", + "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", + "HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", + "HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", + "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED" + ]) assert.equal(cloudApiEnv.has(name), false); assert.equal(cloudApiEnv.get("HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED"), "true"); assert.equal(cloudApiEnv.get("HWLAB_KAFKA_BOOTSTRAP_SERVERS"), "platform-infra-kafka-kafka-bootstrap.platform-infra.svc.cluster.local:9092"); assert.equal(cloudApiEnv.get("HWLAB_KAFKA_STDIO_TOPIC"), "codex-stdio.raw.v1"); @@ -709,8 +712,14 @@ test("v03 render keeps node identity as data instead of generated structure", as const cloudWeb = (workloadsJson.items ?? []).find((item) => item.kind === "Deployment" && item.metadata?.name === "hwlab-cloud-web"); const cloudWebContainer = collectContainersFromItem(cloudWeb).find((container) => container.name === "hwlab-cloud-web"); const cloudWebEnv = new Map((cloudWebContainer?.env ?? []).map((entry) => [entry.name, entry.value])); - assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED"), "true"); - assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED"), "false"); + for (const name of [ + "HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", + "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", + "HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", + "HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", + "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED" + ]) assert.equal(cloudWebEnv.has(name), false); assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED"), "true"); assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_ENABLED"), "1"); assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_ENTRIES"), "200"); diff --git a/scripts/runtime-realtime-capabilities-config.test.mjs b/scripts/runtime-realtime-capabilities-config.test.mjs index 86b74b8f..a037d55d 100644 --- a/scripts/runtime-realtime-capabilities-config.test.mjs +++ b/scripts/runtime-realtime-capabilities-config.test.mjs @@ -3,13 +3,14 @@ import test from "node:test"; import { readStructuredFile } from "./src/structured-config.mjs"; -const PRODUCT_CAPABILITIES = Object.freeze({ - HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "true", - HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true", - HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "false", - HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "false", - HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false" -}); +const OBSOLETE_ARCHITECTURE_ENV = Object.freeze([ + "HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", + "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", + "HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", + "HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", + "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED" +]); const LIVE_KAFKA_CONTRACT = Object.freeze({ HWLAB_KAFKA_STDIO_TOPIC: "codex-stdio.raw.v1", @@ -21,7 +22,6 @@ const LIVE_KAFKA_CONTRACT = Object.freeze({ }); const UNIDESK_OWNED_REFRESH_REPLAY_ENV = Object.freeze([ - "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX", "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS", "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT", @@ -37,15 +37,16 @@ const ISOLATED_DEBUG_CONTRACT = Object.freeze({ HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_TIMEOUT_MS: "15000" }); -test("v03 YAML keeps pure Kafka realtime capabilities independently composable", async () => { +test("v03 YAML fixes one transactional projection architecture without architecture env", async () => { const deploy = await readStructuredFile(process.cwd(), "deploy/deploy.yaml"); const cloudApi = deploy.lanes?.v03?.services?.find((service) => service.serviceId === "hwlab-cloud-api"); const cloudWeb = deploy.lanes?.v03?.serviceDeclarations?.["hwlab-cloud-web"]; assert.ok(cloudApi, "deploy.lanes.v03.services must declare hwlab-cloud-api"); assert.ok(cloudWeb, "deploy.lanes.v03.serviceDeclarations must declare hwlab-cloud-web"); - for (const [name, expected] of Object.entries(PRODUCT_CAPABILITIES)) { - assert.equal(cloudApi.env?.[name], expected, `${name} must be explicitly YAML-owned`); + for (const name of OBSOLETE_ARCHITECTURE_ENV) { + assert.equal(cloudApi.env?.[name], undefined, `${name} must not be rendered into cloud-api`); + assert.equal(cloudWeb.env?.[name], undefined, `${name} must not be rendered into cloud-web`); } for (const [name, expected] of Object.entries(LIVE_KAFKA_CONTRACT)) { assert.equal(cloudApi.env?.[name], expected, `${name} must preserve the pure Kafka live contract`); @@ -59,8 +60,5 @@ test("v03 YAML keeps pure Kafka realtime capabilities independently composable", assert.equal(cloudApi.env?.HWLAB_SESSION_COOKIE_DOMAIN, undefined, "internal and public origins require host-scoped session cookies"); assert.notEqual(cloudApi.env?.HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID, "hwlab-v03-agentrun-event-bridge"); - assert.equal(cloudWeb.env?.HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED, "true"); - assert.equal(cloudWeb.env?.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED, "true"); - assert.equal(cloudWeb.env?.HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED, "false"); assert.equal(cloudWeb.env?.HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED, "true"); }); diff --git a/web/hwlab-cloud-web/scripts/check.ts b/web/hwlab-cloud-web/scripts/check.ts index 2c7f9d1b..c8c2eab8 100644 --- a/web/hwlab-cloud-web/scripts/check.ts +++ b/web/hwlab-cloud-web/scripts/check.ts @@ -199,7 +199,6 @@ assertIncludes(workbenchRealtimePlanSource, "afterOutboxSeq: finiteNumber(recove assert.doesNotMatch(workbenchRealtimePlanSource, /force:\s*true/u, "Realtime recovery planner must not turn transport recovery into force-refresh work"); assertIncludes(workbenchColadaSource, "const state = await queryCache.refresh(entry);", "Workbench reads must preserve Colada staleTime/min-interval governance"); assert.doesNotMatch(workbenchColadaSource, /queryCache\.fetch\(entry/u, "Workbench reads must not call queryCache.fetch(entry), which bypasses Colada freshness governance"); -assertIncludes(workbenchStoreSource, "runtimePolicy.sessionListRealtimeRefreshDelayMs", "Realtime recovery delay must come from runtime policy instead of store constants"); assertIncludes(workbenchStoreSource, "workbenchColadaQueries.fetchSession", "Realtime session detail recovery must enter Colada query facade"); assertIncludes(workbenchStoreSource, "runtimePolicy.workbenchSessionDetailMinRefreshMs", "Realtime session detail recovery budget must come from runtime policy"); assert.doesNotMatch(workbenchStoreSource, /refreshRealtimeSessionMessages[\s\S]{0,900}refreshSessionMessageProjectionPage\(id, \{ force: true \}\)/u, "Realtime session message recovery must not force-bypass the message projection refresh budget"); diff --git a/web/hwlab-cloud-web/scripts/workbench-fixed-transactional-architecture.test.ts b/web/hwlab-cloud-web/scripts/workbench-fixed-transactional-architecture.test.ts index 173fa22a..50b05742 100644 --- a/web/hwlab-cloud-web/scripts/workbench-fixed-transactional-architecture.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-fixed-transactional-architecture.test.ts @@ -7,6 +7,7 @@ const repositoryRoot = fileURLToPath(new URL("../../..", import.meta.url)); const productionFiles = [ "internal/cloud/kafka-event-bridge.ts", "internal/cloud/server-workbench-realtime-http.ts", + "internal/cloud/server-workbench-facts.ts", "internal/cloud/server-code-agent-admission-http.ts", "internal/cloud/server.ts", "web/hwlab-cloud-web/src/config/runtime.ts", @@ -14,8 +15,11 @@ const productionFiles = [ "web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts", "web/hwlab-cloud-web/src/utils/workbench-realtime-runtime.ts", "web/hwlab-cloud-web/src/stores/workbench.ts", + "web/hwlab-cloud-web/src/stores/workbench-colada-queries.ts", "web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts", - "web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts" + "web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts", + "web/hwlab-cloud-web/src/stores/workbench-session.ts", + "web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts" ]; const forbiddenArchitectureIdentifiers = [ "directPublish", @@ -34,7 +38,26 @@ const forbiddenArchitectureIdentifiers = [ "workbenchHistoryAuthorityPolicy", "workbenchRealtimeTransportEnabled", "workbenchRealtimeTraceIdForCapabilities", - "workbenchProjectionEventStreamPath" + "workbenchProjectionEventStreamPath", + "scheduleSessionListRefresh", + "invalidateSessionList", + "fetchWorkbenchTurnStatus", + "refreshTurnStatusByTraceId", + "refreshMessageProjectionForTrace", + "selectActiveTurnStatusRefreshTraceIds", + "sessionDetailAutoReadDecision", + "projectionMergeCommitSummary", + "forceRead", + "handleWorkbenchSyncHttp", + "/v1/workbench/sync" +]; +const obsoleteArchitectureEnv = [ + "HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", + "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", + "HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", + "HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", + "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED" ]; test("Workbench production source has one fixed transactional projection architecture", async () => { @@ -48,6 +71,12 @@ test("Workbench production source has one fixed transactional projection archite assert.deepEqual(violations, []); }); +test("production deploy does not render architecture selection env", async () => { + const deploySource = await readFile(`${repositoryRoot}/deploy/deploy.yaml`, "utf8"); + const violations = obsoleteArchitectureEnv.filter((name) => deploySource.includes(name)); + assert.deepEqual(violations, []); +}); + test("raw hwlab Kafka fixtures remain isolated to the explicit debug path", async () => { const fixtureSource = await readFile(`${repositoryRoot}/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts`, "utf8"); const debugSource = await readFile(`${repositoryRoot}/web/hwlab-cloud-web/src/stores/workbench-isolated-kafka-debug.ts`, "utf8"); diff --git a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts index fd4d943b..19e74333 100644 --- a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts @@ -25,7 +25,6 @@ import { planWorkbenchRealtimeApply, planWorkbenchRealtimeRecovery } from "../sr import { WORKBENCH_REALTIME_AUTHORITY_VERSION, workbenchRealtimePrimaryAuthorityDecision } from "../src/stores/workbench-realtime-authority.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"]); @@ -471,42 +470,6 @@ 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(realtimeEvent({ type: "trace.event", traceId: "trc_1", event: { traceId: "trc_1", label: "delta" }, snapshot: { traceId: "trc_1", status: "running" }, entity: { family: "traceEvents", id: "trc_1:1", version: 1, projectionRevision: "prj_1" } }), "workbench.trace.event"); assert.equal(trace.activityLabel, "realtime:trace.event"); diff --git a/web/hwlab-cloud-web/src/api/workbench.ts b/web/hwlab-cloud-web/src/api/workbench.ts index e19e14ca..7f351581 100644 --- a/web/hwlab-cloud-web/src/api/workbench.ts +++ b/web/hwlab-cloud-web/src/api/workbench.ts @@ -129,7 +129,6 @@ export const workbenchAPI = { return fetchJson(sessionDetailPath(sessionId, detailOptions), { timeoutMs: requestTimeoutMs(detailOptions.timeoutMs, SESSION_DETAIL_TIMEOUT_MS), timeoutName: "workbench session" }); }, sessionMessages: (sessionId: string, options: SessionMessageOptions = {}): Promise> => fetchJson(sessionMessagesPath(sessionId, options), { timeoutMs: requestTimeoutMs(options.timeoutMs, SESSION_DETAIL_TIMEOUT_MS), timeoutName: "workbench session messages" }), - turn: async (traceId: string, timeoutMs = 8000, activityRef?: ApiRequestOptions["activityRef"]): Promise> => normalizeWorkbenchTurnResult(await fetchJson>(`/v1/workbench/turns/${encodeURIComponent(traceId)}`, { timeoutMs, timeoutName: "workbench turn", activityRef }), traceId), traceEvents: async (traceId: string, timeoutMs = 8000, activityRef?: ApiRequestOptions["activityRef"], options?: WorkbenchTraceRequestOptions): Promise> => normalizeWorkbenchTraceResult(await fetchJson>(workbenchTracePath(traceId, options), { timeoutMs, timeoutName: "workbench trace", activityRef }), traceId) }; @@ -186,42 +185,6 @@ function workbenchTracePath(traceId: string, options: WorkbenchTraceRequestOptio return `/v1/workbench/traces/${encodeURIComponent(traceId)}/events${query ? `?${query}` : ""}`; } -function normalizeWorkbenchTurnResult(result: ApiResult>, traceId: string): ApiResult { - if (!result.ok || !result.data) return result as ApiResult; - const turn = recordValue(result.data.turn) ?? result.data; - const trace = recordValue(turn.trace); - const status = normalizeStatus(textValue(turn.status)) ?? "unknown"; - const projection = normalizeProjectionDiagnostic(result.data.projection, turn.projection, result.data); - const timing = normalizeTimingProjection(turn); - return { - ...result, - data: { - ...turn, - projection, - timing, - startedAt: timing?.startedAt ?? undefined, - lastEventAt: timing?.lastEventAt ?? undefined, - finishedAt: timing?.finishedAt ?? undefined, - durationMs: timing?.durationMs ?? undefined, - projectionStatus: projection?.projectionStatus ?? textValue(result.data.projectionStatus) ?? undefined, - projectionHealth: projection?.projectionHealth ?? textValue(result.data.projectionHealth) ?? undefined, - lastProjectedSeq: Number.isFinite(Number(result.data.lastProjectedSeq)) ? Number(result.data.lastProjectedSeq) : undefined, - sourceRunId: projection?.sourceRunId ?? textValue(result.data.sourceRunId) ?? undefined, - sourceCommandId: projection?.sourceCommandId ?? textValue(result.data.sourceCommandId) ?? undefined, - staleMs: projection?.staleMs ?? numberValue(result.data.staleMs) ?? undefined, - blocker: projection?.blocker ?? recordValue(result.data.blocker) ?? undefined, - traceId: textValue(turn.traceId) ?? traceId, - status, - running: typeof turn.running === "boolean" ? turn.running : undefined, - terminal: typeof turn.terminal === "boolean" ? turn.terminal : undefined, - sessionId: textValue(turn.sessionId) ?? undefined, - threadId: textValue(turn.threadId) ?? undefined, - agentRun: recordValue(turn.agentRun) ?? undefined, - updatedAt: textValue(turn.updatedAt) ?? textValue(result.data.observedAt) ?? undefined - } as AgentChatResultResponse - }; -} - function normalizeWorkbenchTraceResult(result: ApiResult>, traceId: string): ApiResult { if (!result.ok || !result.data) return result as ApiResult; const events = Array.isArray(result.data.events) ? result.data.events : []; @@ -308,13 +271,6 @@ function normalizeTimingProjection(value: unknown): WorkbenchTurnTimingProjectio } as WorkbenchTurnTimingProjection; } -function normalizeStatus(value: unknown): string | null { - const text = textValue(value); - if (!text) return null; - const normalized = text.toLowerCase().replace(/_/gu, "-"); - return normalized === "cancelled" ? "canceled" : normalized; -} - function recordValue(value: unknown): Record | null { return value && typeof value === "object" ? value as Record : null; } diff --git a/web/hwlab-cloud-web/src/config/workbench-runtime-policy.ts b/web/hwlab-cloud-web/src/config/workbench-runtime-policy.ts index ac94a12c..ec8b778b 100644 --- a/web/hwlab-cloud-web/src/config/workbench-runtime-policy.ts +++ b/web/hwlab-cloud-web/src/config/workbench-runtime-policy.ts @@ -14,7 +14,6 @@ export interface WorkbenchRuntimePolicy { sessionListPageLimit: number; workbenchDetailReadConcurrency: number; workbenchReadFailureCooldownMs: number; - workbenchTurnStatusMinRefreshMs: number; workbenchTraceEventsMinRefreshMs: number; workbenchSessionDetailMinRefreshMs: number; workbenchSessionMessagesMinRefreshMs: number; @@ -22,8 +21,6 @@ export interface WorkbenchRuntimePolicy { workbenchTraceMessagesWindowLimit: number; workbenchRealtimeSessionMessagesMinRefreshMs: number; workbenchTraceEventsTimeoutMs: number; - sessionListRealtimeRefreshDelayMs: number; - sessionListTerminalRefreshDelayMs: number; sessionListMinRefreshIntervalMs: number; workbenchRealtimeErrorSyncReplayMinMs: number; workbenchRealtimeFlushMaxItemsPerChunk: number; @@ -44,7 +41,6 @@ const DEFAULT_WORKBENCH_RUNTIME_POLICY: WorkbenchRuntimePolicy = Object.freeze({ sessionListPageLimit: 20, workbenchDetailReadConcurrency: 3, workbenchReadFailureCooldownMs: 5_000, - workbenchTurnStatusMinRefreshMs: 2_000, workbenchTraceEventsMinRefreshMs: 4_000, workbenchSessionDetailMinRefreshMs: 5_000, workbenchSessionMessagesMinRefreshMs: 5_000, @@ -52,8 +48,6 @@ const DEFAULT_WORKBENCH_RUNTIME_POLICY: WorkbenchRuntimePolicy = Object.freeze({ workbenchTraceMessagesWindowLimit: 8, workbenchRealtimeSessionMessagesMinRefreshMs: 1_000, workbenchTraceEventsTimeoutMs: 5_000, - sessionListRealtimeRefreshDelayMs: 5_000, - sessionListTerminalRefreshDelayMs: 1_500, sessionListMinRefreshIntervalMs: 15_000, workbenchRealtimeErrorSyncReplayMinMs: 2_000, workbenchRealtimeFlushMaxItemsPerChunk: 4, @@ -76,7 +70,6 @@ export function workbenchRuntimePolicy(input: unknown = runtimePolicyConfig()): sessionListPageLimit: positiveInteger(source.sessionListPageLimit, DEFAULT_WORKBENCH_RUNTIME_POLICY.sessionListPageLimit), workbenchDetailReadConcurrency: positiveInteger(source.workbenchDetailReadConcurrency ?? source.workbenchReadHydrationConcurrency, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchDetailReadConcurrency), workbenchReadFailureCooldownMs: positiveNumber(source.workbenchReadFailureCooldownMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchReadFailureCooldownMs), - workbenchTurnStatusMinRefreshMs: nonNegativeNumber(source.workbenchTurnStatusMinRefreshMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchTurnStatusMinRefreshMs), workbenchTraceEventsMinRefreshMs: nonNegativeNumber(source.workbenchTraceEventsMinRefreshMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchTraceEventsMinRefreshMs), workbenchSessionDetailMinRefreshMs: nonNegativeNumber(source.workbenchSessionDetailMinRefreshMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchSessionDetailMinRefreshMs), workbenchSessionMessagesMinRefreshMs: nonNegativeNumber(source.workbenchSessionMessagesMinRefreshMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchSessionMessagesMinRefreshMs), @@ -84,8 +77,6 @@ export function workbenchRuntimePolicy(input: unknown = runtimePolicyConfig()): workbenchTraceMessagesWindowLimit: positiveInteger(source.workbenchTraceMessagesWindowLimit, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchTraceMessagesWindowLimit), workbenchRealtimeSessionMessagesMinRefreshMs: nonNegativeNumber(source.workbenchRealtimeSessionMessagesMinRefreshMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchRealtimeSessionMessagesMinRefreshMs), workbenchTraceEventsTimeoutMs: positiveNumber(source.workbenchTraceEventsTimeoutMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchTraceEventsTimeoutMs), - sessionListRealtimeRefreshDelayMs: nonNegativeNumber(source.sessionListRealtimeRefreshDelayMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.sessionListRealtimeRefreshDelayMs), - sessionListTerminalRefreshDelayMs: nonNegativeNumber(source.sessionListTerminalRefreshDelayMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.sessionListTerminalRefreshDelayMs), sessionListMinRefreshIntervalMs: nonNegativeNumber(source.sessionListMinRefreshIntervalMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.sessionListMinRefreshIntervalMs), workbenchRealtimeErrorSyncReplayMinMs: nonNegativeNumber(source.workbenchRealtimeErrorSyncReplayMinMs ?? source.workbenchRealtimeErrorGapFillMinMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchRealtimeErrorSyncReplayMinMs), workbenchRealtimeFlushMaxItemsPerChunk: positiveInteger(source.workbenchRealtimeFlushMaxItemsPerChunk, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchRealtimeFlushMaxItemsPerChunk), diff --git a/web/hwlab-cloud-web/src/stores/workbench-colada-keys.ts b/web/hwlab-cloud-web/src/stores/workbench-colada-keys.ts index 7767983f..2714e3a3 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-colada-keys.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-colada-keys.ts @@ -31,8 +31,6 @@ export const workbenchColadaKeys = { historySessionMessagesRoot: (): EntryKey => ["workbench", "history", "session-messages"], historySessionMessages: (sessionId: string, input: WorkbenchSessionMessagesKeyInput = {}): EntryKey => ["workbench", "history", "session-messages", sessionId, normalizeKeyObject({ cursor: input.cursor ?? null, limit: finiteNumber(input.limit) })], detailRoot: (): EntryKey => ["workbench", "detail"], - detailTurnRoot: (): EntryKey => ["workbench", "detail", "turn"], - detailTurn: (traceId: string): EntryKey => ["workbench", "detail", "turn", traceId], detailTraceEventsRoot: (): EntryKey => ["workbench", "detail", "trace-events"], detailTraceEvents: (traceId: string, input: WorkbenchTraceEventsKeyInput = {}): EntryKey => ["workbench", "detail", "trace-events", traceId, normalizeKeyObject({ afterProjectedSeq: finiteNumber(input.afterProjectedSeq), limit: finiteNumber(input.limit) })], sessionsRoot: (): EntryKey => workbenchColadaKeys.snapshotSessionsRoot(), @@ -41,8 +39,6 @@ export const workbenchColadaKeys = { session: (sessionId: string): EntryKey => workbenchColadaKeys.snapshotSession(sessionId), sessionMessagesRoot: (): EntryKey => workbenchColadaKeys.historySessionMessagesRoot(), sessionMessages: (sessionId: string, input: WorkbenchSessionMessagesKeyInput = {}): EntryKey => workbenchColadaKeys.historySessionMessages(sessionId, input), - turnRoot: (): EntryKey => workbenchColadaKeys.detailTurnRoot(), - turn: (traceId: string): EntryKey => workbenchColadaKeys.detailTurn(traceId), traceEventsRoot: (): EntryKey => workbenchColadaKeys.detailTraceEventsRoot(), traceEvents: (traceId: string, input: WorkbenchTraceEventsKeyInput = {}): EntryKey => workbenchColadaKeys.detailTraceEvents(traceId, input), providerProfiles: (): EntryKey => ["workbench", "provider-profiles"], diff --git a/web/hwlab-cloud-web/src/stores/workbench-colada-queries.ts b/web/hwlab-cloud-web/src/stores/workbench-colada-queries.ts index 5b992c37..97844916 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-colada-queries.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-colada-queries.ts @@ -3,7 +3,6 @@ import { useQueryCache, type EntryKey, type QueryCache } from "@pinia/colada"; import { api } from "@/api"; -import type { ApiRequestOptions } from "@/api/client"; import type { AgentChatResultResponse, ApiResult } from "@/types"; import type { SessionListOptions, SessionMessageOptions, WorkbenchMessagePageResponse, WorkbenchSessionDetailResponse, WorkbenchSessionListResponse, WorkbenchTraceRequestOptions } from "@/api/workbench"; import { workbenchColadaKeys } from "./workbench-colada-keys"; @@ -13,14 +12,9 @@ interface QueryRunOptions { minIntervalMs?: number | null; } -export interface WorkbenchTurnQueryOptions extends QueryRunOptions { - timeoutMs?: number; - activityRef?: ApiRequestOptions["activityRef"]; -} - export interface WorkbenchTraceEventsQueryOptions extends QueryRunOptions, WorkbenchTraceRequestOptions { timeoutMs?: number; - activityRef?: ApiRequestOptions["activityRef"]; + activityRef?: import("@/api/client").ApiRequestOptions["activityRef"]; } export interface WorkbenchSessionDetailQueryOptions extends QueryRunOptions { @@ -33,13 +27,10 @@ export interface WorkbenchColadaQueries { fetchSessions: (options?: SessionListOptions & QueryRunOptions) => Promise>; fetchSession: (sessionId: string, options?: WorkbenchSessionDetailQueryOptions) => Promise>; fetchSessionMessages: (sessionId: string, options?: SessionMessageOptions & QueryRunOptions) => Promise>; - fetchTurn: (traceId: string, options?: WorkbenchTurnQueryOptions) => Promise>; fetchTraceEvents: (traceId: string, options?: WorkbenchTraceEventsQueryOptions) => Promise>; fetchProviderProfiles: (options?: QueryRunOptions) => Promise>; - invalidateSessionList: () => Promise; invalidateSession: (sessionId: string | null | undefined) => Promise; invalidateSessionMessages: (sessionId: string | null | undefined) => Promise; - invalidateTurn: (traceId: string | null | undefined) => Promise; invalidateTraceEvents: (traceId: string | null | undefined) => Promise; invalidateTraceScope: (input: { sessionId?: string | null; traceId?: string | null }) => Promise; } @@ -52,13 +43,10 @@ export function useWorkbenchColadaQueries(): WorkbenchColadaQueries { fetchSessions: (options = {}) => runWorkbenchQuery(queryCache, workbenchColadaKeys.sessions(options), () => api.workbench.sessions(options), options), fetchSession: (sessionId, options = {}) => runWorkbenchQuery(queryCache, workbenchColadaKeys.session(sessionId), () => api.workbench.session(sessionId, { timeoutMs: options.timeoutMs ?? null, includeMessages: false }), options), fetchSessionMessages: (sessionId, options = {}) => runWorkbenchQuery(queryCache, workbenchColadaKeys.sessionMessages(sessionId, options), () => api.workbench.sessionMessages(sessionId, options), options), - fetchTurn: (traceId, options = {}) => runWorkbenchQuery(queryCache, workbenchColadaKeys.turn(traceId), () => api.workbench.turn(traceId, options.timeoutMs ?? 8000, options.activityRef), options), fetchTraceEvents: (traceId, options = {}) => runWorkbenchQuery(queryCache, workbenchColadaKeys.traceEvents(traceId, options), () => api.workbench.traceEvents(traceId, options.timeoutMs ?? 8000, options.activityRef, { afterProjectedSeq: options.afterProjectedSeq, limit: options.limit }), options), fetchProviderProfiles: (options = {}) => runWorkbenchQuery(queryCache, workbenchColadaKeys.providerProfiles(), () => api.providerProfiles.catalog(), options), - invalidateSessionList: () => queryCache.invalidateQueries({ key: workbenchColadaKeys.sessionsRoot() }), invalidateSession: (sessionId) => sessionId ? queryCache.invalidateQueries({ key: workbenchColadaKeys.session(sessionId) }) : Promise.resolve(), invalidateSessionMessages: (sessionId) => sessionId ? queryCache.invalidateQueries({ key: workbenchColadaKeys.sessionMessagesRoot().concat(sessionId) }) : Promise.resolve(), - invalidateTurn: (traceId) => traceId ? queryCache.invalidateQueries({ key: workbenchColadaKeys.turn(traceId) }) : Promise.resolve(), invalidateTraceEvents: (traceId) => traceId ? queryCache.invalidateQueries({ key: workbenchColadaKeys.traceEventsRoot().concat(traceId) }) : Promise.resolve(), invalidateTraceScope: async (input) => Promise.all([ queryCache.invalidateQueries({ key: workbenchColadaKeys.snapshotRoot() }), diff --git a/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts b/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts index df1eea60..aa5498c2 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts @@ -137,7 +137,8 @@ test("workbench active terminal paths seal final response from turn authority", const projectBlock = source.slice(source.indexOf("function projectTurnAuthorityToMessages"), source.indexOf("async function submitMessage")); const realtimeTurnBlock = source.slice(source.indexOf("function applyRealtimeTurnSnapshot"), source.indexOf("function scheduleRealtimeTurnProjection")); const realtimeTurnProjectionBlock = source.slice(source.indexOf("function flushRealtimeTurnProjection"), source.indexOf("function installRealtimeVisibilityHandler")); - const completeBlock = source.slice(source.indexOf("function completeTrace"), source.indexOf("async function hydrateTerminalMessageDiagnostics")); + const completeBlock = source.slice(source.indexOf("function completeTrace"), source.indexOf("function applyTerminalResultDiagnostics")); + const terminalDiagnosticsBlock = source.slice(source.indexOf("function applyTerminalResultDiagnostics"), source.indexOf("function applyRealtimeProjectionError")); const crossTabSyncBlock = source.slice(source.indexOf("async function refreshCrossTabSessionFromSyncReplay"), source.indexOf("function completeTrace")); const sessionDetailReadBlock = source.slice(source.indexOf("function fetchSessionDetailPage"), source.indexOf("function sessionMessageProjectionWindowLimit")); const traceDetailReadBlock = source.slice(source.indexOf("async function readTraceEventsForExplicitDetailPages"), source.indexOf("async function fetchTraceDetailEventsPage")); @@ -147,7 +148,7 @@ test("workbench active terminal paths seal final response from turn authority", assert.match(projectBlock, /terminalAuthorityMessageFromTurnResult\(result\)/u); assert.match(projectBlock, /type:\s*"message\.upsert"/u); assert.match(projectBlock, /\.\.\.terminalPatch/u); - assert.match(source, /createKeyedSingleflight>/u); + assert.match(source, /const traceDetailReadSingleflight = createKeyedSingleflight\(\)/u); assert.match(source, /traceDetailReadSingleflight\.run/u); assert.doesNotMatch(source, new RegExp(["refresh", "RealtimeSessionMessages"].join(""), "u")); assert.match(realtimeTurnBlock, /rememberTurnStatus\(traceId, result\)[\s\S]*scheduleRealtimeTurnProjection\(\{ traceId, result, terminalTurn \}\)/u); @@ -156,7 +157,8 @@ test("workbench active terminal paths seal final response from turn authority", assert.match(realtimeTurnProjectionBlock, /workbench_realtime_turn_projection_budget/u); assert.doesNotMatch(realtimeTurnProjectionBlock, /refreshWorkbenchSyncReplay|refreshTerminalTraceFromSyncReplay/u); assert.match(completeBlock, /projectTurnAuthorityToMessages\(traceId, result, "complete-trace"\)/u); - assert.match(completeBlock, /options\.forceRead[\s\S]*refreshMessageProjectionForTrace\(ownerSessionId, traceId, \{ force: true \}\)/u); + assert.doesNotMatch(completeBlock, /forceRead|refreshMessageProjectionForTrace|fetchWorkbenchTurnStatus|refreshTurnStatusByTraceId/u); + assert.doesNotMatch(terminalDiagnosticsBlock, /readTraceEventsForMessages|fetchSessionMessagesPage|fetchWorkbenchTurnStatus/u); assert.doesNotMatch(completeBlock, /scheduleActiveTraceSyncReplay|refreshWorkbenchSyncReplay/u); assert.match(source.slice(source.indexOf("function handleWorkbenchProjectionSignal"), source.indexOf("function applyTraceSnapshot")), /restartRealtime\("cross-tab-session-projection", true\)/u); assert.doesNotMatch(source, /scheduleActiveTraceSyncReplay|refreshActiveTraceFromSyncReplay/u); @@ -169,6 +171,8 @@ test("workbench active terminal paths seal final response from turn authority", 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.match(loadBlock, /fetchSessionMessagesPage\(normalizedRequestId, \{ limit: messageLimit, reason: "load-session:messages", force: true \}\)/u); + assert.doesNotMatch(loadBlock, /messages:\s*\[\]/u); assert.doesNotMatch(loadBlock, /limit:\s*100/u); assert.match(restoreSealBlock, /workbench_restored_turn_auto_read_disabled/u); assert.doesNotMatch(restoreSealBlock, /fetchWorkbenchTurnStatus|Promise\.all/u); @@ -181,7 +185,7 @@ test("workbench active SSE path only recovers by reconnecting the events cursor" source.slice(source.indexOf("async function submitMessage"), source.indexOf("async function cancelAgentMessage")), source.slice(source.indexOf("function reattachTrace"), source.indexOf("function restartRealtime")), source.slice(source.indexOf("function flushRealtimeTurnProjection"), source.indexOf("function installRealtimeVisibilityHandler")), - source.slice(source.indexOf("function completeTrace"), source.indexOf("async function hydrateTerminalMessageDiagnostics")), + source.slice(source.indexOf("function completeTrace"), source.indexOf("function applyTerminalResultDiagnostics")), source.slice(source.indexOf("async function sealRestoredActiveTurnMessages"), source.indexOf("function reattachRestoredActiveTrace")) ].join("\n"); diff --git a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts index 7a21db54..d319205d 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts @@ -1,13 +1,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; -import type { ChatMessage } from "@/types"; -import { projectionMergeCommitSummary, sessionDetailAutoReadDecision, traceEventsAutoReadDecision, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey, workbenchTraceEventsReadKey } from "./workbench-session-messages-read-budget"; -import { selectActiveTurnStatusRefreshTraceIds } from "./workbench-session"; - -function message(id: string): ChatMessage { - return { id, role: "agent", text: "", status: "running", createdAt: "2026-07-02T00:00:00.000Z" } as ChatMessage; -} +import { traceEventsAutoReadDecision, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey, workbenchTraceEventsReadKey } from "./workbench-session-messages-read-budget"; test("session messages singleflight key separates limit and force class", () => { const limit8 = workbenchSessionMessagesReadKey({ sessionId: "ses_key", limit: 8, force: false }); @@ -18,7 +12,6 @@ test("session messages singleflight key separates limit and force class", () => assert.notEqual(limit20, force20); 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 }); @@ -26,7 +19,6 @@ test("session detail singleflight key separates force freshness class", () => { assert.notEqual(fresh, forced); assert.equal(fresh, workbenchSessionDetailReadKey({ sessionId: "ses_key" })); }); - test("trace events auto read skips initial and completed equivalent ranges", () => { const key = workbenchTraceEventsReadKey({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50 }); @@ -35,25 +27,3 @@ test("trace events auto read skips initial and completed equivalent ranges", () assert.equal(traceEventsAutoReadDecision({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50, cachedRange: { traceId: "trc_range", afterProjectedSeq: 12, limit: 50, nextProjectedSeq: 40, hasMore: true } }).read, true); assert.equal(key, workbenchTraceEventsReadKey({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50 })); }); - -test("session detail auto read defers trace scoped realtime refresh to turn authority", () => { - assert.deepEqual(sessionDetailAutoReadDecision({ traceId: "trc_turn", nowMs: 10_000, minIntervalMs: 5_000 }), { read: false, reason: "turn-authority-first" }); - assert.deepEqual(sessionDetailAutoReadDecision({ nowMs: 10_000, lastReadAtMs: 9_000, minIntervalMs: 5_000 }), { read: false, reason: "min-interval" }); - assert.deepEqual(sessionDetailAutoReadDecision({ nowMs: 10_000, lastReadAtMs: 1_000, minIntervalMs: 5_000 }), { read: true, reason: "metadata-only" }); -}); - -test("projection merge summary reports no write without blocking follow-up hydrate", () => { - const existing = [{ ...message("msg_same"), traceId: "trc_current", sessionId: "ses_same" }]; - const summary = projectionMergeCommitSummary(existing, existing); - const hydrateTraceIds = selectActiveTurnStatusRefreshTraceIds({ messages: existing, currentRequestTraceId: "trc_current", limit: 1 }); - - assert.deepEqual(summary, { changed: false, changedCount: 0 }); - assert.deepEqual(hydrateTraceIds, ["trc_current"]); -}); - -test("projection merge summary reports changed message references", () => { - const existing = [message("msg_old")]; - const merged = [message("msg_new")]; - - assert.deepEqual(projectionMergeCommitSummary(existing, merged), { changed: true, changedCount: 1 }); -}); diff --git a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts index 125593b8..abf2adfe 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts @@ -1,7 +1,6 @@ // SPEC: pikasTech/HWLAB#2356 Workbench bounded request storm follow-up. // Responsibility: Pure budgeting helpers for Workbench detail read de-duplication and projection merge writes. -import type { ChatMessage } from "@/types"; import { firstNonEmptyString } from "@/utils"; import { composeWorkbenchScopedKey } from "@/utils/workbench-key"; @@ -16,11 +15,6 @@ export interface SessionDetailReadKeyInput { force?: boolean | null; } -export interface ProjectionMergeCommitSummary { - changed: boolean; - changedCount: number; -} - export interface TraceEventsReadRangeInput { traceId: string | null | undefined; afterProjectedSeq: number | null | undefined; @@ -38,15 +32,6 @@ export interface TraceEventsAutoReadDecisionInput extends TraceEventsReadRangeIn cachedRange?: TraceEventsReadRangeRecord | null; } -export interface SessionDetailAutoReadDecisionInput { - force?: boolean | null; - traceId?: string | null; - terminalBodyVisible?: boolean | null; - lastReadAtMs?: number | null; - nowMs: number; - minIntervalMs: number; -} - export interface ReadDetailDecision { read: boolean; reason: string; @@ -89,25 +74,6 @@ export function traceEventsAutoReadDecision(input: TraceEventsAutoReadDecisionIn return { read: true, reason: "delta-range" }; } -export function sessionDetailAutoReadDecision(input: SessionDetailAutoReadDecisionInput): ReadDetailDecision { - if (input.force === true) return { read: true, reason: "force" }; - if (input.terminalBodyVisible === true) return { read: false, reason: "terminal-visible" }; - if (firstNonEmptyString(input.traceId)) return { read: false, reason: "turn-authority-first" }; - const lastReadAtMs = Number(input.lastReadAtMs); - const minIntervalMs = Math.max(0, Math.trunc(Number(input.minIntervalMs))); - if (Number.isFinite(lastReadAtMs) && input.nowMs - lastReadAtMs < minIntervalMs) return { read: false, reason: "min-interval" }; - return { read: true, reason: "metadata-only" }; -} - -export function projectionMergeCommitSummary(existing: ChatMessage[], merged: ChatMessage[]): ProjectionMergeCommitSummary { - let changedCount = Math.max(0, merged.length - existing.length); - const sharedLength = Math.min(existing.length, merged.length); - for (let index = 0; index < sharedLength; index += 1) { - if (existing[index] !== merged[index]) changedCount += 1; - } - return { changed: changedCount > 0, changedCount }; -} - function sessionMessagesReadLimitPart(value: number | null | undefined): number { const number = Number(value); return Number.isFinite(number) && number > 0 ? Math.trunc(number) : 1; diff --git a/web/hwlab-cloud-web/src/stores/workbench-session.test.ts b/web/hwlab-cloud-web/src/stores/workbench-session.test.ts index e24615f0..86d69f48 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session.test.ts @@ -2,7 +2,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; import type { ChatMessage, WorkbenchSessionRecord } from "../types"; -import { resolveCancelableAgentMessage, resolveComposerState, selectActiveTurnStatusRefreshTraceIds, sessionToSessionTab, shouldReadTerminalTraceDetail } from "./workbench-session"; +import { resolveCancelableAgentMessage, resolveComposerState, sessionToSessionTab, shouldReadTerminalTraceDetail } from "./workbench-session"; import type { TurnStatusAuthority } from "./workbench-session"; function agentMessage(input: Partial & Pick): ChatMessage { @@ -96,22 +96,6 @@ test("cancel target remains available for failed message until terminal body exi assert.equal(resolveCancelableAgentMessage({ messages: [sealed], targetTraceId: traceId, targetSessionId: sessionId, turnStatusAuthority: { [traceId]: turnStatus({ traceId, sessionId, status: "running", running: true, terminal: false }) } }), null); }); -test("turn status refresh selector keeps automatic hydrate bounded to the active trace", () => { - const sessionId = "ses_turn_hydrate_budget"; - const older = agentMessage({ id: "msg_older", traceId: "trc_older", sessionId, status: "running" }); - const current = agentMessage({ id: "msg_current", traceId: "trc_current", sessionId, status: "running" }); - - assert.deepEqual(selectActiveTurnStatusRefreshTraceIds({ messages: [older, current], currentRequestTraceId: current.traceId, limit: 1 }), [current.traceId]); -}); - -test("turn status refresh selector skips sealed terminal traces", () => { - const sessionId = "ses_turn_detail_sealed"; - const traceId = "trc_sealed"; - const sealed = agentMessage({ id: "msg_sealed", traceId, sessionId, status: "completed", text: "final" }); - - assert.deepEqual(selectActiveTurnStatusRefreshTraceIds({ messages: [sealed], currentRequestTraceId: traceId, turnStatusAuthority: { [traceId]: turnStatus({ traceId, sessionId, status: "completed", running: false, terminal: true }) }, limit: 1 }), []); -}); - test("terminal detail read skips sealed terminal trace authority", () => { const sessionId = "ses_terminal_detail_sealed"; const traceId = "trc_terminal_detail_sealed"; diff --git a/web/hwlab-cloud-web/src/stores/workbench-session.ts b/web/hwlab-cloud-web/src/stores/workbench-session.ts index f4c3deca..56a89311 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session.ts @@ -117,22 +117,6 @@ 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(); - 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); - if (traceIds.size >= limit) return [...traceIds]; - 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 shouldReadTerminalTraceDetail(input: { traceId?: string | null; messages: ChatMessage[]; turnStatusAuthority?: TurnStatusAuthorityMap }): boolean { const traceId = firstNonEmptyString(input.traceId); if (!traceId) return false; @@ -317,21 +301,6 @@ function latestMessageForTrace(messages: ChatMessage[], traceId: string): ChatMe 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 (messageIsSealedTerminal(message)) return false; - return true; -} - function resolveSessionTabStatus(session: WorkbenchSessionRecord, authority: SessionStatusAuthority | null | undefined, latestAgent: ChatMessage | null = latestAgentMessage(session.messages)): string { const latestTraceId = firstNonEmptyString(latestAgent?.traceId, latestAgent?.runnerTrace?.traceId); const authorityTraceId = firstNonEmptyString(authority?.lastTraceId); diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index a4ba37fd..49edf41e 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -17,14 +17,14 @@ import type { AgentChatResponse, AgentChatResultResponse, AgentRunProvenance, Ap import { firstNonEmptyString, nextProtocolId, normalizeWorkbenchSessionId, normalizeWorkbenchSessionRouteId } from "@/utils"; import { composeWorkbenchScopedKey, workbenchRealtimeScopeKey } 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, selectActiveTurnStatusRefreshTraceIds, shouldReadTerminalTraceDetail, 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, shouldReadTerminalTraceDetail, shouldShowSessionListLoading, sortSessionTabs, stableSessionList, type DraftEntry, type ProviderProfileOption, type TurnStatusAuthority } from "./workbench-session"; import { initialWorkbenchSessionIdFromLocation } from "./workbench-projection"; import { cleanupWorkbenchServerStateSessions, selectActiveMessages, selectActiveSession, selectSessionList, selectSessionStatusAuthority, selectTraceAuthorityById, selectTurnStatusAuthority, type WorkbenchServerAction } from "./workbench-server-state"; import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "./workbench-session-cache"; import { reduceWorkbenchRealtimeEvent, type WorkbenchRealtimeAction } from "./workbench-event-reducer"; import { projectRejectedWorkbenchAdmission } from "./workbench-admission-failure"; import { messageHasSealedTerminalResult, messageIsSealedTerminal, traceAuthorityIsSealed } from "./workbench-terminal-authority"; -import { boundedProjectionMessageLimit, mergeBoundedProjectionMessages, selectProjectionMessageWindow, traceProjectionIsTerminalSealed } from "./workbench-message-projection-budget"; +import { boundedProjectionMessageLimit, traceProjectionIsTerminalSealed } from "./workbench-message-projection-budget"; import { agentErrorDisplayText, agentErrorFromApiFailure, @@ -37,7 +37,6 @@ import { isTerminalMessageStatus, isTraceActiveStatus, messageHasTerminalResponse, - messageNeedsTerminalDiagnostics, messageNeedsTraceDetailRead, messageStatusPatchForTerminalMerge, terminalAuthorityMessageFromTurnResult, @@ -46,7 +45,6 @@ import { messageTimingPatch, messageTimingPatchForMerge, messageTimingPatchFromProjection, - messageTimingSealPatchForProjectionMerge, mergeTerminalResultTrace, nonBlockingProjection, normalizeAgentError, @@ -63,7 +61,6 @@ import { terminalMessageTimingPatchForNormalize, turnResultIsTerminalForMerge, turnResultStatusForMerge, - traceHasTerminalResponse, traceHasEvents, traceResultHasTerminalEvidence, } from "./workbench-message-projection-runtime"; @@ -71,7 +68,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, traceEventsAutoReadDecision, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey, workbenchTraceEventsReadKey, type TraceEventsReadRangeRecord } from "./workbench-session-messages-read-budget"; +import { traceEventsAutoReadDecision, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey, workbenchTraceEventsReadKey, type TraceEventsReadRangeRecord } from "./workbench-session-messages-read-budget"; import { realtimeSnapshotToTraceSnapshot, terminalSealResultWithoutTraceEvents, traceDetailProjectedSeq, traceNextProjectedSeq } from "./workbench-trace-detail"; import { appendRawHwlabIngressFrame, createRawHwlabIngressState } from "./workbench-raw-hwlab-ingress"; @@ -121,7 +118,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { const workbenchColadaReducer = useWorkbenchColadaReducer(); const workbenchColadaQueries = useWorkbenchColadaQueries(); const workbenchColadaMutations = useWorkbenchColadaMutations(); - const turnStatusReadSingleflight = createKeyedSingleflight>(); const traceDetailReadSingleflight = createKeyedSingleflight(); const sessionMessagesReadSingleflight = createKeyedSingleflight>(); const sessionDetailReadSingleflight = createKeyedSingleflight>(); @@ -386,15 +382,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { if (sessions.value.length === 0) error.value = response.error ?? "session list unavailable"; } - function scheduleSessionListRefresh(includeSessionId: string | null | undefined = activeSessionId.value, delayMs = runtimePolicy.sessionListRealtimeRefreshDelayMs): void { - void includeSessionId; - if (typeof window === "undefined") { - void workbenchColadaQueries.invalidateSessionList(); - return; - } - window.setTimeout(() => { void workbenchColadaQueries.invalidateSessionList(); }, delayMs); - } - async function loadMoreSessions(): Promise { if (sessionListLoadingMore.value || !sessionListHasMore.value) return; const cursor = sessionListNextCursor.value; @@ -516,11 +503,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { void source; } - function fetchWorkbenchTurnStatus(traceId: string, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId), options: { force?: boolean } = {}): Promise> { - const activitySource = useActivityTimeout ? () => activityRef.value : null; - return turnStatusReadSingleflight.run(traceId, () => workbenchColadaQueries.fetchTurn(traceId, { timeoutMs: 8000, activityRef: activitySource, minIntervalMs: runtimePolicy.workbenchTurnStatusMinRefreshMs, force: options.force }), { reason: options.force ? "force-turn-status" : "turn-status" }); - } - function fetchWorkbenchTraceEvents(traceId: string, afterProjectedSeq: number, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId), options: { force?: boolean } = {}): Promise> { const activitySource = useActivityTimeout ? () => activityRef.value : null; return workbenchColadaQueries.fetchTraceEvents(traceId, { timeoutMs: runtimePolicy.workbenchTraceEventsTimeoutMs, activityRef: activitySource, afterProjectedSeq, limit: runtimePolicy.traceDetailPageLimit, minIntervalMs: runtimePolicy.workbenchTraceEventsMinRefreshMs, force: options.force }); @@ -537,25 +519,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { return isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status); } - async function refreshMessageProjectionForTrace(sessionId: string | null | undefined, traceId: string, options: { force?: boolean } = {}): Promise { - const id = normalizeWorkbenchSessionId(sessionId); - if (!id) return; - const existing = serverState.value.messagesBySessionId[id] ?? []; - if (traceProjectionIsTerminalSealed(traceId, existing)) return; - const response = await fetchSessionMessagesPage(id, { limit: traceMessageProjectionWindowLimit(), reason: `trace-message-page:${traceId}`, force: options.force }); - if (!response.ok || !response.data) { - if (shouldSuppressTransientWorkbenchReadFailure(response)) return; - if (traceHasTerminalResponse(traceId, messages.value)) return; - applyProjectionDiagnostic(traceId, projectionDiagnosticFromApiFailure(response, { code: "message_projection_refresh_failed", message: response.error ?? "消息投影刷新失败,主消息正文保持上一份 canonical projection。", health: response.status === 0 ? "unavailable" : "degraded" })); - return; - } - const pageMessages = Array.isArray(response.data.messages) ? response.data.messages.map((message) => normalizeChatMessage(message as ChatMessage)) : []; - const merged = mergeMessageProjectionPage(id, pageMessages, { traceId, limit: traceMessageProjectionWindowLimit() }); - commitMergedSessionMessages(id, pageMessages, merged, { traceId, limit: traceMessageProjectionWindowLimit(), reason: "trace-message-page" }); - await hydrateTurnStatusAuthority(merged, { traceId, limit: 1, reason: "trace-message-page" }); - if (!traceProjectionIsTerminalSealed(traceId, merged)) readTerminalTraceDetailGaps(merged, `trace-message-page:${traceId}`); - } - function fetchSessionMessagesPage(sessionId: string, options: SessionMessagesReadOptions): Promise> { const key = workbenchSessionMessagesReadKey({ sessionId, limit: options.limit, force: options.force }); return sessionMessagesReadSingleflight.run(key, () => workbenchColadaQueries.fetchSessionMessages(sessionId, { limit: options.limit, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs, force: options.force }), { reason: options.reason }); @@ -570,121 +533,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { return boundedProjectionMessageLimit(runtimePolicy.workbenchSessionMessagesWindowLimit, runtimePolicy.sessionListPageLimit); } - function traceMessageProjectionWindowLimit(): number { - return boundedProjectionMessageLimit(runtimePolicy.workbenchTraceMessagesWindowLimit, runtimePolicy.traceDetailAutoQueueLimit); - } - - function mergeMessageProjectionPage(sessionId: string, pageMessages: ChatMessage[], options: { traceId?: string | null; limit: number }): ChatMessage[] { - const existing = serverState.value.messagesBySessionId[sessionId] ?? []; - const windowMessages = selectProjectionMessageWindow(pageMessages, options); - const incoming = windowMessages.map((message) => mergeMessageProjectionMessage(message, existing)); - return mergeBoundedProjectionMessages(existing, incoming); - } - - function commitMergedSessionMessages(sessionId: string, incoming: ChatMessage[], merged: ChatMessage[], options: { traceId?: string | null; limit: number; reason: string }): void { - const existing = serverState.value.messagesBySessionId[sessionId] ?? []; - const summary = projectionMergeCommitSummary(existing, merged); - recordWorkbenchRuntimeDiagnostic({ - module: "workbench-message-projection", - sessionId, - traceId: firstNonEmptyString(options.traceId) ?? null, - outcome: "ok", - diagnostic: { - code: "workbench_message_projection_merge", - reason: options.reason, - source: "session-messages-page", - inputCount: incoming.length, - existingCount: existing.length, - mergedCount: merged.length, - changedCount: summary.changedCount, - limit: options.limit, - valuesRedacted: true - } - }); - if (summary.changed) rememberSessionMessages(sessionId, merged); - } - - function mergeMessageProjectionMessage(message: ChatMessage, existing: ChatMessage[]): ChatMessage { - const previous = findExistingProjectionMessage(message, existing); - const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId, previous?.traceId, previous?.runnerTrace?.traceId); - const authority = traceId ? traceAuthorityById.value[traceId] ?? null : null; - const preservedTrace = authority ?? previous?.runnerTrace ?? null; - const timingSealPatch = messageTimingSealPatchForProjectionMerge(message, previous); - if (!preservedTrace) return { ...message, ...timingSealPatch }; - const runnerTrace = message.runnerTrace ? mergeRunnerTrace(message.runnerTrace, preservedTrace) : preservedTrace; - return { ...message, ...timingSealPatch, runnerTrace }; - } - - function findExistingProjectionMessage(message: ChatMessage, existing: ChatMessage[]): ChatMessage | null { - const messageId = firstNonEmptyString(message.messageId, message.id); - if (messageId) { - const byId = existing.find((item) => firstNonEmptyString(item.messageId, item.id) === messageId); - if (byId) return byId; - } - const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId); - if (!traceId) return null; - return existing.find((item) => item.role === message.role && firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === traceId) ?? null; - } - - async function hydrateTurnStatusAuthority(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string; force?: boolean } = {}): Promise { - const traceIds = activeTurnStatusRefreshTraceIds(source, options); - if (options.force !== true) { - if (traceIds.length > 0) { - recordWorkbenchRuntimeDiagnostic({ - module: "workbench-turn-status", - traceId: options.traceId ?? traceIds[0] ?? null, - outcome: "ok", - diagnostic: { code: "workbench_turn_status_auto_read_disabled", reason: options.reason ?? "auto-turn-status-hydrate", source: "projection-sse-authority", requestedCount: traceIds.length, valuesRedacted: true } - }); - } - return; - } - for (const traceId of traceIds) { - await refreshTurnStatusByTraceId(traceId); - } - } - - function activeTurnStatusRefreshTraceIds(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string; force?: boolean } = {}): string[] { - const requestedTraceId = firstNonEmptyString(options.traceId, currentRequest.value?.traceId); - const traceIds = selectActiveTurnStatusRefreshTraceIds({ messages: source, currentRequestTraceId: requestedTraceId, turnStatusAuthority: turnStatusAuthority.value, limit: options.limit ?? 1 }); - recordWorkbenchRuntimeDiagnostic({ - module: "workbench-turn-status", - traceId: requestedTraceId ?? traceIds[0] ?? null, - outcome: "ok", - diagnostic: { - code: "workbench_turn_status_hydrate_budget", - reason: options.reason ?? "auto-turn-status-hydrate", - source: "session-message-projection", - requestedCount: traceIds.length, - limit: options.limit ?? 1, - valuesRedacted: true - } - }); - return traceIds; - } - - async function refreshTurnStatusByTraceId(traceId: string | null | undefined, options: { force?: boolean } = {}): Promise { - const id = firstNonEmptyString(traceId); - if (!id) return; - if (options.force !== true) { - recordWorkbenchRuntimeDiagnostic({ module: "workbench-turn-status", traceId: id, outcome: "ok", diagnostic: { code: "workbench_turn_status_auto_read_disabled", source: "projection-sse-authority", valuesRedacted: true } }); - return; - } - if (traceTerminalBodyIsVisible(id)) { - recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_turn_skip", source: "turn-status", valuesRedacted: true } }); - return; - } - const response = await fetchWorkbenchTurnStatus(id, shouldUseActivityTimeoutForTrace(id), { force: options.force }); - if (response.ok && response.data) { - applyTurnStatusSnapshot(id, response.data); - if (turnResultIsTerminalForMerge(response.data as AgentChatResultResponse)) completeTrace(id, response.data, { forceRead: options.force }); - return; - } - if (shouldSuppressTransientWorkbenchReadFailure(response)) return; - if (traceHasTerminalResponse(id, messages.value)) return; - applyProjectionDiagnostic(id, projectionDiagnosticFromApiFailure(response, { code: "turn_status_poll_failed", message: response.error ?? "状态更新失败,当前 turn 状态暂不可见。", health: response.status === 0 ? "unavailable" : "degraded" })); - } - function applyTurnStatusSnapshot(traceId: string, result: AgentChatResultResponse | TraceSnapshot): void { const ownerSessionId = traceOwnerSessionId(traceId, traceResultSessionId(result)); if (!ownerSessionId) return; @@ -816,7 +664,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { status: "running" }; void readTraceEventsForMessages(messages.value); - scheduleSessionListRefresh(sessionId, runtimePolicy.sessionListTerminalRefreshDelayMs); restartRealtime("steer"); return true; } @@ -838,7 +685,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { }; applyTurnStatusSnapshot(canonicalTraceId, response.data); publishWorkbenchProjectionSignal(sessionId, canonicalTraceId, "submit-admitted"); - scheduleSessionListRefresh(sessionId, runtimePolicy.sessionListTerminalRefreshDelayMs); if (turnResultIsTerminalForMerge(response.data as AgentChatResultResponse)) { completeTrace(canonicalTraceId, response.data as AgentChatResultResponse); return true; @@ -860,7 +706,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { if (message.status === "running") chatPending.value = false; currentRequest.value = null; void clearActiveTrace(traceId, "cancel-agent-message"); - scheduleSessionListRefresh(sessionId, runtimePolicy.sessionListTerminalRefreshDelayMs); } async function cancelRunningTrace(): Promise { @@ -1465,10 +1310,9 @@ export const useWorkbenchStore = defineStore("workbench", () => { return { ...message, ...messageTimingPatchForMerge(message, trace), ...messageStatusPatchForTerminalMerge(message, traceStatus, terminal), runnerTrace, error: clearCompletedDiagnostics ? null : error ?? message.error ?? null, projection, projectionStatus: projection?.projectionStatus ?? null, projectionHealth: projection?.projectionHealth ?? null, blocker: projection?.blocker ?? null, updatedAt: new Date().toISOString() }; })); markWorkbenchTraceProjected(traceId); - if (!traceProjectionIsTerminalSealed(traceId, serverState.value.messagesBySessionId[ownerSessionId] ?? [])) scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListRealtimeRefreshDelayMs); } - function completeTrace(traceId: string, result: AgentChatResultResponse, options: { forceRead?: boolean } = {}): void { + function completeTrace(traceId: string, result: AgentChatResultResponse): void { const authoritySessionId = traceResultSessionId(result); const ownerSessionId = traceOwnerSessionId(traceId, authoritySessionId); if (!ownerSessionId) return; @@ -1476,25 +1320,12 @@ export const useWorkbenchStore = defineStore("workbench", () => { projectTurnAuthorityToMessages(traceId, result, "complete-trace"); rememberTurnStatus(traceId, result); markWorkbenchTraceProjected(traceId); - const ownerMessages = serverState.value.messagesBySessionId[ownerSessionId] ?? []; - const terminalMessage = latestMessageForTrace(traceId, ownerMessages); - if (!traceProjectionIsTerminalSealed(traceId, ownerMessages)) { - if (options.forceRead) void refreshMessageProjectionForTrace(ownerSessionId, traceId, { force: true }); - } - if (options.forceRead && terminalMessage && !traceProjectionIsTerminalSealed(traceId, ownerMessages)) void readTraceEventsForMessage(terminalMessage, { force: true }); if (ownerSessionId === activeSessionId.value) { chatPending.value = false; currentRequest.value = null; void clearActiveTrace(traceId, "trace-terminal"); restartRealtime("trace-terminal"); } - if (!traceProjectionIsTerminalSealed(traceId, serverState.value.messagesBySessionId[ownerSessionId] ?? [])) scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs); - } - - async function hydrateTerminalMessageDiagnostics(): Promise { - const targets = messages.value.filter(messageNeedsTerminalDiagnostics).slice(-6); - void targets; - void readTraceEventsForMessages(messages.value); } function applyTerminalResultDiagnostics(traceId: string, result: AgentChatResultResponse): void { @@ -1516,8 +1347,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { const projection = clearCompletedDiagnostics ? nonBlockingProjection(resultProjection) : resultProjection ?? (terminal ? null : runnerTrace.projection ?? message.projection ?? null); return { ...message, ...messageTimingPatchForMerge(message, result), ...messageStatusPatchForTerminalMerge(message, resultStatus, terminal), title: normalizeWorkbenchMessageTitle(message.role, message.title), runnerTrace, error, projection, projectionStatus: projection?.projectionStatus ?? null, projectionHealth: projection?.projectionHealth ?? null, blocker: projection?.blocker ?? null, agentRun: agentRun ?? undefined, updatedAt: new Date().toISOString() }; })); - void readTraceEventsForMessages(serverState.value.messagesBySessionId[ownerSessionId] ?? []); - scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs); } function applyRealtimeProjectionError(event: WorkbenchRealtimeEvent): void { @@ -1527,7 +1356,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const code = firstNonEmptyString(errorRecord.code, "workbench_realtime_error") ?? "workbench_realtime_error"; const phase = firstNonEmptyString(event.phase, errorRecord.phase, "realtime") ?? "realtime"; const message = firstNonEmptyString(errorRecord.message, errorRecord.summary, "Workbench 实时状态更新异常,最新运行记录暂不可见。") ?? "Workbench 实时状态更新异常,最新运行记录暂不可见。"; - error.value = `${phase} · ${code} · ${message}`; + error.value = `${phase} · ${code} · ${message}`; return; } const projection = normalizeProjectionDiagnostic(event) ?? projectionDiagnosticFromFailure({ code: firstNonEmptyString(errorRecord.code, "workbench_realtime_error") ?? "workbench_realtime_error", message: firstNonEmptyString(errorRecord.message, errorRecord.summary, "Workbench 实时状态更新异常,最新运行记录暂不可见。") ?? "Workbench 实时状态更新异常,最新运行记录暂不可见。", health: "degraded", diagnostic: normalizeErrorDiagnostic(errorRecord.diagnostic, recordValue(event.projection)?.diagnostic, recordValue(event)?.diagnostic), apiError: normalizeApiErrorRecord(errorRecord, "Workbench 实时状态更新异常,最新运行记录暂不可见。") }); @@ -1558,7 +1387,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { chatPending.value = false; currentRequest.value = null; void clearActiveTrace(traceId, "trace-infrastructure-error"); - scheduleSessionListRefresh(selectedSessionId.value, runtimePolicy.sessionListTerminalRefreshDelayMs); } function projectRejectedAdmissionFailure(input: { traceId: string; message: string; error: ChatMessage["error"]; projection: ProjectionDiagnostic; submittedAt: string }): void { @@ -1700,8 +1528,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { rememberSessionDetail({ ...session, messages: projectedMessages, messageCount: session.messageCount ?? projectedMessages.length }); sessionsReady.value = true; currentRequest.value = null; - void hydrateTurnStatusAuthority(messages.value); - void hydrateTerminalMessageDiagnostics(); void readTraceEventsForMessages(messages.value); reattachRestoredActiveTrace(); restartRealtime("apply-selected-session"); @@ -1727,7 +1553,11 @@ export const useWorkbenchStore = defineStore("workbench", () => { const requestId = normalizeWorkbenchSessionRouteId(sessionId); if (!requestId) return null; const normalizedRequestId = normalizeWorkbenchSessionId(requestId); - const detail = await fetchSessionDetailPage(requestId, { reason: "load-session:detail", force: true }); + const messageLimit = sessionMessageProjectionWindowLimit(); + const [detail, messagePage] = await Promise.all([ + fetchSessionDetailPage(requestId, { reason: "load-session:detail", force: true }), + normalizedRequestId ? fetchSessionMessagesPage(normalizedRequestId, { limit: messageLimit, reason: "load-session:messages", force: true }) : Promise.resolve(null) + ]); if (!detail.ok) return null; const detailSession = sessionFromWorkbenchSession(detail.data?.session, { includeMessages: false }); const id = detailSession?.sessionId ?? normalizedRequestId ?? seed?.sessionId; @@ -1735,7 +1565,10 @@ export const useWorkbenchStore = defineStore("workbench", () => { const fallbackMessages = seed?.sessionId === id ? seed.messages ?? [] : []; const base = detailSession ? { ...detailSession, messages: fallbackMessages } : seed; if (!base) return null; - return { ...base, sessionId: id, messages: [], messageCount: base.messageCount ?? 0 }; + const hydratedMessages = messagePage?.ok && Array.isArray(messagePage.data?.messages) + ? messagePage.data.messages.map((message) => normalizeChatMessage(message as ChatMessage)) + : fallbackMessages; + return { ...base, sessionId: id, messages: hydratedMessages, messageCount: base.messageCount ?? hydratedMessages.length }; } async function sealRestoredActiveTurnMessages(source: ChatMessage[]): Promise { @@ -1756,7 +1589,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { } function reattachRestoredActiveTrace(): void { - void readTraceEventsForMessages(messages.value); const traceId = activeTraceIdFromMessages(messages.value, turnStatusAuthority.value); if (traceId) reattachTrace(traceId); } diff --git a/web/hwlab-cloud-web/src/types/global.d.ts b/web/hwlab-cloud-web/src/types/global.d.ts index 5106f04e..88fad71f 100644 --- a/web/hwlab-cloud-web/src/types/global.d.ts +++ b/web/hwlab-cloud-web/src/types/global.d.ts @@ -50,7 +50,6 @@ declare global { /** @deprecated Use workbenchDetailReadConcurrency. */ workbenchReadHydrationConcurrency?: number; workbenchReadFailureCooldownMs?: number; - workbenchTurnStatusMinRefreshMs?: number; workbenchTraceEventsMinRefreshMs?: number; workbenchSessionDetailMinRefreshMs?: number; workbenchSessionMessagesMinRefreshMs?: number; @@ -58,8 +57,6 @@ declare global { workbenchTraceMessagesWindowLimit?: number; workbenchRealtimeSessionMessagesMinRefreshMs?: number; workbenchTraceEventsTimeoutMs?: number; - sessionListRealtimeRefreshDelayMs?: number; - sessionListTerminalRefreshDelayMs?: number; sessionListMinRefreshIntervalMs?: number; workbenchRealtimeErrorSyncReplayMinMs?: number; /** @deprecated Use workbenchRealtimeErrorSyncReplayMinMs. */