From e08df2759cc8127d02a121d1cef6fa1d91739358 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 02:35:30 +0000 Subject: [PATCH 1/2] test(server): streamed projection payloads are not re-parsed for timestamps Regression test for #25157: message.updated and turn-item.updated parse the full encoded payload JSON only to read indexed timestamps. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 92 ++++++++++++++++++- 1 file changed, 91 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 7f70a2ec25eee..a28c2f5e1019b 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, @@ -1344,6 +1344,96 @@ 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", + driver, + 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, From fd5d54523b82defd5c559f7549731ee2badb7695 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 02:40:04 +0000 Subject: [PATCH 2/2] fix(server): index streamed projection timestamps without re-parsing payloads message.updated and turn-item.updated encoded the full payload to JSON and then parsed it back only to read createdAt/updatedAt for the index columns. Encode those timestamps straight from the typed payload with the same DateTimeUtcFromString codec the payload schemas use, so the indexed values stay byte-identical to payload_json. Also drops an invalid `driver` field from the regression test's assistant_message turn item, which failed typecheck. Fixes pingdotgg/t3code#25157 Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/ProjectionStore.test.ts | 1 - apps/server/src/orchestration-v2/ProjectionStore.ts | 13 ++++++++----- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index a28c2f5e1019b..38382b0aa4d37 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1398,7 +1398,6 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { parentItemId: null, ordinal: 1, type: "assistant_message", - driver, status: "running", title: "Assistant", messageId, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index ed064325bfe16..222ed6dcc0b11 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1217,6 +1217,8 @@ const decodePlanPayload = (json: string) => decodePlanJson(json); 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), ); @@ -2546,7 +2548,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 +2569,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 +2620,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 +2647,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)