perf(server): streamed writes no longer re-parse their payload

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* () {

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<ProjectionStoreV2, never, SqlClient.SqlClient> =
             }
             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<ProjectionStoreV2, never, SqlClient.SqlClient> =
                 ${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<ProjectionStoreV2, never, SqlClient.SqlClient> =
             }
             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<ProjectionStoreV2, never, SqlClient.SqlClient> =
                 ${event.payload.ordinal},
                 ${event.payload.type},
                 ${event.payload.status},
-                ${stringField(payload, "updatedAt")},
+                ${updatedAt},
                 ${payloadJson}
               )
               ON CONFLICT(turn_item_id)