From 2d14185f32b6c0141a3f1192caf4a8d181cbe10f Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 01:37:01 -0700 Subject: [PATCH] fix(server): large file edits no longer store their full diff twice (#19316) Co-authored-by: Claude Opus 5.5 --- .../ProviderEventIngestor.test.ts | 80 +++++++++++++++++++ .../orchestration-v2/ProviderEventIngestor.ts | 5 +- .../src/orchestration-v2/WireProjection.ts | 24 +++++- 3 files changed, 106 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index 19232ddeeac4f0b4e40964e368ec705800b35efe..aed7ec03c739afb13cc78bd08b55ea58b14c99b8 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -29,6 +29,7 @@ import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; @@ -58,6 +59,7 @@ const layerTestEventSink = EventSink.layer.pipe( ); const layerTest = Layer.mergeAll( + layerTestDatabase, layerTestStores, layerTestEventSink, IdAllocator.layer, @@ -1514,6 +1516,84 @@ layer("ProviderEventIngestorV2", (it) => { }), ); + it.effect("stores a bounded copy of a large file-change diff in both SQLite tables", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const eventSink = yield* EventSink.EventSinkV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const threadEvent = yield* threadCreatedEvent(now); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId: threadEvent.threadId, + }); + const largeDiff = `@@ -0,0 +1 @@\n${"+generated line\n".repeat(150_000)}`; + const fileChange = ( + id: string, + ordinal: number, + diffStr: string, + ): OrchestrationV2TurnItem => ({ + id: TurnItemId.make(id), + threadId: threadEvent.threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "file_change", + fileName: "src/generated.ts", + diffStr, + }); + const large = fileChange("turn-item:large-diff", 1, largeDiff); + const small = fileChange("turn-item:small-diff", 2, "@@ -1 +1 @@\n-old\n+new\n"); + + yield* eventSink.write({ events: [threadEvent] }); + for (const turnItem of [large, small]) { + yield* ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId: threadEvent.threadId, + event: { type: "turn_item.updated", driver: CODEX_DRIVER, turnItem }, + }); + } + + const sql = yield* SqlClient.SqlClient; + const storedBytes = yield* sql<{ readonly events: number; readonly projection: number }>` + SELECT + (SELECT SUM(LENGTH(payload_json)) FROM orchestration_events + WHERE stream_id = ${threadEvent.threadId} AND event_type = 'turn-item.updated') AS events, + (SELECT SUM(LENGTH(payload_json)) FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${threadEvent.threadId}) AS projection + `; + const projectedLarge = yield* projectionStore.getTurnItem({ + threadId: threadEvent.threadId, + itemId: large.id, + }); + const projectedSmall = yield* projectionStore.getTurnItem({ + threadId: threadEvent.threadId, + itemId: small.id, + }); + + assert.isAbove(largeDiff.length, 2_000_000); + assert.isBelow(storedBytes[0]!.events, 100_000); + assert.isBelow(storedBytes[0]!.projection, 100_000); + assert.isTrue(projectedLarge?.type === "file_change"); + if (projectedLarge?.type !== "file_change") return; + assert.equal(projectedLarge.fileName, "src/generated.ts"); + assert.isBelow(projectedLarge.diffStr?.length ?? 0, 40_000); + assert.isTrue(largeDiff.startsWith(projectedLarge.diffStr?.split("\n…")[0] ?? "-")); + assert.deepEqual(projectedSmall, small); + }), + ); + it.effect("routes provider-owned child artifacts to their child app thread", () => Effect.gen(function* () { const now = yield* DateTime.now; diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index d02e78ce55bc93c64106fbd67db0cf972a6406c2..37361e7198d57faa019cf10417ffb90674b059cc 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -36,6 +36,7 @@ import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter" import { makeProviderFailureTurnItem } from "@t3tools/provider-core/server/failure"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import { stripUnservedToolOutputImageBytes } from "./toolOutputImageBytes.ts"; +import { boundFileChangeForStorage } from "./WireProjection.ts"; export class ProviderEventNormalizeError extends Schema.TaggedError()( "ProviderEventNormalizeError", @@ -644,7 +645,9 @@ export const layer: Layer.Layer< yield* makeDomainEvent(input, { type: "turn-item.updated", threadId: input.event.turnItem.threadId, - payload: stripUnservedToolOutputImageBytes(input.event.turnItem), + payload: boundFileChangeForStorage( + stripUnservedToolOutputImageBytes(input.event.turnItem), + ), runId: input.event.turnItem.runId, nodeId: input.event.turnItem.nodeId, }), diff --git a/apps/server/src/orchestration-v2/WireProjection.ts b/apps/server/src/orchestration-v2/WireProjection.ts index f4e8f7542ad400926102c9c703ffe35410daea11..c3b954c89d5b1caa0ecfb18b1c29e3ac2918595c 100644 --- a/apps/server/src/orchestration-v2/WireProjection.ts +++ b/apps/server/src/orchestration-v2/WireProjection.ts @@ -34,6 +34,26 @@ function truncateDetail( return `${prefix}\n… output truncated for transport`; } +/** + * Bounds the text a provider file change keeps in the event log and its + * projection. Clients never receive more than this (see projectTurnItemForWire), + * so a full diff of a large file would only grow the database and the heap that + * reads it back. Agents reading the thread over MCP get the bounded text. + */ +export function boundFileChangeForStorage(item: OrchestrationV2TurnItem): OrchestrationV2TurnItem { + if (item.type !== "file_change") return item; + const { diffStr, oldStr, newStr } = item; + const bounded = { + ...item, + ...(diffStr === undefined ? {} : { diffStr: truncateDetail(diffStr) }), + ...(oldStr === undefined ? {} : { oldStr: truncateDetail(oldStr) }), + ...(newStr === undefined ? {} : { newStr: truncateDetail(newStr) }), + }; + return bounded.diffStr === diffStr && bounded.oldStr === oldStr && bounded.newStr === newStr + ? item + : bounded; +} + function summarizeDynamicValue(value: unknown): unknown { let serialized: string; try { @@ -95,8 +115,8 @@ export function projectTurnItemForWire(item: OrchestrationV2TurnItem): Orchestra }; } case "file_change": { - // File identity and counts are enough for activity. Full diffs already - // have a dedicated read path and remain intact in persistence. + // File identity and counts are enough for activity. Full diffs come from + // checkpoint diffs; storage keeps only a bounded copy of this text. const { diffStr, oldStr: _old, newStr: _new, ...projected } = item; // A failed edit stores the provider's error where the diff would be. return item.status === "failed" && diffStr?.trim() -- 2.54.0