diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 7dc248e6..938986f5 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -240,6 +240,19 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option outboxTraceSnapshots.add(rowTraceId); await writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "outbox-session" }); } + if (row.commitType === "message") { + await writeMessageRealtimeSnapshotSafe({ + writeEvent, + readModel, + actor: auth.actor, + sessionId: rowSessionId, + threadId: streamThreadId, + traceId: rowTraceId, + reason: "outbox-message", + cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq } + }); + continue; + } if (row.commitType === "terminal") { writeEvent("workbench.trace.snapshot", { type: "trace.snapshot", diff --git a/internal/db/runtime-store.ts b/internal/db/runtime-store.ts index ff35d43d..d0011c1e 100644 --- a/internal/db/runtime-store.ts +++ b/internal/db/runtime-store.ts @@ -1315,6 +1315,27 @@ export class PostgresCloudRuntimeStore { 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.messages) { + const event = eventIndex.get(`message:${fact.messageId}`) ?? eventIndex.get(`messages:${fact.messageId}`) ?? 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: "message", + terminal: Boolean(fact.terminal), + sealed: Boolean(fact.sealed), + payload: { role: fact.role, status: fact.status, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null, projectedSeq: fact.projectedSeq, valuesRedacted: true }, + createdAt: fact.updatedAt ?? fact.createdAt ?? this.now() + }, client); + } for (const fact of facts.traceEvents) { const event = eventIndex.get(`traceEvent:${fact.id}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null; await this.persistWorkbenchProjectionOutbox({