diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 179971019e6e3..cb9ecf8654443 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1,4 +1,4 @@ -import { assert, it } from "@effect/vitest"; +import { assert, it, vi } from "@effect/vitest"; import { EnvironmentId, EventId, @@ -1942,6 +1942,95 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + // regression: #25157 streamed payloads are not re-parsed to read their indexed timestamps + it.effect("indexes streamed payload timestamps without parsing the payload JSON", () => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const threadId = yield* addRolledBackRecoveryCandidate("streamed-payload-parse"); + const runId = RunId.make("run:streamed-payload-parse:rolled-back"); + const createdAt = DateTime.makeUnsafe("2026-01-02T03:04:05.006Z"); + const updatedAt = DateTime.makeUnsafe("2026-01-02T03:04:06.007Z"); + const text = "x".repeat(128 * 1024); + const messageId = MessageId.make("message:streamed-payload-parse"); + const itemId = TurnItemId.make("item:streamed-payload-parse"); + + const parsedPayloads = yield* Effect.acquireUseRelease( + Effect.sync(() => vi.spyOn(JSON, "parse")), + (parse) => + Effect.gen(function* () { + yield* store.apply({ + id: EventId.make("event:streamed-payload-parse:message"), + type: "message.updated", + threadId, + occurredAt: updatedAt, + payload: { + createdBy: "agent", + creationSource: "server", + id: messageId, + threadId, + runId, + nodeId: null, + role: "assistant", + text, + attachments: [], + streaming: true, + createdAt, + updatedAt, + }, + }); + yield* store.apply({ + id: EventId.make("event:streamed-payload-parse:item"), + type: "turn-item.updated", + threadId, + runId, + occurredAt: updatedAt, + payload: { + id: itemId, + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + type: "assistant_message", + status: "running", + title: "Assistant", + messageId, + text, + streaming: true, + startedAt: createdAt, + completedAt: null, + updatedAt, + }, + }); + return parse.mock.calls + .filter(([json]) => typeof json === "string" && json.includes(text)) + .map(([json]) => (JSON.parse(json) as { readonly id: string }).id); + }), + (parse) => Effect.sync(() => parse.mockRestore()), + ); + + assert.deepStrictEqual(parsedPayloads, []); + assert.deepStrictEqual( + yield* sql` + SELECT created_at, updated_at FROM orchestration_v2_projection_messages + WHERE message_id = ${messageId} + `, + [{ created_at: DateTime.formatIso(createdAt), updated_at: DateTime.formatIso(updatedAt) }], + ); + assert.deepStrictEqual( + yield* sql` + SELECT updated_at FROM orchestration_v2_projection_turn_items + WHERE turn_item_id = ${itemId} + `, + [{ updated_at: DateTime.formatIso(updatedAt) }], + ); + }), + ); + it.effect( "SQL recovery selects unfinished runless native subagents", () => runlessNativeSubagentNeedsRecovery, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index ab8d69f9a66d1..cacf0fe5688c8 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1081,6 +1081,8 @@ const encodePlanPayload = Schema.encodeEffect( const encodeTurnItemPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2TurnItemJsonSchema), ); +// The codec every *Json payload schema uses for timestamps, so index columns match payload_json. +const encodePayloadTimestamp = Schema.encodeEffect(Schema.DateTimeUtcFromString); const encodeCheckpointScopePayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2CheckpointScopeJsonSchema), ); @@ -2453,7 +2455,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, @@ -2473,8 +2476,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) @@ -2524,7 +2527,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, @@ -2551,7 +2554,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)