diff --git a/internal/db/migrations/0001_cloud_core_skeleton.sql b/internal/db/migrations/0001_cloud_core_skeleton.sql index 75bcf7d8..de88b334 100644 --- a/internal/db/migrations/0001_cloud_core_skeleton.sql +++ b/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"}' ) diff --git a/internal/db/runtime-store.test.ts b/internal/db/runtime-store.test.ts index 4409056b..ad055157 100644 --- a/internal/db/runtime-store.test.ts +++ b/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") diff --git a/internal/db/runtime-store.ts b/internal/db/runtime-store.ts index a707e84a..ff35d43d 100644 --- a/internal/db/runtime-store.ts +++ b/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, ""); diff --git a/internal/db/schema.ts b/internal/db/schema.ts index ac93ad39..a9ca6095 100644 --- a/internal/db/schema.ts +++ b/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",