From c1dfb3f91a712d286f2941ee5a3b7490a718b50a Mon Sep 17 00:00:00 2001 From: Emanuel Medina Date: Sun, 25 Oct 2026 20:04:17 +0100 Subject: [PATCH] perf(server): streamed writes no longer re-parse their payload Projection writes for message.updated and turn-item.updated parsed the payload they had just encoded, full streamed text included, only to read its timestamps. They now encode the timestamps directly with the schema the payload uses. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 95 +++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 15 ++- 2 files changed, 105 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 7f70a2ec25eee..002805171ff66 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1319,6 +1319,101 @@ it.effect("lists only active subagent IDs without loading shell data", () => ); it.layer(layerTest)("ProjectionStoreV2", (it) => { + it.effect("indexes streamed messages and turn items by the timestamps in their payloads", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread:streamed-timestamps"); + const messageId = MessageId.make("message:streamed-timestamps"); + const turnItemId = TurnItemId.make("turn-item:streamed-timestamps"); + const createdAt = DateTime.makeUnsafe("2026-09-14T08:00:00.125Z"); + const updatedAt = DateTime.makeUnsafe("2026-09-14T08:00:03.5Z"); + const text = "Streamed reply.\n\n".repeat(200); + + yield* projectionStore.apply({ + id: EventId.make("event:streamed-timestamps:message"), + type: "message.updated", + threadId, + occurredAt: updatedAt, + payload: { + id: messageId, + threadId, + runId: null, + nodeId: null, + createdBy: "agent", + creationSource: "provider", + role: "assistant", + text, + attachments: [], + streaming: true, + createdAt, + updatedAt, + }, + }); + yield* projectionStore.apply({ + id: EventId.make("event:streamed-timestamps:turn-item"), + type: "turn-item.updated", + threadId, + occurredAt: updatedAt, + payload: { + id: turnItemId, + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: null, + startedAt: createdAt, + completedAt: null, + updatedAt, + type: "assistant_message", + messageId, + text, + streaming: true, + }, + }); + + const [message] = yield* sql<{ + readonly created_at: string; + readonly updated_at: string; + readonly payload_created_at: string; + readonly payload_updated_at: string; + }>` + SELECT + created_at, + updated_at, + json_extract(payload_json, '$.createdAt') AS payload_created_at, + json_extract(payload_json, '$.updatedAt') AS payload_updated_at + FROM orchestration_v2_projection_messages + WHERE message_id = ${messageId} + `; + assert.deepEqual(message, { + created_at: DateTime.formatIso(createdAt), + updated_at: DateTime.formatIso(updatedAt), + payload_created_at: DateTime.formatIso(createdAt), + payload_updated_at: DateTime.formatIso(updatedAt), + }); + const [turnItem] = yield* sql<{ + readonly updated_at: string; + readonly payload_updated_at: string; + }>` + SELECT + updated_at, + json_extract(payload_json, '$.updatedAt') AS payload_updated_at + FROM orchestration_v2_projection_turn_items + WHERE turn_item_id = ${turnItemId} + `; + assert.deepEqual(turnItem, { + updated_at: DateTime.formatIso(updatedAt), + payload_updated_at: DateTime.formatIso(updatedAt), + }); + }), + ); + // regression: #21969 keep corrupt plan JSON in the typed read error it.effect("returns a typed read error for malformed stored plan JSON", () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index ed064325bfe16..c4cdb2e503509 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1182,6 +1182,10 @@ const encodeThreadPayload = Schema.encodeEffect( const encodeThreadUpdatedAt = Schema.encodeEffect( OrchestrationV2AppThreadJsonSchema.fields.updatedAt, ); +// Message and turn-item timestamps encode the same way inside their payloads. +// Encoding them directly avoids parsing the payload JSON, which carries the +// full streamed text, back just to read them. +const encodePayloadTimestamp = Schema.encodeEffect(Schema.DateTimeUtcFromString); const encodeRunPayload = Schema.encodeEffect(Schema.fromJsonString(OrchestrationV2RunJsonSchema)); const encodeRestartCancelledBackgroundWork = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2RunJsonSchema.fields.restartCancelledBackgroundWork), @@ -2546,7 +2550,8 @@ export const layer: Layer.Layer = } case "message.updated": { const payloadJson = yield* encodeMessagePayload(event.payload); - const payload = parseEncodedPayload(payloadJson); + const createdAt = yield* encodePayloadTimestamp(event.payload.createdAt); + const updatedAt = yield* encodePayloadTimestamp(event.payload.updatedAt); yield* sql` INSERT INTO orchestration_v2_projection_messages ( message_id, @@ -2566,8 +2571,8 @@ export const layer: Layer.Layer = ${event.payload.nodeId}, ${event.payload.role}, ${booleanInt(event.payload.streaming)}, - ${stringField(payload, "createdAt")}, - ${stringField(payload, "updatedAt")}, + ${createdAt}, + ${updatedAt}, ${payloadJson} ) ON CONFLICT(message_id) @@ -2617,7 +2622,7 @@ export const layer: Layer.Layer = } case "turn-item.updated": { const payloadJson = yield* encodeTurnItemPayload(event.payload); - const payload = parseEncodedPayload(payloadJson); + const updatedAt = yield* encodePayloadTimestamp(event.payload.updatedAt); yield* sql` INSERT INTO orchestration_v2_projection_turn_items ( turn_item_id, @@ -2644,7 +2649,7 @@ export const layer: Layer.Layer = ${event.payload.ordinal}, ${event.payload.type}, ${event.payload.status}, - ${stringField(payload, "updatedAt")}, + ${updatedAt}, ${payloadJson} ) ON CONFLICT(turn_item_id)