feat: route workbench realtime recovery through sync replay

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>): 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<string, unknown>) {
+  return { realtimeAuthority: WORKBENCH_REALTIME_AUTHORITY_VERSION, contractVersion: "workbench-sync-v1", ...input };
+}

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<string, unknown> | null;
+  cursor?: Record<string, unknown> | null;
+  events?: WorkbenchRealtimeEvent[];
+  delta?: WorkbenchRealtimeEvent[];
+  families?: Record<string, unknown> | null;
+  authority?: Record<string, unknown> | null;
   [key: string]: unknown;
 }
 
@@ -146,6 +184,21 @@ export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): Wo
   };
 }
 
+export async function fetchWorkbenchSyncReplay(input: WorkbenchSyncReplayRequest, options: ApiRequestOptions = {}): Promise<ApiResult<WorkbenchSyncReplayResponse>> {
+  return fetchJson<WorkbenchSyncReplayResponse>(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);

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;
 }

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<string>();
+  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<string, unknown>[] {
+  return Array.isArray(value) ? value.filter((item): item is Record<string, unknown> => Boolean(recordValue(item))) : [];
+}
+
+function recordValue(value: unknown): Record<string, unknown> | null {
+  return value && typeof value === "object" ? value as Record<string, unknown> : 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;
+}

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> = T extends WorkbenchRealtimeRecoveryStep ? Omit<T, "authority" | "force"> : 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>): 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[] {

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<void> {
+    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();

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<string, unknown>;
 }
 
@@ -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 {