fix: add workbench aggregate event stream (#2064)

internal/db/migrations/0001_cloud_core_skeleton.sql

@@ -1,3 +1,4 @@
+-- SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-24-p0-aggregate-event-stream.
 -- HWLAB L1 Cloud Core schema skeleton.
 -- This file is source schema only; applying it to DEV/PROD is a separate
 -- operator action and is not performed by source tests or smokes.
@@ -360,6 +361,47 @@ CREATE UNIQUE INDEX idx_workbench_trace_events_trace_source_event ON workbench_t
 DROP INDEX IF EXISTS idx_workbench_trace_events_trace_projected_seq;
 CREATE UNIQUE INDEX idx_workbench_trace_events_trace_projected_seq ON workbench_trace_events(trace_id, projected_seq);
 
+CREATE TABLE IF NOT EXISTS workbench_event_sequences (
+  aggregate_id TEXT PRIMARY KEY,
+  aggregate_type TEXT NOT NULL DEFAULT 'trace',
+  session_id TEXT,
+  turn_id TEXT,
+  trace_id TEXT,
+  last_seq INTEGER NOT NULL DEFAULT 0,
+  created_at TEXT NOT NULL,
+  updated_at TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_workbench_event_sequences_session ON workbench_event_sequences(session_id, updated_at DESC, aggregate_id);
+CREATE INDEX IF NOT EXISTS idx_workbench_event_sequences_trace ON workbench_event_sequences(trace_id, updated_at DESC, aggregate_id);
+
+CREATE TABLE IF NOT EXISTS workbench_events (
+  event_seq BIGSERIAL PRIMARY KEY,
+  event_id TEXT NOT NULL UNIQUE,
+  aggregate_id TEXT NOT NULL,
+  aggregate_type TEXT NOT NULL DEFAULT 'trace',
+  aggregate_seq INTEGER NOT NULL,
+  session_id TEXT,
+  turn_id TEXT,
+  trace_id TEXT,
+  message_id TEXT,
+  source_run_id TEXT,
+  source_command_id TEXT,
+  source_seq INTEGER NOT NULL DEFAULT 0,
+  source_event_id TEXT,
+  event_type TEXT NOT NULL DEFAULT 'event',
+  projection_revision INTEGER NOT NULL DEFAULT 0,
+  terminal BOOLEAN NOT NULL DEFAULT false,
+  sealed BOOLEAN NOT NULL DEFAULT false,
+  payload_json TEXT NOT NULL DEFAULT '{}',
+  occurred_at TEXT NOT NULL,
+  committed_at TEXT NOT NULL
+);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_workbench_events_aggregate_seq ON workbench_events(aggregate_id, aggregate_seq);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_workbench_events_aggregate_source_event ON workbench_events(aggregate_id, source_event_id) WHERE source_event_id IS NOT NULL;
+CREATE INDEX IF NOT EXISTS idx_workbench_events_trace_seq ON workbench_events(trace_id, event_seq);
+CREATE INDEX IF NOT EXISTS idx_workbench_events_session_seq ON workbench_events(session_id, event_seq);
+CREATE INDEX IF NOT EXISTS idx_workbench_events_type_seq ON workbench_events(event_type, event_seq);
+
 CREATE TABLE IF NOT EXISTS workbench_projection_checkpoints (
   trace_id TEXT PRIMARY KEY,
   session_id TEXT,
@@ -383,6 +425,10 @@ CREATE INDEX IF NOT EXISTS idx_workbench_projection_checkpoints_session_updated
 
 CREATE TABLE IF NOT EXISTS workbench_projection_outbox (
   outbox_seq BIGSERIAL PRIMARY KEY,
+  event_seq BIGINT,
+  aggregate_id TEXT,
+  aggregate_seq INTEGER NOT NULL DEFAULT 0,
+  projection_revision INTEGER NOT NULL DEFAULT 0,
   trace_id TEXT NOT NULL,
   session_id TEXT,
   turn_id TEXT,
@@ -396,9 +442,14 @@ CREATE TABLE IF NOT EXISTS workbench_projection_outbox (
   payload_json TEXT NOT NULL DEFAULT '{}',
   created_at TEXT NOT NULL
 );
+ALTER TABLE workbench_projection_outbox ADD COLUMN IF NOT EXISTS event_seq BIGINT;
+ALTER TABLE workbench_projection_outbox ADD COLUMN IF NOT EXISTS aggregate_id TEXT;
+ALTER TABLE workbench_projection_outbox ADD COLUMN IF NOT EXISTS aggregate_seq INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE workbench_projection_outbox ADD COLUMN IF NOT EXISTS projection_revision INTEGER NOT NULL DEFAULT 0;
 CREATE INDEX IF NOT EXISTS idx_workbench_projection_outbox_trace_seq ON workbench_projection_outbox(trace_id, outbox_seq);
 CREATE INDEX IF NOT EXISTS idx_workbench_projection_outbox_session_seq ON workbench_projection_outbox(session_id, outbox_seq);
 CREATE INDEX IF NOT EXISTS idx_workbench_projection_outbox_after_seq ON workbench_projection_outbox(outbox_seq);
+CREATE INDEX IF NOT EXISTS idx_workbench_projection_outbox_event_seq ON workbench_projection_outbox(event_seq) WHERE event_seq IS NOT NULL;
 
 INSERT INTO workbench_sessions (
   session_id,
@@ -573,7 +624,7 @@ CREATE TABLE IF NOT EXISTS hwlab_schema_migrations (
 INSERT INTO hwlab_schema_migrations (id, schema_version, applied_at, migration_json)
 VALUES (
   '0001_cloud_core_skeleton',
-  'runtime-durable-postgres-v5',
+  'runtime-durable-postgres-v6',
   CURRENT_TIMESTAMP,
   '{"path":"internal/db/migrations/0001_cloud_core_skeleton.sql","runtime":"cloud-api"}'
 )

internal/db/runtime-store.test.ts

@@ -857,6 +857,80 @@ test("configured postgres runtime persists and queries Workbench durable facts",
   assert.ok(queryClient.calls.some((call) => call.sql.startsWith("SELECT session_json FROM workbench_sessions WHERE session_id = $1")));
 });
 
+test("configured postgres runtime appends Workbench aggregate events and outbox in one transaction", async () => {
+  const queryClient = createFakePostgresClient({ migrationReady: true });
+  const store = createConfiguredCloudRuntimeStore({
+    env: {
+      HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres",
+      HWLAB_CLOUD_RUNTIME_DURABLE: "true"
+    },
+    dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab",
+    queryClient,
+    now: () => "2026-06-24T12:00:00.000Z"
+  });
+
+  const write = await store.writeWorkbenchFacts({ facts: {
+    traceEvents: [{ id: "wte_event_stream_1", traceId: "trc_event_stream", sessionId: "ses_event_stream", turnId: "turn_event_stream", sourceSeq: 1, sourceEventId: "source-event-1", projectedSeq: 1, eventType: "assistant", occurredAt: "2026-06-24T12:00:01.000Z" }],
+    turns: [{ turnId: "turn_event_stream", sessionId: "ses_event_stream", traceId: "trc_event_stream", status: "completed", sourceSeq: 2, sourceEventId: "source-terminal-2", projectedSeq: 2, terminal: true, sealed: true, updatedAt: "2026-06-24T12:00:02.000Z" }]
+  } });
+  const outbox = await store.readWorkbenchProjectionOutbox({ afterSeq: 0, traceId: "trc_event_stream", limit: 10 });
+
+  assert.equal(write.events.length, 2);
+  assert.deepEqual(write.events.map((event) => event.aggregateId), ["session:ses_event_stream", "session:ses_event_stream"]);
+  assert.deepEqual(write.events.map((event) => event.aggregateSeq), [1, 2]);
+  assert.equal(outbox.length, 2);
+  assert.equal(outbox[0].eventSeq, write.events[0].eventSeq);
+  assert.equal(outbox[0].aggregateId, "session:ses_event_stream");
+  assert.equal(outbox[1].commitType, "terminal");
+  assert.equal(outbox[1].aggregateSeq, 2);
+  assert.ok(queryClient.calls.some((call) => call.sql === "BEGIN"));
+  assert.ok(queryClient.calls.some((call) => call.sql === "COMMIT"));
+  assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO workbench_events")));
+  assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO workbench_projection_outbox")));
+});
+
+test("configured postgres runtime reuses aggregate event sequence for duplicate source event", async () => {
+  const queryClient = createFakePostgresClient({ migrationReady: true });
+  const store = createConfiguredCloudRuntimeStore({
+    env: {
+      HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres",
+      HWLAB_CLOUD_RUNTIME_DURABLE: "true"
+    },
+    dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab",
+    queryClient,
+    now: () => "2026-06-24T12:10:00.000Z"
+  });
+  const facts = { traceEvents: [{ id: "wte_duplicate_source", traceId: "trc_duplicate_source", sessionId: "ses_duplicate_source", turnId: "turn_duplicate_source", sourceSeq: 1, sourceEventId: "source-dup", projectedSeq: 1, eventType: "assistant", occurredAt: "2026-06-24T12:10:01.000Z" }] };
+
+  const first = await store.writeWorkbenchFacts({ facts });
+  const repeated = await store.writeWorkbenchFacts({ facts });
+
+  assert.equal(first.events[0].eventSeq, repeated.events[0].eventSeq);
+  assert.equal(first.events[0].aggregateSeq, repeated.events[0].aggregateSeq);
+  assert.equal(first.events[0].projectionRevision, repeated.events[0].projectionRevision);
+});
+
+test("configured postgres runtime rolls back Workbench facts when outbox append fails", async () => {
+  const queryClient = createFakePostgresClient({ migrationReady: true, outboxErrorCode: "XX000" });
+  const store = createConfiguredCloudRuntimeStore({
+    env: {
+      HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres",
+      HWLAB_CLOUD_RUNTIME_DURABLE: "true"
+    },
+    dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab",
+    queryClient,
+    now: () => "2026-06-24T12:20:00.000Z"
+  });
+
+  await assert.rejects(
+    () => store.writeWorkbenchFacts({ facts: { traceEvents: [{ id: "wte_outbox_fail", traceId: "trc_outbox_fail", sessionId: "ses_outbox_fail", turnId: "turn_outbox_fail", sourceSeq: 1, sourceEventId: "source-outbox-fail", projectedSeq: 1, eventType: "assistant", occurredAt: "2026-06-24T12:20:01.000Z" }] } }),
+    /projection outbox insert failed/u
+  );
+  assert.ok(queryClient.calls.some((call) => call.sql === "BEGIN"));
+  assert.ok(queryClient.calls.some((call) => call.sql === "ROLLBACK"));
+  assert.equal(queryClient.calls.some((call) => call.sql === "COMMIT"), false);
+});
+
 test("configured postgres runtime allocates Workbench projectedSeq idempotently by source event", async () => {
   const queryClient = createFakePostgresClient({ migrationReady: true });
   const store = createConfiguredCloudRuntimeStore({
@@ -999,7 +1073,7 @@ test("configured postgres runtime orders Workbench sessions by sessionsOrder upd
   assert.match(readCall?.sql ?? "", /ORDER BY updated_at DESC/u);
 });
 
-test("configured postgres runtime queries requested workbench fact families concurrently", async () => {
+test("configured postgres runtime queries requested workbench fact families sequentially", async () => {
   const expectedReads = 4;
   let started = 0;
   let releaseReads;
@@ -1090,7 +1164,7 @@ test("configured postgres runtime queries requested workbench fact families conc
   releaseReads();
   const loaded = await pending;
 
-  assert.equal(startedConcurrently, true);
+  assert.equal(startedConcurrently, false);
   assert.equal(started, expectedReads);
   assert.equal(loaded.count, 4);
   assert.equal(loaded.facts.messages[0].messageId, "msg_parallel");
@@ -1134,6 +1208,7 @@ function createFakePostgresClient({
   migrationErrorCode = null,
   countErrorCode = null,
   readErrorCode = null,
+  outboxErrorCode = null,
   beforeWorkbenchFactRead = null
 } = {}) {
   const state = {
@@ -1151,7 +1226,10 @@ function createFakePostgresClient({
     workbench_parts: new Map(),
     workbench_turns: new Map(),
     workbench_trace_events: new Map(),
+    workbench_event_sequences: new Map(),
+    workbench_events: new Map(),
     workbench_projection_checkpoints: new Map(),
+    workbench_projection_outbox: new Map(),
     hwlab_schema_migrations: new Map()
   };
   if (migrationReady) {
@@ -1216,6 +1294,12 @@ function createFakePostgresClient({
         const checkpointMax = Number(state.workbench_projection_checkpoints.get(traceId)?.projected_seq ?? 0);
         return { rows: [{ max_projected_seq: Math.max(eventMax, checkpointMax) }] };
       }
+      if (sql.startsWith("SELECT event_seq, aggregate_seq, projection_revision FROM workbench_events")) {
+        const row = [...state.workbench_events.values()]
+          .filter((record) => record.event_id === params[0] || (params[1] !== null && params[1] !== undefined && record.aggregate_id === params[2] && record.source_event_id === params[1]))
+          .sort((left, right) => Number(left.event_seq) - Number(right.event_seq))[0] ?? null;
+        return { rows: row ? [{ event_seq: row.event_seq, aggregate_seq: row.aggregate_seq, projection_revision: row.projection_revision }] : [] };
+      }
       if (sql.startsWith("SELECT gateway_session_json FROM gateway_sessions")) {
         return jsonSelect(state.gateway_sessions, params[0], "gateway_session_json");
       }
@@ -1299,6 +1383,19 @@ function createFakePostgresClient({
         if (beforeWorkbenchFactRead) await beforeWorkbenchFactRead(sql, params);
         return workbenchFactRows(state.workbench_projection_checkpoints, "checkpoint_json", sql, params, readErrorCode);
       }
+      if (sql.startsWith("SELECT outbox_seq, event_seq, aggregate_id")) {
+        const afterSeq = Number(params[0] ?? 0);
+        const traceFilter = sql.includes("trace_id = $2") ? params[1] : null;
+        const sessionFilter = sql.includes("session_id = $2") ? params[1] : sql.includes("session_id = $3") ? params[2] : null;
+        const limit = Number(params[params.length - 1] ?? 100);
+        const rows = [...state.workbench_projection_outbox.values()]
+          .filter((record) => Number(record.outbox_seq) > afterSeq)
+          .filter((record) => !traceFilter || record.trace_id === traceFilter)
+          .filter((record) => !sessionFilter || record.session_id === sessionFilter)
+          .sort((left, right) => Number(left.outbox_seq) - Number(right.outbox_seq))
+          .slice(0, limit);
+        return { rows };
+      }
       if (sql.startsWith("SELECT metadata_json FROM evidence_records")) {
         if (readErrorCode) {
           const error = new Error("evidence read query failed");
@@ -1365,6 +1462,49 @@ function createFakePostgresClient({
         });
         return { rows: [] };
       }
+      if (sql.startsWith("INSERT INTO workbench_event_sequences")) {
+        if (!state.workbench_event_sequences.has(params[0])) {
+          state.workbench_event_sequences.set(params[0], {
+            aggregate_id: params[0], aggregate_type: params[1], session_id: params[2], turn_id: params[3], trace_id: params[4], last_seq: 0, created_at: params[5], updated_at: params[6]
+          });
+        }
+        return { rows: [] };
+      }
+      if (sql.startsWith("UPDATE workbench_event_sequences SET last_seq")) {
+        const record = state.workbench_event_sequences.get(params[0]);
+        const nextSeq = Number(record?.last_seq ?? 0) + 1;
+        state.workbench_event_sequences.set(params[0], { ...(record ?? { aggregate_id: params[0] }), last_seq: nextSeq, updated_at: params[1] });
+        return { rows: [{ last_seq: nextSeq }] };
+      }
+      if (sql.startsWith("INSERT INTO workbench_events")) {
+        const eventSeq = state.workbench_events.size + 1;
+        const record = {
+          event_seq: eventSeq,
+          event_id: params[0],
+          aggregate_id: params[1],
+          aggregate_type: params[2],
+          aggregate_seq: params[3],
+          session_id: params[4],
+          turn_id: params[5],
+          trace_id: params[6],
+          message_id: params[7],
+          source_run_id: params[8],
+          source_command_id: params[9],
+          source_seq: params[10],
+          source_event_id: params[11],
+          event_type: params[12],
+          projection_revision: params[13],
+          terminal: params[14],
+          sealed: params[15],
+          payload_json: params[16],
+          occurred_at: params[17],
+          committed_at: params[18]
+        };
+        const existing = [...state.workbench_events.values()].find((item) => item.event_id === record.event_id) ?? null;
+        state.workbench_events.set(existing?.event_seq ?? eventSeq, existing ? { ...existing, projection_revision: Math.max(Number(existing.projection_revision ?? 0), Number(record.projection_revision ?? 0)), terminal: Boolean(existing.terminal) || Boolean(record.terminal), sealed: Boolean(existing.sealed) || Boolean(record.sealed), payload_json: record.payload_json, committed_at: record.committed_at } : record);
+        const saved = existing ? state.workbench_events.get(existing.event_seq) : record;
+        return { rows: [{ event_seq: saved.event_seq, aggregate_seq: saved.aggregate_seq, projection_revision: saved.projection_revision }] };
+      }
       if (sql.startsWith("INSERT INTO workbench_sessions")) {
         state.workbench_sessions.set(params[0], {
           session_id: params[0], owner_user_id: params[1], project_id: params[2], conversation_id: params[3], thread_id: params[4], status: params[5], last_trace_id: params[6], projected_seq: params[7], source_seq: params[8], source_event_id: params[9], terminal: params[10], sealed: params[11], session_json: params[12], created_at: params[13], updated_at: params[14]
@@ -1395,6 +1535,34 @@ function createFakePostgresClient({
         });
         return { rows: [] };
       }
+      if (sql.startsWith("INSERT INTO workbench_projection_outbox")) {
+        if (outboxErrorCode) {
+          const error = new Error("projection outbox insert failed");
+          error.code = outboxErrorCode;
+          throw error;
+        }
+        const outboxSeq = state.workbench_projection_outbox.size + 1;
+        state.workbench_projection_outbox.set(outboxSeq, {
+          outbox_seq: outboxSeq,
+          event_seq: params[0],
+          aggregate_id: params[1],
+          aggregate_seq: params[2],
+          projection_revision: params[3],
+          trace_id: params[4],
+          session_id: params[5],
+          turn_id: params[6],
+          message_id: params[7],
+          projected_seq: params[8],
+          source_seq: params[9],
+          source_event_id: params[10],
+          commit_type: params[11],
+          terminal: params[12],
+          sealed: params[13],
+          payload_json: params[14],
+          created_at: params[15]
+        });
+        return { rows: [] };
+      }
       if (sql.startsWith("INSERT INTO workbench_projection_checkpoints")) {
         const previous = state.workbench_projection_checkpoints.get(params[0]) ?? null;
         const projectedSeq = sql.includes("RETURNING projected_seq")

internal/db/runtime-store.ts

@@ -1,5 +1,5 @@
 /*
- * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; draft-2026-06-20-p1-zero-split-durable-realtime.
+ * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; draft-2026-06-20-p1-zero-split-durable-realtime; draft-2026-06-24-p0-aggregate-event-stream.
  * 职责: Cloud runtime durable store。Code Agent trace/session/projection cursor 投影事实必须从同一持久化适配器写入和恢复。
  */
 import { createHash, randomUUID } from "node:crypto";
@@ -59,6 +59,8 @@ const postgresCountTables = Object.freeze([
   "workbench_parts",
   "workbench_turns",
   "workbench_trace_events",
+  "workbench_event_sequences",
+  "workbench_events",
   "workbench_projection_checkpoints"
 ]);
 
@@ -81,6 +83,11 @@ const postgresRuntimeReadIndexes = Object.freeze([
   "CREATE INDEX IF NOT EXISTS idx_workbench_turns_trace ON workbench_turns(trace_id, turn_id)",
   "CREATE INDEX IF NOT EXISTS idx_workbench_trace_events_trace_projected_seq ON workbench_trace_events(trace_id, projected_seq)",
   "CREATE INDEX IF NOT EXISTS idx_workbench_trace_events_session_projected_seq ON workbench_trace_events(session_id, projected_seq)",
+  "CREATE INDEX IF NOT EXISTS idx_workbench_event_sequences_session ON workbench_event_sequences(session_id, updated_at DESC, aggregate_id)",
+  "CREATE INDEX IF NOT EXISTS idx_workbench_event_sequences_trace ON workbench_event_sequences(trace_id, updated_at DESC, aggregate_id)",
+  "CREATE INDEX IF NOT EXISTS idx_workbench_events_trace_seq ON workbench_events(trace_id, event_seq)",
+  "CREATE INDEX IF NOT EXISTS idx_workbench_events_session_seq ON workbench_events(session_id, event_seq)",
+  "CREATE INDEX IF NOT EXISTS idx_workbench_events_type_seq ON workbench_events(event_type, event_seq)",
   "CREATE INDEX IF NOT EXISTS idx_workbench_projection_checkpoints_status_updated ON workbench_projection_checkpoints(projection_status, updated_at DESC, trace_id)",
   "CREATE INDEX IF NOT EXISTS idx_workbench_projection_checkpoints_session_updated ON workbench_projection_checkpoints(session_id, updated_at DESC, trace_id)"
 ]);
@@ -157,6 +164,8 @@ export class CloudRuntimeStore {
     this.workbenchParts = new Map();
     this.workbenchTurns = new Map();
     this.workbenchTraceEvents = new Map();
+    this.workbenchEventSequences = new Map();
+    this.workbenchEvents = new Map();
     this.workbenchProjectionCheckpoints = new Map();
   }
 
@@ -186,6 +195,8 @@ export class CloudRuntimeStore {
         workbenchParts: this.workbenchParts.size,
         workbenchTurns: this.workbenchTurns.size,
         workbenchTraceEvents: this.workbenchTraceEvents.size,
+        workbenchEventSequences: this.workbenchEventSequences.size,
+        workbenchEvents: this.workbenchEvents.size,
         workbenchProjectionCheckpoints: this.workbenchProjectionCheckpoints.size
       }
     });
@@ -1292,37 +1303,25 @@ export class PostgresCloudRuntimeStore {
     if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
       await this.assertReadyForWrites();
     }
-    const result = this.memory.writeWorkbenchFacts(params, requestMeta);
-    for (const fact of result.facts.sessions) await this.persistWorkbenchSessionFact(fact);
-    for (const fact of result.facts.messages) await this.persistWorkbenchMessageFact(fact);
-    for (const fact of result.facts.parts) await this.persistWorkbenchPartFact(fact);
-    for (const fact of result.facts.turns) await this.persistWorkbenchTurnFact(fact);
-    for (const fact of result.facts.traceEvents) await this.persistWorkbenchTraceEventFact(fact);
-    for (const fact of result.facts.checkpoints) await this.persistWorkbenchProjectionCheckpoint(fact);
-    for (const fact of result.facts.traceEvents) {
-      try {
+    const facts = normalizeWorkbenchFacts(params.facts ?? params, requestMeta, this.now());
+    const aggregateEvents = normalizeWorkbenchAggregateEventsForFacts(facts, requestMeta, this.now());
+    const txResult = await this.withDurableTransaction("workbench.facts.write", async (client) => {
+      const persistedEvents = [];
+      for (const event of aggregateEvents) persistedEvents.push(await this.persistWorkbenchAggregateEvent(event, client));
+      const eventIndex = indexWorkbenchAggregateEvents(persistedEvents);
+      for (const fact of facts.sessions) await this.persistWorkbenchSessionFact(fact, client);
+      for (const fact of facts.messages) await this.persistWorkbenchMessageFact(fact, client);
+      for (const fact of facts.parts) await this.persistWorkbenchPartFact(fact, client);
+      for (const fact of facts.turns) await this.persistWorkbenchTurnFact(fact, client);
+      for (const fact of facts.traceEvents) await this.persistWorkbenchTraceEventFact(fact, client);
+      for (const fact of facts.checkpoints) await this.persistWorkbenchProjectionCheckpoint(fact, client);
+      for (const fact of facts.traceEvents) {
+        const event = eventIndex.get(`traceEvent:${fact.id}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
         await this.persistWorkbenchProjectionOutbox({
-        traceId: fact.traceId,
-        sessionId: fact.sessionId,
-        turnId: fact.turnId,
-        messageId: fact.messageId,
-        projectedSeq: fact.projectedSeq,
-        sourceSeq: fact.sourceSeq,
-        sourceEventId: fact.sourceEventId,
-        commitType: fact.terminal ? "terminal" : "event",
-        terminal: Boolean(fact.terminal),
-        sealed: Boolean(fact.sealed),
-        payload: { type: fact.eventType, projectedSeq: fact.projectedSeq, valuesRedacted: true },
-        createdAt: fact.occurredAt ?? fact.updatedAt ?? this.now()
-      });
-    } catch (outboxError) {
-      console.error(`[workbench-projection] outbox write failed: traceId=${fact.traceId} projectedSeq=${fact.projectedSeq} error=${outboxError?.message ?? outboxError}`);
-    }
-  }
-    for (const fact of result.facts.turns) {
-      if (fact.terminal || fact.sealed) {
-        try {
-          await this.persistWorkbenchProjectionOutbox({
+          eventSeq: event?.eventSeq ?? null,
+          aggregateId: event?.aggregateId ?? null,
+          aggregateSeq: event?.aggregateSeq ?? 0,
+          projectionRevision: event?.projectionRevision ?? fact.projectedSeq,
           traceId: fact.traceId,
           sessionId: fact.sessionId,
           turnId: fact.turnId,
@@ -1330,18 +1329,40 @@ export class PostgresCloudRuntimeStore {
           projectedSeq: fact.projectedSeq,
           sourceSeq: fact.sourceSeq,
           sourceEventId: fact.sourceEventId,
-          commitType: "terminal",
-          terminal: true,
-          sealed: true,
-          payload: { status: fact.status, failureKind: fact.failureKind, valuesRedacted: true },
-          createdAt: fact.updatedAt ?? this.now()
-        });
-      } catch (outboxError) {
-        console.error(`[workbench-projection] terminal outbox write failed: traceId=${fact.traceId} error=${outboxError?.message ?? outboxError}`);
+          commitType: fact.terminal ? "terminal" : "event",
+          terminal: Boolean(fact.terminal),
+          sealed: Boolean(fact.sealed),
+          payload: { type: fact.eventType, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null, projectedSeq: fact.projectedSeq, valuesRedacted: true },
+          createdAt: fact.occurredAt ?? fact.updatedAt ?? this.now()
+        }, client);
       }
-    }
-  }
-    return withPersistence(result, this.summary());
+      for (const fact of facts.turns) {
+        if (fact.terminal || fact.sealed) {
+          const event = eventIndex.get(`turn:${fact.turnId}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
+          await this.persistWorkbenchProjectionOutbox({
+            eventSeq: event?.eventSeq ?? null,
+            aggregateId: event?.aggregateId ?? null,
+            aggregateSeq: event?.aggregateSeq ?? 0,
+            projectionRevision: event?.projectionRevision ?? fact.projectedSeq,
+            traceId: fact.traceId,
+            sessionId: fact.sessionId,
+            turnId: fact.turnId,
+            messageId: fact.messageId,
+            projectedSeq: fact.projectedSeq,
+            sourceSeq: fact.sourceSeq,
+            sourceEventId: fact.sourceEventId,
+            commitType: "terminal",
+            terminal: true,
+            sealed: true,
+            payload: { status: fact.status, failureKind: fact.failureKind, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null, valuesRedacted: true },
+            createdAt: fact.updatedAt ?? this.now()
+          }, client);
+        }
+      }
+      return { events: persistedEvents };
+    });
+    this.memory.writeWorkbenchFacts({ facts }, requestMeta);
+    return withPersistence({ written: true, facts, events: txResult.events }, this.summary());
   }
 
   async queryWorkbenchFacts(params = {}) {
@@ -1666,45 +1687,82 @@ export class PostgresCloudRuntimeStore {
     );
   }
 
-  async persistWorkbenchSessionFact(record) {
-    await this.query(
+  async persistWorkbenchAggregateEvent(record, client = this) {
+    const existing = await client.query(
+      "SELECT event_seq, aggregate_seq, projection_revision FROM workbench_events WHERE event_id = $1 OR ($2::text IS NOT NULL AND aggregate_id = $3 AND source_event_id = $2) ORDER BY event_seq ASC LIMIT 1",
+      [record.eventId, record.sourceEventId, record.aggregateId]
+    );
+    const existingRow = existing.rows?.[0] ?? null;
+    if (existingRow) {
+      return {
+        ...record,
+        eventSeq: Number(existingRow.event_seq),
+        aggregateSeq: Number(existingRow.aggregate_seq),
+        projectionRevision: Number(existingRow.projection_revision)
+      };
+    }
+    await client.query(
+      "INSERT INTO workbench_event_sequences (aggregate_id, aggregate_type, session_id, turn_id, trace_id, last_seq, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,0,$6,$7) ON CONFLICT (aggregate_id) DO UPDATE SET aggregate_type = EXCLUDED.aggregate_type, session_id = COALESCE(workbench_event_sequences.session_id, EXCLUDED.session_id), turn_id = COALESCE(workbench_event_sequences.turn_id, EXCLUDED.turn_id), trace_id = COALESCE(workbench_event_sequences.trace_id, EXCLUDED.trace_id), updated_at = EXCLUDED.updated_at",
+      [record.aggregateId, record.aggregateType, record.sessionId, record.turnId, record.traceId, record.committedAt, record.committedAt]
+    );
+    const reserved = await client.query(
+      "UPDATE workbench_event_sequences SET last_seq = last_seq + 1, updated_at = $2 WHERE aggregate_id = $1 RETURNING last_seq",
+      [record.aggregateId, record.committedAt]
+    );
+    const aggregateSeq = nonNegativeInteger(reserved.rows?.[0]?.last_seq);
+    const projectionRevision = nonNegativeInteger(record.projectionRevision) || aggregateSeq;
+    const inserted = await client.query(
+      "INSERT INTO workbench_events (event_id, aggregate_id, aggregate_type, aggregate_seq, session_id, turn_id, trace_id, message_id, source_run_id, source_command_id, source_seq, source_event_id, event_type, projection_revision, terminal, sealed, payload_json, occurred_at, committed_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19) ON CONFLICT (event_id) DO UPDATE SET projection_revision = GREATEST(workbench_events.projection_revision, EXCLUDED.projection_revision), terminal = workbench_events.terminal OR EXCLUDED.terminal, sealed = workbench_events.sealed OR EXCLUDED.sealed, payload_json = EXCLUDED.payload_json, committed_at = EXCLUDED.committed_at RETURNING event_seq, aggregate_seq, projection_revision",
+      [record.eventId, record.aggregateId, record.aggregateType, aggregateSeq, record.sessionId, record.turnId, record.traceId, record.messageId, record.sourceRunId, record.sourceCommandId, record.sourceSeq, record.sourceEventId, record.eventType, projectionRevision, Boolean(record.terminal), Boolean(record.sealed), stableJson(record.payload ?? record), record.occurredAt, record.committedAt]
+    );
+    const row = inserted.rows?.[0] ?? {};
+    return {
+      ...record,
+      eventSeq: Number(row.event_seq),
+      aggregateSeq: Number(row.aggregate_seq) || aggregateSeq,
+      projectionRevision: Number(row.projection_revision) || projectionRevision
+    };
+  }
+
+  async persistWorkbenchSessionFact(record, client = this) {
+    await client.query(
       "INSERT INTO workbench_sessions (session_id, owner_user_id, project_id, conversation_id, thread_id, status, last_trace_id, projected_seq, source_seq, source_event_id, terminal, sealed, session_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) ON CONFLICT (session_id) DO UPDATE SET owner_user_id = EXCLUDED.owner_user_id, project_id = EXCLUDED.project_id, conversation_id = EXCLUDED.conversation_id, thread_id = EXCLUDED.thread_id, status = EXCLUDED.status, last_trace_id = EXCLUDED.last_trace_id, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, session_json = EXCLUDED.session_json, updated_at = EXCLUDED.updated_at",
       [record.sessionId, record.ownerUserId, record.projectId, record.conversationId, record.threadId, record.status, record.lastTraceId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.terminal, record.sealed, stableJson(record), record.createdAt, record.updatedAt]
     );
   }
 
-  async persistWorkbenchMessageFact(record) {
-    await this.query(
+  async persistWorkbenchMessageFact(record, client = this) {
+    await client.query(
       "INSERT INTO workbench_messages (message_id, session_id, turn_id, trace_id, role, status, projected_seq, source_seq, source_event_id, terminal, sealed, message_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT (message_id) DO UPDATE SET session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, trace_id = EXCLUDED.trace_id, role = EXCLUDED.role, status = EXCLUDED.status, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, message_json = EXCLUDED.message_json, updated_at = EXCLUDED.updated_at",
       [record.messageId, record.sessionId, record.turnId, record.traceId, record.role, record.status, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.terminal, record.sealed, stableJson(record), record.createdAt, record.updatedAt]
     );
   }
 
-  async persistWorkbenchPartFact(record) {
-    await this.query(
+  async persistWorkbenchPartFact(record, client = this) {
+    await client.query(
       "INSERT INTO workbench_parts (part_id, message_id, session_id, turn_id, trace_id, part_index, part_type, status, projected_seq, source_seq, source_event_id, terminal, sealed, part_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (part_id) DO UPDATE SET message_id = EXCLUDED.message_id, session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, trace_id = EXCLUDED.trace_id, part_index = EXCLUDED.part_index, part_type = EXCLUDED.part_type, status = EXCLUDED.status, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, part_json = EXCLUDED.part_json, updated_at = EXCLUDED.updated_at",
       [record.partId, record.messageId, record.sessionId, record.turnId, record.traceId, record.partIndex, record.partType, record.status, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.terminal, record.sealed, stableJson(record), record.createdAt, record.updatedAt]
     );
   }
 
-  async persistWorkbenchTurnFact(record) {
-    await this.query(
+  async persistWorkbenchTurnFact(record, client = this) {
+    await client.query(
       "INSERT INTO workbench_turns (turn_id, session_id, trace_id, message_id, status, projected_seq, source_seq, source_event_id, terminal, sealed, final_response_json, diagnostic_json, turn_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) ON CONFLICT (turn_id) DO UPDATE SET session_id = EXCLUDED.session_id, trace_id = EXCLUDED.trace_id, message_id = EXCLUDED.message_id, status = EXCLUDED.status, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, final_response_json = EXCLUDED.final_response_json, diagnostic_json = EXCLUDED.diagnostic_json, turn_json = EXCLUDED.turn_json, updated_at = EXCLUDED.updated_at",
       [record.turnId, record.sessionId, record.traceId, record.messageId, record.status, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.terminal, record.sealed, stableJson(record.finalResponse), stableJson(record.diagnostic), stableJson(record), record.createdAt, record.updatedAt]
     );
   }
 
-  async persistWorkbenchTraceEventFact(record) {
-    await this.query(
+  async persistWorkbenchTraceEventFact(record, client = this) {
+    await client.query(
       "INSERT INTO workbench_trace_events (id, trace_id, session_id, turn_id, message_id, source_seq, source_event_id, projected_seq, event_type, terminal, sealed, event_json, occurred_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT (id) DO UPDATE SET trace_id = EXCLUDED.trace_id, session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, message_id = EXCLUDED.message_id, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, projected_seq = EXCLUDED.projected_seq, event_type = EXCLUDED.event_type, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, event_json = EXCLUDED.event_json, occurred_at = EXCLUDED.occurred_at, updated_at = EXCLUDED.updated_at",
       [record.id, record.traceId, record.sessionId, record.turnId, record.messageId, record.sourceSeq, record.sourceEventId, record.projectedSeq, record.eventType, record.terminal, record.sealed, stableJson(record), record.occurredAt, record.updatedAt]
     );
   }
 
-  async persistWorkbenchProjectionOutbox(record) {
-    await this.query(
-      "INSERT INTO workbench_projection_outbox (trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)",
-      [record.traceId, record.sessionId, record.turnId, record.messageId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.commitType ?? "event", Boolean(record.terminal), Boolean(record.sealed), stableJson(record.payload ?? record), record.createdAt ?? this.now()]
+  async persistWorkbenchProjectionOutbox(record, client = this) {
+    await client.query(
+      "INSERT INTO workbench_projection_outbox (event_seq, aggregate_id, aggregate_seq, projection_revision, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)",
+      [record.eventSeq, record.aggregateId, nonNegativeInteger(record.aggregateSeq), nonNegativeInteger(record.projectionRevision), record.traceId, record.sessionId, record.turnId, record.messageId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.commitType ?? "event", Boolean(record.terminal), Boolean(record.sealed), stableJson(record.payload ?? record), record.createdAt ?? this.now()]
     );
   }
 
@@ -1725,11 +1783,15 @@ export class PostgresCloudRuntimeStore {
     }
     params.push(Math.min(Math.max(Number(limit) || 100, 1), 500));
     const result = await this.query(
-      `SELECT outbox_seq, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE ${where} ORDER BY outbox_seq ASC LIMIT $${paramIdx}`,
+      `SELECT outbox_seq, event_seq, aggregate_id, aggregate_seq, projection_revision, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE ${where} ORDER BY outbox_seq ASC LIMIT $${paramIdx}`,
       params
     );
     return (result.rows ?? []).map((row) => ({
       outboxSeq: Number(row.outbox_seq),
+      eventSeq: row.event_seq === null || row.event_seq === undefined ? null : Number(row.event_seq),
+      aggregateId: row.aggregate_id,
+      aggregateSeq: Number(row.aggregate_seq ?? 0),
+      projectionRevision: Number(row.projection_revision ?? 0),
       traceId: row.trace_id,
       sessionId: row.session_id,
       turnId: row.turn_id,
@@ -1746,8 +1808,8 @@ export class PostgresCloudRuntimeStore {
     }));
   }
 
-  async persistWorkbenchProjectionCheckpoint(record) {
-    await this.query(
+  async persistWorkbenchProjectionCheckpoint(record, client = this) {
+    await client.query(
       "INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET session_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.session_id ELSE workbench_projection_checkpoints.session_id END, turn_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.turn_id ELSE workbench_projection_checkpoints.turn_id END, run_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.run_id ELSE workbench_projection_checkpoints.run_id END, command_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.command_id ELSE workbench_projection_checkpoints.command_id END, projected_seq = GREATEST(workbench_projection_checkpoints.projected_seq, EXCLUDED.projected_seq), source_seq = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.source_seq ELSE workbench_projection_checkpoints.source_seq END, source_event_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.source_event_id ELSE workbench_projection_checkpoints.source_event_id END, projection_status = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.projection_status ELSE workbench_projection_checkpoints.projection_status END, projection_health = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.projection_health ELSE workbench_projection_checkpoints.projection_health END, terminal = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.terminal ELSE workbench_projection_checkpoints.terminal END, sealed = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.sealed ELSE workbench_projection_checkpoints.sealed END, diagnostic_json = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.diagnostic_json ELSE workbench_projection_checkpoints.diagnostic_json END, checkpoint_json = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.checkpoint_json ELSE workbench_projection_checkpoints.checkpoint_json END, updated_at = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.updated_at ELSE workbench_projection_checkpoints.updated_at END",
       [record.traceId, record.sessionId, record.turnId, record.runId, record.commandId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.projectionStatus, record.projectionHealth, record.terminal, record.sealed, stableJson(record.diagnostic), stableJson(record), record.createdAt, record.updatedAt]
     );
@@ -2200,6 +2262,121 @@ function normalizeWorkbenchFacts(value, requestMeta = {}, now) {
   };
 }
 
+function normalizeWorkbenchAggregateEventsForFacts(facts = {}, requestMeta = {}, now) {
+  const events = [];
+  for (const fact of arrayInput(facts.sessions)) appendWorkbenchFactEvent(events, "sessions", fact, requestMeta, now);
+  for (const fact of arrayInput(facts.messages)) appendWorkbenchFactEvent(events, "messages", fact, requestMeta, now);
+  for (const fact of arrayInput(facts.parts)) appendWorkbenchFactEvent(events, "parts", fact, requestMeta, now);
+  for (const fact of arrayInput(facts.traceEvents)) appendWorkbenchFactEvent(events, "traceEvents", fact, requestMeta, now);
+  for (const fact of arrayInput(facts.checkpoints)) appendWorkbenchFactEvent(events, "checkpoints", fact, requestMeta, now);
+  for (const fact of arrayInput(facts.turns)) appendWorkbenchFactEvent(events, "turns", fact, requestMeta, now);
+  return events;
+}
+
+function appendWorkbenchFactEvent(events, factFamily, fact, requestMeta, now) {
+  const aggregate = workbenchAggregateForFact(fact, requestMeta);
+  const factId = textOr(fact.id ?? fact.sessionId ?? fact.messageId ?? fact.partId ?? fact.turnId ?? fact.traceId, "");
+  const rawSourceEventId = textOr(fact.sourceEventId ?? fact.eventId, "") || factId;
+  const sourceEventId = rawSourceEventId ? `${factFamily}:${rawSourceEventId}` : null;
+  const eventType = workbenchAggregateEventType(factFamily, fact);
+  const projectedSeq = nonNegativeInteger(fact.projectedSeq);
+  const sourceSeq = nonNegativeInteger(fact.sourceSeq);
+  const occurredAt = timestampOr(fact.occurredAt ?? fact.updatedAt ?? fact.createdAt, now);
+  events.push(normalizeWorkbenchAggregateEvent({
+    ...aggregate,
+    factFamily,
+    factId,
+    factKey: `${factFamily.slice(0, -1) || factFamily}:${factId}`,
+    eventId: stableWorkbenchFactId("wbe", { aggregateId: aggregate.aggregateId, factFamily, sourceEventId, eventType, factId }),
+    messageId: textOr(fact.messageId, "") || null,
+    sourceRunId: textOr(fact.sourceRunId ?? fact.runId, "") || null,
+    sourceCommandId: textOr(fact.sourceCommandId ?? fact.commandId, "") || null,
+    sourceSeq,
+    sourceEventId,
+    eventType,
+    projectionRevision: projectedSeq || sourceSeq,
+    terminal: Boolean(fact.terminal),
+    sealed: Boolean(fact.sealed),
+    payload: {
+      factFamily,
+      factId,
+      status: textOr(fact.status ?? fact.projectionStatus, "") || null,
+      eventType,
+      projectedSeq,
+      sourceSeq,
+      originalSourceEventId: rawSourceEventId || null,
+      valuesRedacted: true
+    },
+    occurredAt,
+    committedAt: timestampOr(fact.updatedAt ?? occurredAt, now)
+  }, requestMeta, now));
+}
+
+function workbenchAggregateForFact(fact = {}, requestMeta = {}) {
+  const sessionId = textOr(fact.sessionId ?? requestMeta.sessionId, "") || null;
+  const turnId = textOr(fact.turnId ?? requestMeta.turnId, "") || null;
+  const traceId = textOr(fact.traceId ?? requestMeta.traceId, "") || null;
+  if (sessionId) return { aggregateId: `session:${sessionId}`, aggregateType: "session", sessionId, turnId, traceId };
+  if (turnId) return { aggregateId: `turn:${turnId}`, aggregateType: "turn", sessionId, turnId, traceId };
+  if (traceId) return { aggregateId: `trace:${traceId}`, aggregateType: "trace", sessionId, turnId, traceId };
+  throw requiredWorkbenchFactError("workbench aggregate event", ["sessionId", "turnId", "traceId"]);
+}
+
+function normalizeWorkbenchAggregateEvent(value, requestMeta = {}, now) {
+  const input = normalizeJsonObject(value);
+  const aggregateId = textOr(input.aggregateId, "");
+  if (!aggregateId) throw requiredWorkbenchFactError("workbench aggregate event", ["aggregateId"]);
+  const timestamp = timestampOr(input.committedAt ?? input.occurredAt ?? input.updatedAt ?? input.createdAt, now);
+  const sourceSeq = nonNegativeInteger(input.sourceSeq);
+  const projectionRevision = nonNegativeInteger(input.projectionRevision ?? input.projectedSeq ?? sourceSeq);
+  return pruneUndefined({
+    ...input,
+    eventSeq: nonNegativeInteger(input.eventSeq),
+    eventId: textOr(input.eventId ?? input.id, "") || stableWorkbenchFactId("wbe", { aggregateId, eventType: input.eventType, sourceEventId: input.sourceEventId, sourceSeq, projectionRevision }),
+    aggregateId,
+    aggregateType: textOr(input.aggregateType, "trace"),
+    aggregateSeq: nonNegativeInteger(input.aggregateSeq),
+    sessionId: textOr(input.sessionId ?? requestMeta.sessionId, "") || null,
+    turnId: textOr(input.turnId ?? requestMeta.turnId, "") || null,
+    traceId: textOr(input.traceId ?? requestMeta.traceId, "") || null,
+    messageId: textOr(input.messageId ?? requestMeta.messageId, "") || null,
+    sourceRunId: textOr(input.sourceRunId ?? input.runId, "") || null,
+    sourceCommandId: textOr(input.sourceCommandId ?? input.commandId, "") || null,
+    sourceSeq,
+    sourceEventId: textOr(input.sourceEventId, "") || null,
+    eventType: textOr(input.eventType ?? input.type, "event"),
+    projectionRevision,
+    terminal: Boolean(input.terminal),
+    sealed: Boolean(input.sealed),
+    payload: normalizeJsonObject(input.payload),
+    occurredAt: timestampOr(input.occurredAt, timestamp),
+    committedAt: timestamp,
+    valuesPrinted: false
+  });
+}
+
+function workbenchAggregateEventType(factFamily, fact = {}) {
+  if (factFamily === "sessions") return "session_upsert";
+  if (factFamily === "messages") return `${textOr(fact.role, "agent")}_message`;
+  if (factFamily === "parts") return `part_${textOr(fact.partType ?? fact.type, "text")}`;
+  if (factFamily === "turns") return fact.terminal || fact.sealed ? "turn_terminal" : "turn_update";
+  if (factFamily === "traceEvents") return `trace_${textOr(fact.eventType ?? fact.type, "event")}`;
+  if (factFamily === "checkpoints") return `checkpoint_${textOr(fact.projectionStatus, "projecting")}`;
+  return "event";
+}
+
+function indexWorkbenchAggregateEvents(events = []) {
+  const index = new Map();
+  for (const event of events) {
+    if (event?.factKey) index.set(event.factKey, event);
+    if (event?.factFamily && event?.factId) index.set(`${event.factFamily}:${event.factId}`, event);
+    if (event?.factFamily === "turns" && event?.turnId) index.set(`turn:${event.turnId}`, event);
+    if (event?.factFamily === "traceEvents" && event?.factId) index.set(`traceEvent:${event.factId}`, event);
+    if (event?.sourceEventId) index.set(`sourceEvent:${event.sourceEventId}`, event);
+  }
+  return index;
+}
+
 function normalizeWorkbenchSessionFact(value, requestMeta = {}, now) {
   const input = normalizeJsonObject(value);
   const sessionId = textOr(input.sessionId ?? input.id ?? requestMeta.sessionId, "");

internal/db/schema.ts

@@ -1,11 +1,11 @@
 /*
- * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0
+ * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-0104010803 Workbench唯一投影 draft-2026-06-24-p0-aggregate-event-stream.
  * 职责: Cloud runtime durable schema contract。Workbench/Code Agent 投影表必须纳入启动 readiness 和迁移一致性检查。
  */
 import { TABLES } from "../protocol/index.mjs";
 
 export const CLOUD_CORE_MIGRATION_ID = "0001_cloud_core_skeleton";
-export const CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION = "runtime-durable-postgres-v5";
+export const CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION = "runtime-durable-postgres-v6";
 export const CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE = "hwlab_schema_migrations";
 
 export const FROZEN_CLOUD_CORE_TABLES = Object.freeze([...TABLES]);
@@ -249,6 +249,38 @@ export const CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS = Object.freeze({
     "occurred_at",
     "updated_at"
   ]),
+  workbench_event_sequences: Object.freeze([
+    "aggregate_id",
+    "aggregate_type",
+    "session_id",
+    "turn_id",
+    "trace_id",
+    "last_seq",
+    "created_at",
+    "updated_at"
+  ]),
+  workbench_events: Object.freeze([
+    "event_seq",
+    "event_id",
+    "aggregate_id",
+    "aggregate_type",
+    "aggregate_seq",
+    "session_id",
+    "turn_id",
+    "trace_id",
+    "message_id",
+    "source_run_id",
+    "source_command_id",
+    "source_seq",
+    "source_event_id",
+    "event_type",
+    "projection_revision",
+    "terminal",
+    "sealed",
+    "payload_json",
+    "occurred_at",
+    "committed_at"
+  ]),
   workbench_projection_checkpoints: Object.freeze([
     "trace_id",
     "session_id",
@@ -269,6 +301,10 @@ export const CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS = Object.freeze({
   ]),
   workbench_projection_outbox: Object.freeze([
     "outbox_seq",
+    "event_seq",
+    "aggregate_id",
+    "aggregate_seq",
+    "projection_revision",
     "trace_id",
     "session_id",
     "turn_id",