| author |  Theo Browne <me@t3.gg> 2026-10-14 01:37:01-07:00 |
| committer |  GitHub <noreply@github.com> 2026-10-14 01:37:01-07:00 |
| log | 2d14185f32b6c0141a3f1192caf4a8d181cbe10f |
| tree | 3c43ace8fc6dbd183ae6692354ad8f96962f07a7 |
| parent | fd9cdaed17f615e364ee113829ec39b2bd2ed6fa |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(server): large file edits no longer store their full diff twice (#19316)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>3 files changed, 106 insertions(+), 3 deletions(-)
apps
server
src
apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts+80
| ... | @@ -29,6 +29,7 @@ import * as Deferred from "effect/Deferred"; | ... | @@ -29,6 +29,7 @@ import * as Deferred from "effect/Deferred"; |
| 29 | import * as Fiber from "effect/Fiber"; | 29 | import * as Fiber from "effect/Fiber"; |
| 30 | import * as Effect from "effect/Effect"; | 30 | import * as Effect from "effect/Effect"; |
| 31 | import * as Layer from "effect/Layer"; | 31 | import * as Layer from "effect/Layer"; |
| 32 | import * as SqlClient from "effect/sql/SqlClient"; | ||
| 32 | import * as Stream from "effect/Stream"; | 33 | import * as Stream from "effect/Stream"; |
| 33 | import * as TestClock from "effect/testing/TestClock"; | 34 | import * as TestClock from "effect/testing/TestClock"; |
| 34 | 35 | ||
| ... | @@ -58,6 +59,7 @@ const layerTestEventSink = EventSink.layer.pipe( | ... | @@ -58,6 +59,7 @@ const layerTestEventSink = EventSink.layer.pipe( |
| 58 | ); | 59 | ); |
| 59 | 60 | ||
| 60 | const layerTest = Layer.mergeAll( | 61 | const layerTest = Layer.mergeAll( |
| 62 | layerTestDatabase, | ||
| 61 | layerTestStores, | 63 | layerTestStores, |
| 62 | layerTestEventSink, | 64 | layerTestEventSink, |
| 63 | IdAllocator.layer, | 65 | IdAllocator.layer, |
| ... | @@ -1514,6 +1516,84 @@ layer("ProviderEventIngestorV2", (it) => { | ... | @@ -1514,6 +1516,84 @@ layer("ProviderEventIngestorV2", (it) => { |
| 1514 | }), | 1516 | }), |
| 1515 | ); | 1517 | ); |
| 1516 | 1518 | ||
| 1519 | it.effect("stores a bounded copy of a large file-change diff in both SQLite tables", () => | ||
| 1520 | Effect.gen(function* () { | ||
| 1521 | const now = yield* DateTime.now; | ||
| 1522 | const eventSink = yield* EventSink.EventSinkV2; | ||
| 1523 | const projectionStore = yield* ProjectionStore.ProjectionStoreV2; | ||
| 1524 | const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; | ||
| 1525 | const idAllocator = yield* IdAllocator.IdAllocatorV2; | ||
| 1526 | const threadEvent = yield* threadCreatedEvent(now); | ||
| 1527 | const providerSessionId = yield* idAllocator.allocate.providerSession({ | ||
| 1528 | providerInstanceId: modelSelection.instanceId, | ||
| 1529 | threadId: threadEvent.threadId, | ||
| 1530 | }); | ||
| 1531 | const largeDiff = `@@ -0,0 +1 @@\n${"+generated line\n".repeat(150_000)}`; | ||
| 1532 | const fileChange = ( | ||
| 1533 | id: string, | ||
| 1534 | ordinal: number, | ||
| 1535 | diffStr: string, | ||
| 1536 | ): OrchestrationV2TurnItem => ({ | ||
| 1537 | id: TurnItemId.make(id), | ||
| 1538 | threadId: threadEvent.threadId, | ||
| 1539 | runId: null, | ||
| 1540 | nodeId: null, | ||
| 1541 | providerThreadId: null, | ||
| 1542 | providerTurnId: null, | ||
| 1543 | nativeItemRef: null, | ||
| 1544 | parentItemId: null, | ||
| 1545 | ordinal, | ||
| 1546 | status: "completed", | ||
| 1547 | title: null, | ||
| 1548 | startedAt: now, | ||
| 1549 | completedAt: now, | ||
| 1550 | updatedAt: now, | ||
| 1551 | type: "file_change", | ||
| 1552 | fileName: "src/generated.ts", | ||
| 1553 | diffStr, | ||
| 1554 | }); | ||
| 1555 | const large = fileChange("turn-item:large-diff", 1, largeDiff); | ||
| 1556 | const small = fileChange("turn-item:small-diff", 2, "@@ -1 +1 @@\n-old\n+new\n"); | ||
| 1557 | |||
| 1558 | yield* eventSink.write({ events: [threadEvent] }); | ||
| 1559 | for (const turnItem of [large, small]) { | ||
| 1560 | yield* ingestor.ingestNormalized({ | ||
| 1561 | providerSessionId, | ||
| 1562 | providerInstanceId: modelSelection.instanceId, | ||
| 1563 | threadId: threadEvent.threadId, | ||
| 1564 | event: { type: "turn_item.updated", driver: CODEX_DRIVER, turnItem }, | ||
| 1565 | }); | ||
| 1566 | } | ||
| 1567 | |||
| 1568 | const sql = yield* SqlClient.SqlClient; | ||
| 1569 | const storedBytes = yield* sql<{ readonly events: number; readonly projection: number }>` | ||
| 1570 | SELECT | ||
| 1571 | (SELECT SUM(LENGTH(payload_json)) FROM orchestration_events | ||
| 1572 | WHERE stream_id = ${threadEvent.threadId} AND event_type = 'turn-item.updated') AS events, | ||
| 1573 | (SELECT SUM(LENGTH(payload_json)) FROM orchestration_v2_projection_turn_items | ||
| 1574 | WHERE thread_id = ${threadEvent.threadId}) AS projection | ||
| 1575 | `; | ||
| 1576 | const projectedLarge = yield* projectionStore.getTurnItem({ | ||
| 1577 | threadId: threadEvent.threadId, | ||
| 1578 | itemId: large.id, | ||
| 1579 | }); | ||
| 1580 | const projectedSmall = yield* projectionStore.getTurnItem({ | ||
| 1581 | threadId: threadEvent.threadId, | ||
| 1582 | itemId: small.id, | ||
| 1583 | }); | ||
| 1584 | |||
| 1585 | assert.isAbove(largeDiff.length, 2_000_000); | ||
| 1586 | assert.isBelow(storedBytes[0]!.events, 100_000); | ||
| 1587 | assert.isBelow(storedBytes[0]!.projection, 100_000); | ||
| 1588 | assert.isTrue(projectedLarge?.type === "file_change"); | ||
| 1589 | if (projectedLarge?.type !== "file_change") return; | ||
| 1590 | assert.equal(projectedLarge.fileName, "src/generated.ts"); | ||
| 1591 | assert.isBelow(projectedLarge.diffStr?.length ?? 0, 40_000); | ||
| 1592 | assert.isTrue(largeDiff.startsWith(projectedLarge.diffStr?.split("\n…")[0] ?? "-")); | ||
| 1593 | assert.deepEqual(projectedSmall, small); | ||
| 1594 | }), | ||
| 1595 | ); | ||
| 1596 | |||
| 1517 | it.effect("routes provider-owned child artifacts to their child app thread", () => | 1597 | it.effect("routes provider-owned child artifacts to their child app thread", () => |
| 1518 | Effect.gen(function* () { | 1598 | Effect.gen(function* () { |
| 1519 | const now = yield* DateTime.now; | 1599 | const now = yield* DateTime.now; |
apps/server/src/orchestration-v2/ProviderEventIngestor.ts+4-1
| ... | @@ -36,6 +36,7 @@ import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter" | ... | @@ -36,6 +36,7 @@ import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter" |
| 36 | import { makeProviderFailureTurnItem } from "@t3tools/provider-core/server/failure"; | 36 | import { makeProviderFailureTurnItem } from "@t3tools/provider-core/server/failure"; |
| 37 | import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; | 37 | import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; |
| 38 | import { stripUnservedToolOutputImageBytes } from "./toolOutputImageBytes.ts"; | 38 | import { stripUnservedToolOutputImageBytes } from "./toolOutputImageBytes.ts"; |
| 39 | import { boundFileChangeForStorage } from "./WireProjection.ts"; | ||
| 39 | 40 | ||
| 40 | export class ProviderEventNormalizeError extends Schema.TaggedError<ProviderEventNormalizeError>()( | 41 | export class ProviderEventNormalizeError extends Schema.TaggedError<ProviderEventNormalizeError>()( |
| 41 | "ProviderEventNormalizeError", | 42 | "ProviderEventNormalizeError", |
| ... | @@ -644,7 +645,9 @@ export const layer: Layer.Layer< | ... | @@ -644,7 +645,9 @@ export const layer: Layer.Layer< |
| 644 | yield* makeDomainEvent(input, { | 645 | yield* makeDomainEvent(input, { |
| 645 | type: "turn-item.updated", | 646 | type: "turn-item.updated", |
| 646 | threadId: input.event.turnItem.threadId, | 647 | threadId: input.event.turnItem.threadId, |
| 647 | payload: stripUnservedToolOutputImageBytes(input.event.turnItem), | 648 | payload: boundFileChangeForStorage( |
| 649 | stripUnservedToolOutputImageBytes(input.event.turnItem), | ||
| 650 | ), | ||
| 648 | runId: input.event.turnItem.runId, | 651 | runId: input.event.turnItem.runId, |
| 649 | nodeId: input.event.turnItem.nodeId, | 652 | nodeId: input.event.turnItem.nodeId, |
| 650 | }), | 653 | }), |
apps/server/src/orchestration-v2/WireProjection.ts+22-2
| ... | @@ -34,6 +34,26 @@ function truncateDetail( | ... | @@ -34,6 +34,26 @@ function truncateDetail( |
| 34 | return `${prefix}\n… output truncated for transport`; | 34 | return `${prefix}\n… output truncated for transport`; |
| 35 | } | 35 | } |
| 36 | 36 | ||
| 37 | /** | ||
| 38 | * Bounds the text a provider file change keeps in the event log and its | ||
| 39 | * projection. Clients never receive more than this (see projectTurnItemForWire), | ||
| 40 | * so a full diff of a large file would only grow the database and the heap that | ||
| 41 | * reads it back. Agents reading the thread over MCP get the bounded text. | ||
| 42 | */ | ||
| 43 | export function boundFileChangeForStorage(item: OrchestrationV2TurnItem): OrchestrationV2TurnItem { | ||
| 44 | if (item.type !== "file_change") return item; | ||
| 45 | const { diffStr, oldStr, newStr } = item; | ||
| 46 | const bounded = { | ||
| 47 | ...item, | ||
| 48 | ...(diffStr === undefined ? {} : { diffStr: truncateDetail(diffStr) }), | ||
| 49 | ...(oldStr === undefined ? {} : { oldStr: truncateDetail(oldStr) }), | ||
| 50 | ...(newStr === undefined ? {} : { newStr: truncateDetail(newStr) }), | ||
| 51 | }; | ||
| 52 | return bounded.diffStr === diffStr && bounded.oldStr === oldStr && bounded.newStr === newStr | ||
| 53 | ? item | ||
| 54 | : bounded; | ||
| 55 | } | ||
| 56 | |||
| 37 | function summarizeDynamicValue(value: unknown): unknown { | 57 | function summarizeDynamicValue(value: unknown): unknown { |
| 38 | let serialized: string; | 58 | let serialized: string; |
| 39 | try { | 59 | try { |
| ... | @@ -95,8 +115,8 @@ export function projectTurnItemForWire(item: OrchestrationV2TurnItem): Orchestra | ... | @@ -95,8 +115,8 @@ export function projectTurnItemForWire(item: OrchestrationV2TurnItem): Orchestra |
| 95 | }; | 115 | }; |
| 96 | } | 116 | } |
| 97 | case "file_change": { | 117 | case "file_change": { |
| 98 | // File identity and counts are enough for activity. Full diffs already | 118 | // File identity and counts are enough for activity. Full diffs come from |
| 99 | // have a dedicated read path and remain intact in persistence. | 119 | // checkpoint diffs; storage keeps only a bounded copy of this text. |
| 100 | const { diffStr, oldStr: _old, newStr: _new, ...projected } = item; | 120 | const { diffStr, oldStr: _old, newStr: _new, ...projected } = item; |
| 101 | // A failed edit stores the provider's error where the diff would be. | 121 | // A failed edit stores the provider's error where the diff would be. |
| 102 | return item.status === "failed" && diffStr?.trim() | 122 | return item.status === "failed" && diffStr?.trim() |