From 855dc5adcc802468903c62c9c85d453c92dc758f Mon Sep 17 00:00:00 2001 From: root Date: Wed, 8 Jul 2026 14:33:14 +0200 Subject: [PATCH] feat: route workbench realtime recovery through sync replay --- .../workbench-realtime-runtime.test.ts | 57 ++++++-- .../src/api/workbench-events.ts | 62 +++++++- .../src/stores/workbench-event-reducer.ts | 12 +- .../stores/workbench-realtime-authority.ts | 133 ++++++++++++++++++ .../src/stores/workbench-realtime-plan.ts | 22 ++- web/hwlab-cloud-web/src/stores/workbench.ts | 53 +++++-- .../src/utils/workbench-stream-transport.ts | 13 +- 7 files changed, 303 insertions(+), 49 deletions(-) create mode 100644 web/hwlab-cloud-web/src/stores/workbench-realtime-authority.ts 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 160c522c..c7449fb1 100644 --- a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts @@ -19,6 +19,7 @@ import { messageDiagnosticView } from "../src/utils/workbench-error-runtime.ts"; import { WORKBENCH_TIMELINE_OPENCODE_PARITY, buildWorkbenchTimelineRows, normalizeWorkbenchTimelineMessages, workbenchTimelineSignature } from "../src/stores/workbench-timeline-model.ts"; import { reduceWorkbenchRealtimeEvent } from "../src/stores/workbench-event-reducer.ts"; import { planWorkbenchRealtimeApply, planWorkbenchRealtimeRecovery } from "../src/stores/workbench-realtime-plan.ts"; +import { WORKBENCH_REALTIME_AUTHORITY_VERSION, workbenchRealtimePrimaryAuthorityDecision, workbenchSyncReplayEvents } 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"; @@ -385,13 +386,16 @@ test("turn status refresh keeps local request trace priority", () => { }); test("realtime event reducer classifies SSE payloads before store side effects", () => { - const trace = reduceWorkbenchRealtimeEvent({ type: "trace.event", traceId: "trc_1", event: { traceId: "trc_1", label: "delta" }, snapshot: { traceId: "trc_1", status: "running" } }, "workbench.trace.event"); + 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"); assert.equal(trace.action.type, "trace.event"); assert.equal(trace.diagnostic.module, "workbench-event-reducer"); - const emptyMessage = reduceWorkbenchRealtimeEvent({ type: "message.snapshot" }, "workbench.message.snapshot"); - assert.deepEqual(emptyMessage.action, { type: "ignore", reason: "message.snapshot.missing-message" }); + const missingAuthority = reduceWorkbenchRealtimeEvent({ type: "message.snapshot", message: agentMessage({ status: "running", sessionId: "ses_1" }) }, "workbench.message.snapshot"); + assert.deepEqual(missingAuthority.action, { type: "ignore", reason: "workbench_realtime_authority_missing" }); + + const detailOnly = reduceWorkbenchRealtimeEvent(realtimeEvent({ type: "trace.event", traceId: "trc_1", detailProjection: true, authority: "trace-detail-only", event: { traceId: "trc_1", label: "detail" }, entity: { family: "traceEvents", id: "trc_1:2", version: 2, projectionRevision: "prj_1", authority: "trace-detail-only" } }), "workbench.trace.event"); + assert.deepEqual(detailOnly.action, { type: "ignore", reason: "workbench_realtime_detail_only_rejected" }); const error = reduceWorkbenchRealtimeEvent({ type: "error", traceId: "trc_2", error: { message: "offline" } }, "workbench.error"); assert.equal(error.action.type, "projection.error"); @@ -399,7 +403,7 @@ test("realtime event reducer classifies SSE payloads before store side effects", }); test("realtime apply planner turns reducer actions into store steps", () => { - const reduced = reduceWorkbenchRealtimeEvent({ type: "trace.event", traceId: "trc_1", event: { traceId: "trc_1", label: "delta" }, snapshot: { traceId: "trc_1", status: "running" } }, "workbench.trace.event"); + const reduced = 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"); const tracePlan = planWorkbenchRealtimeApply(reduced.action); assert.deepEqual(tracePlan.steps.map((step) => step.type), ["apply-trace-event"]); assert.equal(tracePlan.diagnostic.module, "workbench-realtime-plan"); @@ -411,25 +415,48 @@ test("realtime apply planner turns reducer actions into store steps", () => { assert.deepEqual(ignored.steps, []); }); -test("realtime recovery planner gates refresh steps by transport actions and authority", () => { - const recovery = recoveryEvent(["refresh-session-messages", "schedule-session-list", "refresh-turn-status", "hydrate-trace-events"]); +test("realtime recovery planner uses sync replay instead of legacy repair fan-out", () => { + const recovery = recoveryEvent(["sync-replay"], { outboxSeq: 42 }); const authorized = planWorkbenchRealtimeRecovery(recovery, { selectedSessionId: "ses_1", activeSessionId: "ses_1", fallbackTraceId: "trc_1", activeTraceAuthorized: true }); - assert.deepEqual(authorized.steps.map((step) => step.type), ["refresh-session-messages", "schedule-session-list", "refresh-turn-status", "hydrate-trace-events"]); - assert.deepEqual(authorized.steps.map((step) => step.authority), ["automatic-recovery", "automatic-recovery", "automatic-recovery", "automatic-recovery"]); - assert.deepEqual(authorized.steps.map((step) => step.force), [false, false, false, false]); + assert.deepEqual(authorized.steps.map((step) => step.type), ["sync-replay"]); + assert.deepEqual(authorized.steps.map((step) => step.authority), ["automatic-recovery"]); + assert.equal(authorized.steps[0]?.sinceOutboxSeq, 42); assert.equal(authorized.sessionId, "ses_1"); assert.equal(authorized.traceId, "trc_1"); const unauthorized = planWorkbenchRealtimeRecovery(recovery, { selectedSessionId: "ses_1", activeSessionId: "ses_1", fallbackTraceId: "trc_1", activeTraceAuthorized: false }); - assert.deepEqual(unauthorized.steps.map((step) => step.type), ["refresh-session-messages", "schedule-session-list"]); + assert.deepEqual(unauthorized.steps.map((step) => step.type), ["sync-replay"]); const inactive = planWorkbenchRealtimeRecovery(recovery, { selectedSessionId: "ses_1", activeSessionId: "ses_other", fallbackTraceId: "trc_1", activeTraceAuthorized: true }); - assert.deepEqual(inactive.steps.map((step) => step.type), ["schedule-session-list", "refresh-turn-status", "hydrate-trace-events"]); + assert.deepEqual(inactive.steps.map((step) => step.type), ["sync-replay"]); const terminalSealed = planWorkbenchRealtimeRecovery(recovery, { selectedSessionId: "ses_1", activeSessionId: "ses_1", fallbackTraceId: "trc_1", activeTraceAuthorized: true, terminalTraceSealed: true }); assert.deepEqual(terminalSealed.steps.map((step) => step.type), []); }); +test("realtime authority accepts events and sync replay through the same entity contract", () => { + const event = realtimeEvent({ type: "message.snapshot", sessionId: "ses_1", message: agentMessage({ id: "msg_1", sessionId: "ses_1", status: "completed", text: "done" }), entity: { family: "messages", id: "msg_1", version: 7, outboxSeq: 12, projectionRevision: "prj_7" } }); + const decision = workbenchRealtimePrimaryAuthorityDecision(event); + assert.equal(decision.accepted, true); + assert.equal(decision.entity?.family, "messages"); + assert.equal(decision.entity?.version, 7); + + const replay = workbenchSyncReplayEvents({ contractVersion: "workbench-sync-v1", realtimeAuthority: WORKBENCH_REALTIME_AUTHORITY_VERSION, events: [event], delta: [event] }); + assert.equal(replay.length, 1); + assert.equal(replay[0]?.entity?.id, "msg_1"); +}); + +test("realtime authority rejects trace detail-only and incomplete contract payloads", () => { + const detail = realtimeEvent({ type: "trace.event", traceId: "trc_1", detailProjection: true, authority: "trace-detail-only", event: { traceId: "trc_1" }, entity: { family: "traceEvents", id: "trc_1:1", version: 1, projectionRevision: "prj_1", authority: "trace-detail-only" } }); + assert.equal(workbenchRealtimePrimaryAuthorityDecision(detail).reason, "workbench_realtime_detail_only_rejected"); + + const missingEntity = realtimeEvent({ type: "message.snapshot", message: agentMessage({ status: "running", sessionId: "ses_1" }) }); + assert.equal(workbenchRealtimePrimaryAuthorityDecision(missingEntity).reason, "workbench_realtime_entity_missing"); + + const missingProjection = realtimeEvent({ type: "message.snapshot", message: agentMessage({ status: "running", sessionId: "ses_1" }), entity: { family: "messages", id: "msg_1", version: 1 } }); + assert.equal(workbenchRealtimePrimaryAuthorityDecision(missingProjection).reason, "workbench_realtime_projection_revision_missing"); +}); + test("health probe cache records ok and unavailable states", async () => { const cache = createWorkbenchHealthProbeCache({ cacheMs: 100 }); const ok = await cache.probe({ key: "workbench", fetcher: async () => ({ ready: true }), classify: (value) => value.ready ? "ok" : "degraded" }); @@ -505,7 +532,7 @@ function agentMessage(overrides: Partial): ChatMessage { } as ChatMessage; } -function recoveryEvent(actions: WorkbenchStreamTransportRecovery["actions"]): WorkbenchStreamTransportRecovery { +function recoveryEvent(actions: WorkbenchStreamTransportRecovery["actions"], cursor: { outboxSeq?: number | null; traceSeq?: number | null } = {}): WorkbenchStreamTransportRecovery { return { key: "workbench.realtime|ses_1|trc_1", sessionId: "ses_1", @@ -513,6 +540,12 @@ function recoveryEvent(actions: WorkbenchStreamTransportRecovery["actions"]): Wo tick: 1, reason: "eventsource-error", actions, + outboxSeq: cursor.outboxSeq ?? null, + traceSeq: cursor.traceSeq ?? null, diagnostic: { code: "workbench_sse_recovery", valuesRedacted: true } }; } + +function realtimeEvent(input: Record) { + return { realtimeAuthority: WORKBENCH_REALTIME_AUTHORITY_VERSION, contractVersion: "workbench-sync-v1", ...input }; +} diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts index f7075854..5d0d6431 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -1,7 +1,8 @@ -// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; 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. +// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; 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; PJ2026-010401080313 Workbench实时权威 draft-2026-07-08-p0-workbench-realtime-authority-v2. // Responsibility: Workbench SSE client. Realtime events accelerate UI projection; REST snapshots remain gap-fill authority. -import type { ChatMessage, ProjectionDiagnostic, TraceEvent } from "@/types"; +import { fetchJson, type ApiRequestOptions } from "@/api/client"; +import type { ApiResult, ChatMessage, ProjectionDiagnostic, TraceEvent } from "@/types"; import { createCoalescedEventQueue } from "@/utils/scheduler/coalesced-event-queue"; import { composeWorkbenchScopedKey, firstScopePart } from "@/utils/workbench-key"; import { recordWorkbenchRuntimeDiagnostic, recordWorkbenchSseLifecycle } from "@/utils/workbench-performance"; @@ -23,6 +24,8 @@ export interface WorkbenchRealtimeTraceSnapshot { export interface WorkbenchRealtimeEvent { type?: string; + contractVersion?: string | null; + realtimeAuthority?: string | null; status?: string; serverSentAt?: string | null; eventCreatedAt?: string | null; @@ -38,6 +41,41 @@ export interface WorkbenchRealtimeEvent { traceSeq?: number | null; outboxSeq?: number | null; cursor?: { traceSeq?: number | null; outboxSeq?: number | null; [key: string]: unknown }; + entity?: { + family?: string | null; + id?: string | null; + version?: number | string | null; + entityVersion?: number | string | null; + outboxSeq?: number | string | null; + traceSeq?: number | string | null; + projectionRevision?: string | null; + committedAt?: string | null; + serverCommittedAt?: string | null; + authority?: string | null; + detailProjection?: boolean | null; + [key: string]: unknown; + } | null; + authority?: string | null; + detailProjection?: boolean | null; + projectionRevision?: string | null; + [key: string]: unknown; +} + +export interface WorkbenchSyncReplayRequest { + sessionId?: string | null; + traceId?: string | null; + since?: number | null; +} + +export interface WorkbenchSyncReplayResponse { + contractVersion?: string | null; + realtimeAuthority?: string | null; + scope?: Record | null; + cursor?: Record | null; + events?: WorkbenchRealtimeEvent[]; + delta?: WorkbenchRealtimeEvent[]; + families?: Record | null; + authority?: Record | null; [key: string]: unknown; } @@ -146,6 +184,21 @@ export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): Wo }; } +export async function fetchWorkbenchSyncReplay(input: WorkbenchSyncReplayRequest, options: ApiRequestOptions = {}): Promise> { + return fetchJson(workbenchSyncReplayPath(input), { + ...options, + timeoutName: options.timeoutName ?? "workbench sync replay" + }); +} + +export function workbenchSyncReplayPath(input: WorkbenchSyncReplayRequest): string { + const params = new URLSearchParams(); + appendParam(params, "sessionId", input.sessionId); + appendParam(params, "traceId", input.traceId); + appendNumberParam(params, "since", input.since); + return `/v1/workbench/sync?${params.toString()}`; +} + function appendParam(params: URLSearchParams, key: string, value: string | null | undefined): void { const text = typeof value === "string" ? value.trim() : ""; if (text) params.set(key, text); @@ -177,6 +230,11 @@ function scheduleRealtimeFlushYield(flush: () => void, yieldMs: number | null | } export function realtimeCoalesceKey(event: WorkbenchRealtimeEvent, eventName: string): string | null { + const entity = event.entity; + const entityFamily = firstScopePart(entity?.family); + const entityId = firstScopePart(entity?.id); + const entityVersion = numericCursor(entity?.version ?? entity?.entityVersion); + if (entityFamily && entityId && entityVersion !== null) return composeWorkbenchScopedKey("workbench.realtime.entity", entityFamily, entityId, entityVersion); const sessionId = firstScopePart(event.sessionId, event.message?.sessionId, event.snapshot?.sessionId, event.event?.sessionId, event.turn?.sessionId); const traceId = firstScopePart(event.traceId, event.message?.traceId, event.snapshot?.traceId, event.event?.traceId, event.turn?.traceId); const outboxSeq = numericCursor(event.cursor?.outboxSeq ?? event.outboxSeq); diff --git a/web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts b/web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts index 87c113b6..147611cd 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts @@ -1,10 +1,11 @@ -// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection. +// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010401080313 Workbench实时权威 draft-2026-07-08-p0-workbench-realtime-authority-v2. // Responsibility: OpenCode-style Workbench realtime event reducer; store code consumes actions instead of branching on raw SSE events. // Mechanical source: /root/opencode/packages/app/src/context/global-sync/event-reducer.ts:21-48 global event classification, :93-270 session/message reducer action boundary. // Mechanical source: /root/opencode/packages/opencode/src/cli/cmd/run/session-data.ts:1-17 reducer side-effect boundary and replay notes. import type { WorkbenchRealtimeEvent } from "@/api/workbench-events"; import { firstNonEmptyString } from "@/utils"; +import { workbenchRealtimePrimaryAuthorityDecision } from "./workbench-realtime-authority"; export type WorkbenchRealtimeAction = | { type: "trace.snapshot"; traceId: string | null; snapshot: WorkbenchRealtimeEvent["snapshot"] } @@ -48,6 +49,8 @@ export function reduceWorkbenchRealtimeEvent(event: WorkbenchRealtimeEvent, even } function reduceRealtimeAction(event: WorkbenchRealtimeEvent, eventName: string): WorkbenchRealtimeAction { + const authority = primaryAuthority(event); + if (authority) return authority; switch (event.type) { case "trace.snapshot": return { type: "trace.snapshot", traceId: realtimeTraceId(event), snapshot: event.snapshot ?? null }; @@ -66,6 +69,13 @@ function reduceRealtimeAction(event: WorkbenchRealtimeEvent, eventName: string): return { type: "ignore", reason: firstNonEmptyString(event.type, eventName, "unsupported") ?? "unsupported" }; } +function primaryAuthority(event: WorkbenchRealtimeEvent): WorkbenchRealtimeAction | null { + if (!["trace.snapshot", "trace.event", "message.snapshot", "turn.snapshot"].includes(firstNonEmptyString(event.type) ?? "")) return null; + const decision = workbenchRealtimePrimaryAuthorityDecision(event); + if (decision.accepted) return null; + return { type: "ignore", reason: decision.reason }; +} + function realtimeTraceId(event: WorkbenchRealtimeEvent): string | null { return firstNonEmptyString(event.traceId, event.snapshot?.traceId, event.event?.traceId, event.message?.traceId) ?? null; } diff --git a/web/hwlab-cloud-web/src/stores/workbench-realtime-authority.ts b/web/hwlab-cloud-web/src/stores/workbench-realtime-authority.ts new file mode 100644 index 00000000..e68507f6 --- /dev/null +++ b/web/hwlab-cloud-web/src/stores/workbench-realtime-authority.ts @@ -0,0 +1,133 @@ +// SPEC: PJ2026-010401080313 Workbench实时权威 draft-2026-07-08-p0-workbench-realtime-authority-v2. +// Responsibility: Frontend authority gates for Workbench realtime events and sync replay batches. + +import type { WorkbenchRealtimeEvent, WorkbenchSyncReplayResponse } from "@/api/workbench-events"; + +export const WORKBENCH_REALTIME_AUTHORITY_VERSION = "workbench-realtime-authority-v2"; +export const WORKBENCH_SYNC_CONTRACT_VERSION = "workbench-sync-v1"; + +export interface WorkbenchRealtimeEntityAuthority { + family: string; + id: string; + version: number; + outboxSeq: number | null; + traceSeq: number | null; + projectionRevision: string | null; + committedAt: string | null; + authority: string | null; + detailProjection: boolean; +} + +export interface WorkbenchRealtimeAuthorityDecision { + accepted: boolean; + reason: string; + entity: WorkbenchRealtimeEntityAuthority | null; + diagnostic: { + module: "workbench-realtime-authority"; + code: string; + reason: string; + realtimeAuthority: string | null; + contractVersion: string | null; + entityFamily: string | null; + entityId: string | null; + entityVersion: number | null; + detailProjection: boolean; + valuesRedacted: true; + }; +} + +export function workbenchRealtimePrimaryAuthorityDecision(event: WorkbenchRealtimeEvent): WorkbenchRealtimeAuthorityDecision { + const entity = workbenchRealtimeEntityAuthority(event); + const detailProjection = workbenchRealtimeDetailOnly(event, entity); + const code = authorityFailureCode(event, entity, detailProjection); + const accepted = code === null; + const reason = accepted ? "accepted" : code; + return { + accepted, + reason, + entity, + diagnostic: { + module: "workbench-realtime-authority", + code: accepted ? "workbench_realtime_authority_accept" : code, + reason, + realtimeAuthority: stringValue(event.realtimeAuthority), + contractVersion: stringValue(event.contractVersion), + entityFamily: entity?.family ?? null, + entityId: entity?.id ?? null, + entityVersion: entity?.version ?? null, + detailProjection, + valuesRedacted: true + } + }; +} + +export function workbenchRealtimeEntityAuthority(event: WorkbenchRealtimeEvent): WorkbenchRealtimeEntityAuthority | null { + const source = recordValue(event.entity); + if (!source) return null; + const family = stringValue(source.family); + const id = stringValue(source.id ?? source.entityId); + const version = finiteNumber(source.version ?? source.entityVersion); + if (!family || !id || version === null) return null; + return { + family, + id, + version, + outboxSeq: finiteNumber(source.outboxSeq ?? event.cursor?.outboxSeq ?? event.outboxSeq), + traceSeq: finiteNumber(source.traceSeq ?? event.cursor?.traceSeq ?? event.traceSeq), + projectionRevision: stringValue(source.projectionRevision ?? event.projectionRevision), + committedAt: stringValue(source.committedAt ?? source.serverCommittedAt ?? event.eventCreatedAt ?? event.serverSentAt), + authority: stringValue(source.authority ?? event.authority), + detailProjection: source.detailProjection === true || event.detailProjection === true + }; +} + +export function workbenchSyncReplayEvents(payload: WorkbenchSyncReplayResponse | null | undefined): WorkbenchRealtimeEvent[] { + const events = arrayOfRecords(payload?.events); + const delta = arrayOfRecords(payload?.delta); + const seen = new Set(); + const output: WorkbenchRealtimeEvent[] = []; + for (const value of [...events, ...delta]) { + const event = value as WorkbenchRealtimeEvent; + const key = syncEventKey(event); + if (seen.has(key)) continue; + seen.add(key); + output.push(event); + } + return output; +} + +function authorityFailureCode(event: WorkbenchRealtimeEvent, entity: WorkbenchRealtimeEntityAuthority | null, detailProjection: boolean): string | null { + if (detailProjection) return "workbench_realtime_detail_only_rejected"; + if (stringValue(event.realtimeAuthority) !== WORKBENCH_REALTIME_AUTHORITY_VERSION) return "workbench_realtime_authority_missing"; + if (!entity) return "workbench_realtime_entity_missing"; + if (!entity.projectionRevision) return "workbench_realtime_projection_revision_missing"; + return null; +} + +function workbenchRealtimeDetailOnly(event: WorkbenchRealtimeEvent, entity: WorkbenchRealtimeEntityAuthority | null): boolean { + return event.detailProjection === true || entity?.detailProjection === true || stringValue(event.authority) === "trace-detail-only" || entity?.authority === "trace-detail-only"; +} + +function syncEventKey(event: WorkbenchRealtimeEvent): string { + const entity = workbenchRealtimeEntityAuthority(event); + if (entity) return [entity.family, entity.id, entity.version, entity.outboxSeq ?? "~"].join("|"); + return [event.type ?? "~", event.traceId ?? "~", event.sessionId ?? "~", event.cursor?.outboxSeq ?? event.outboxSeq ?? "~", event.cursor?.traceSeq ?? event.traceSeq ?? "~"].join("|"); +} + +function arrayOfRecords(value: unknown): Record[] { + return Array.isArray(value) ? value.filter((item): item is Record => Boolean(recordValue(item))) : []; +} + +function recordValue(value: unknown): Record | null { + return value && typeof value === "object" ? value as Record : null; +} + +function stringValue(value: unknown): string | null { + const text = typeof value === "string" ? value.trim() : ""; + return text || null; +} + +function finiteNumber(value: unknown): number | null { + const number = Number(value); + return Number.isFinite(number) && number >= 0 ? Math.trunc(number) : null; +} diff --git a/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts b/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts index abc52362..2041c489 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts @@ -1,4 +1,4 @@ -// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first. +// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; PJ2026-010401080313 Workbench实时权威 draft-2026-07-08-p0-workbench-realtime-authority-v2. // Responsibility: OpenCode-style realtime apply/recovery planning; the store executes plan steps without owning branching policy. // Mechanical source: /root/opencode/packages/app/src/context/global-sync/event-reducer.ts:93-270 session/message reducer action boundary. // Mechanical source: /root/opencode/packages/opencode/src/cli/cmd/run/session-data.ts:1-17 reducer side-effect boundary and replay notes. @@ -38,10 +38,7 @@ export interface WorkbenchRealtimeRecoveryContext { export type WorkbenchRealtimeRecoveryAuthority = "automatic-recovery"; export type WorkbenchRealtimeRecoveryStep = - | { type: "refresh-session-messages"; sessionId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority } - | { type: "schedule-session-list"; sessionId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority } - | { type: "refresh-turn-status"; traceId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority } - | { type: "hydrate-trace-events"; traceId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority }; + | { type: "sync-replay"; sessionId: string | null; traceId: string | null; sinceOutboxSeq: number | null; reason: string; authority: WorkbenchRealtimeRecoveryAuthority }; type WorkbenchRealtimeRecoveryStepInput = T extends WorkbenchRealtimeRecoveryStep ? Omit : never; @@ -75,14 +72,8 @@ export function planWorkbenchRealtimeApply(action: WorkbenchRealtimeAction): Wor export function planWorkbenchRealtimeRecovery(recovery: WorkbenchStreamTransportRecovery, context: WorkbenchRealtimeRecoveryContext): WorkbenchRealtimeRecoveryPlan { const sessionId = normalizeWorkbenchSessionId(recovery.sessionId ?? context.selectedSessionId); const traceId = firstNonEmptyString(recovery.traceId, context.fallbackTraceId); - const actions = new Set(recovery.actions); const steps: WorkbenchRealtimeRecoveryStep[] = []; - if (context.terminalTraceSealed !== true && actions.has("refresh-session-messages") && sessionId && sessionId === context.activeSessionId) steps.push(automaticRecoveryStep({ type: "refresh-session-messages", sessionId, reason: "realtime-error:messages" })); - if (context.terminalTraceSealed !== true && actions.has("schedule-session-list") && sessionId) steps.push(automaticRecoveryStep({ type: "schedule-session-list", sessionId, reason: "realtime-error:session-list" })); - if (context.terminalTraceSealed !== true && traceId && context.activeTraceAuthorized === true) { - if (actions.has("refresh-turn-status")) steps.push(automaticRecoveryStep({ type: "refresh-turn-status", traceId, reason: "realtime-error:turn-status" })); - if (actions.has("hydrate-trace-events")) steps.push(automaticRecoveryStep({ type: "hydrate-trace-events", traceId, reason: "realtime-error:trace-events" })); - } + if (context.terminalTraceSealed !== true && recovery.actions.includes("sync-replay") && (sessionId || traceId)) steps.push(automaticRecoveryStep({ type: "sync-replay", sessionId: sessionId ?? null, traceId: traceId ?? null, sinceOutboxSeq: finiteNumber(recovery.outboxSeq), reason: "realtime-error:sync-replay" })); return { sessionId: sessionId ?? null, traceId: traceId ?? null, @@ -98,7 +89,12 @@ export function planWorkbenchRealtimeRecovery(recovery: WorkbenchStreamTransport } function automaticRecoveryStep(step: WorkbenchRealtimeRecoveryStepInput): WorkbenchRealtimeRecoveryStep { - return { ...step, authority: "automatic-recovery", force: false } as WorkbenchRealtimeRecoveryStep; + return { ...step, authority: "automatic-recovery" } as WorkbenchRealtimeRecoveryStep; +} + +function finiteNumber(value: unknown): number | null { + const number = Number(value); + return Number.isFinite(number) && number >= 0 ? Math.trunc(number) : null; } function applySteps(action: WorkbenchRealtimeAction): WorkbenchRealtimeApplyStep[] { diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index eb15f594..1b9dfe41 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -1,9 +1,10 @@ -// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; 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; draft-2026-06-28-p0-d518-session-timeline-consistency; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. +// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; 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; draft-2026-06-28-p0-d518-session-timeline-consistency; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0; PJ2026-010401080313 Workbench实时权威 draft-2026-07-08-p0-workbench-realtime-authority-v2. // Responsibility: Session-first Workbench state orchestration for selection, turn admission, and trace lifecycle rendering. import { computed, nextTick, ref } from "vue"; import { defineStore } from "pinia"; import { api } from "@/api"; +import { fetchWorkbenchSyncReplay } from "@/api/workbench-events"; import { workbenchRuntimePolicy } from "@/config/workbench-runtime-policy"; import { createKeyedSingleflight } from "@/utils/scheduler/keyed-singleflight"; import { createWorkbenchHealthProbeCache } from "@/utils/workbench-health"; @@ -67,6 +68,7 @@ import { traceSnapshotError } from "./workbench-message-projection-runtime"; import { planWorkbenchRealtimeApply, planWorkbenchRealtimeRecovery, type WorkbenchRealtimeApplyStep, type WorkbenchRealtimeRecoveryStep } from "./workbench-realtime-plan"; +import { workbenchSyncReplayEvents } from "./workbench-realtime-authority"; import { useWorkbenchColadaMutations } from "./workbench-colada-mutations"; import { useWorkbenchColadaQueries } from "./workbench-colada-queries"; import { useWorkbenchColadaReducer } from "./workbench-colada-reducer"; @@ -1142,23 +1144,23 @@ export const useWorkbenchStore = defineStore("workbench", () => { function executeRealtimeRecoveryStep(step: WorkbenchRealtimeRecoveryStep): void { switch (step.type) { - case "refresh-session-messages": - void refreshRealtimeSessionMessages(step.sessionId, step.reason, { force: step.force }); + case "sync-replay": + void refreshWorkbenchSyncReplay(step.sessionId, step.traceId, step.sinceOutboxSeq, step.reason); return; - case "schedule-session-list": - scheduleSessionListRefresh(step.sessionId, runtimePolicy.sessionListRealtimeRefreshDelayMs); - return; - case "refresh-turn-status": - void refreshTurnStatusByTraceId(step.traceId, { force: step.force }); - return; - case "hydrate-trace-events": { - const message = latestMessageForTrace(step.traceId); - if (message) void hydrateTraceEventsForMessage(message, { force: step.force }); - return; - } } } + async function refreshWorkbenchSyncReplay(sessionId: string | null, traceId: string | null, sinceOutboxSeq: number | null, reason: string): Promise { + const result = await fetchWorkbenchSyncReplay({ sessionId, traceId, since: sinceOutboxSeq }, { timeoutMs: 8000, activityRef: () => activityRef.value }); + if (!result.ok || !result.data) { + recordWorkbenchRuntimeDiagnostic({ module: "workbench-sync-replay", sessionId, traceId, outcome: "network", diagnostic: { code: "workbench_sync_replay_failed", reason, status: result.status, apiError: result.apiError, valuesRedacted: true } }); + return; + } + const events = workbenchSyncReplayEvents(result.data); + for (const event of events) applyRealtimeEvent(event, realtimeEventName(event)); + recordWorkbenchRuntimeDiagnostic({ module: "workbench-sync-replay", sessionId, traceId, outcome: "ok", diagnostic: { code: "workbench_sync_replay_applied", reason, eventCount: events.length, sinceOutboxSeq, valuesRedacted: true } }); + } + function scheduleActiveTraceRestGapFill(traceId: string | null | undefined, reason: string, delayMs = runtimePolicy.workbenchActiveTraceRestGapFillInitialMs): void { const id = firstNonEmptyString(traceId); if (!id) return; @@ -1365,12 +1367,33 @@ export const useWorkbenchStore = defineStore("workbench", () => { const activeId = activeSessionId.value; const status = firstNonEmptyString(turn.status) ?? undefined; const terminalTurn = turn.terminal === true || isTerminalMessageStatus(status); - if (activeId && !terminalTurn && !messages.value.some((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId)) void refreshRealtimeSessionMessages(activeId, `realtime-turn-gap:${traceId}`); + if (activeId && !terminalTurn && !messages.value.some((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId)) { + recordWorkbenchRuntimeDiagnostic({ module: "workbench-realtime-authority", sessionId: activeId, traceId, outcome: "ok", diagnostic: { code: "workbench_realtime_turn_gap_no_legacy_repair", reason: "realtime-turn-gap", source: "turn-snapshot", valuesRedacted: true } }); + } const result = { ...turn, traceId, status, running: turn.running === true, terminal: turn.terminal === true, sessionId: firstNonEmptyString(turn.sessionId) ?? undefined, threadId: firstNonEmptyString(turn.threadId) ?? undefined, agentRun: turn.agentRun as AgentRunProvenance | undefined } as AgentChatResultResponse; rememberTurnStatus(traceId, result); scheduleRealtimeTurnProjection({ traceId, result, terminalTurn }); } + function realtimeEventName(event: WorkbenchRealtimeEvent): string { + switch (event.type) { + case "trace.snapshot": + return "workbench.trace.snapshot"; + case "trace.event": + return "workbench.trace.event"; + case "message.snapshot": + return "workbench.message.snapshot"; + case "turn.snapshot": + return "workbench.turn.snapshot"; + case "trace.unavailable": + return "workbench.trace.unavailable"; + case "error": + return "workbench.error"; + default: + return "message"; + } + } + function scheduleRealtimeTurnProjection(item: RealtimeTurnProjectionItem): void { realtimeTurnProjectionQueue.set(item.traceId, item); scheduleRealtimeTurnProjectionFlush(); diff --git a/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts b/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts index 1f5528bc..8d11d039 100644 --- a/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts +++ b/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts @@ -1,4 +1,4 @@ -// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first. +// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; PJ2026-010401080313 Workbench实时权威 draft-2026-07-08-p0-workbench-realtime-authority-v2. // Responsibility: Workbench SSE transport lifecycle, cursor ownership, turn-style tick state and recovery actions. // Mechanical source references: // - OpenCode stream.transport.ts:1-17 transport/turn coordination intent. @@ -35,7 +35,7 @@ export interface WorkbenchStreamTransportRestartResult { export type WorkbenchStreamTransportPhase = "idle" | "connecting" | "open" | "event" | "error" | "closed" | "blocked"; -export type WorkbenchStreamTransportRecoveryAction = "refresh-session-messages" | "schedule-session-list" | "refresh-turn-status" | "hydrate-trace-events"; +export type WorkbenchStreamTransportRecoveryAction = "sync-replay"; export interface WorkbenchStreamTransportState { key: string; @@ -58,6 +58,8 @@ export interface WorkbenchStreamTransportRecovery { tick: number; reason: string; actions: WorkbenchStreamTransportRecoveryAction[]; + outboxSeq: number | null; + traceSeq: number | null; diagnostic: Record; } @@ -177,11 +179,10 @@ export class WorkbenchStreamTransportRuntime { this.lastRecoveryAtByKey.set(key, now); const sessionId = input.sessionId ?? null; const traceId = input.traceId ?? null; - const actions: WorkbenchStreamTransportRecoveryAction[] = []; - if (sessionId) actions.push("refresh-session-messages", "schedule-session-list"); - if (traceId) actions.push("refresh-turn-status", "hydrate-trace-events"); + const cursor = this.currentCursor(key); + const actions: WorkbenchStreamTransportRecoveryAction[] = sessionId || traceId ? ["sync-replay"] : []; const diagnostic = this.diagnosticEnvelope("workbench_sse_recovery", reason, key, actions); - input.onRecovery?.({ key, sessionId, traceId, tick: this.wait?.tick ?? this.tick, reason, actions, diagnostic }); + input.onRecovery?.({ key, sessionId, traceId, tick: this.wait?.tick ?? this.tick, reason, actions, outboxSeq: cursor.outboxSeq, traceSeq: cursor.traceSeq, diagnostic }); } private emitState(input: WorkbenchStreamTransportRestartInput, phase: WorkbenchStreamTransportPhase, reason: string | null, errorName: string | null = null): void {