| author |  Martin Bergo <martin.n.bergo@gmail.com> 2026-08-15 14:05:48+02:00 |
| committer |  GitHub <noreply@github.com> 2026-08-15 12:05:48+00:00 |
| log | f075a58119f392137d699cefe5290b2aa6e55935 |
| tree | 874c2b8ddc746c6cb04df59a7998bd8b908fb3b4 |
| parent | 9d0f2fc212858290e79710b44c0635a28dd92c2f |
| signature | Signed by PGP key B5690EEEBB952194 |
perf(server): persist the wire projection for streaming tool.updated data (#6675)
Co-authored-by: mInrOz <14320143+mInrOz@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>2 files changed, 70 insertions(+), 2 deletions(-)
apps
server
src
orchestration
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.activity.test.ts+60
| ... | ... | @@ -82,3 +82,63 @@ describe("runtimeEventToActivities task progress", () => { |
| 82 | 82 | expect(usagePayload).not.toHaveProperty("status"); |
| 83 | 83 | }); |
| 84 | 84 | }); |
| 85 | describe("runtimeEventToActivities tool streaming persistence", () => { | |
| 86 | const accumulatedStdout = [ | |
| 87 | "first line of output", | |
| 88 | ...Array.from({ length: 500 }, (_, index) => `Capturing frame ${index}/9028`), | |
| 89 | ].join("\n"); | |
| 90 | const streamingData = { | |
| 91 | toolCallId: "tool-call-1", | |
| 92 | kind: "execute", | |
| 93 | command: "blender --render", | |
| 94 | rawOutput: { stdout: accumulatedStdout }, | |
| 95 | content: [{ type: "content", content: { type: "text", text: accumulatedStdout } }], | |
| 96 | }; | |
| 97 | ||
| 98 | it("persists tool.updated with the wire projection of data, not the accumulated stream", () => { | |
| 99 | const event = { | |
| 100 | ...base, | |
| 101 | type: "item.updated", | |
| 102 | eventId: EventId.make("evt-tool-streaming-updated"), | |
| 103 | payload: { | |
| 104 | itemType: "command_execution", | |
| 105 | status: "inProgress", | |
| 106 | title: "Render", | |
| 107 | detail: accumulatedStdout, | |
| 108 | data: streamingData, | |
| 109 | }, | |
| 110 | } satisfies ProviderRuntimeEvent; | |
| 111 | ||
| 112 | const activities = runtimeEventToActivities(event); | |
| 113 | ||
| 114 | expect(activities).toHaveLength(1); | |
| 115 | const payload = activities[0]?.payload as Record<string, unknown>; | |
| 116 | const data = payload.data as Record<string, unknown>; | |
| 117 | expect(payload.status).toBe("inProgress"); | |
| 118 | expect(data.toolCallId).toBe("tool-call-1"); | |
| 119 | expect(data.command).toBe("blender --render"); | |
| 120 | expect(data.rawOutput).toEqual({ content: "first line of output" }); | |
| 121 | expect(data.content).toBeUndefined(); | |
| 122 | expect(JSON.stringify(data).length).toBeLessThan(1_000); | |
| 123 | }); | |
| 124 | ||
| 125 | it("persists the full terminal payload on tool.completed", () => { | |
| 126 | const event = { | |
| 127 | ...base, | |
| 128 | type: "item.completed", | |
| 129 | eventId: EventId.make("evt-tool-streaming-completed"), | |
| 130 | payload: { | |
| 131 | itemType: "command_execution", | |
| 132 | status: "completed", | |
| 133 | title: "Render", | |
| 134 | data: streamingData, | |
| 135 | }, | |
| 136 | } satisfies ProviderRuntimeEvent; | |
| 137 | ||
| 138 | const activities = runtimeEventToActivities(event); | |
| 139 | ||
| 140 | expect(activities).toHaveLength(1); | |
| 141 | const payload = activities[0]?.payload as Record<string, unknown>; | |
| 142 | expect(payload.data).toEqual(streamingData); | |
| 143 | }); | |
| 144 | }); |
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+10-2
| ... | ... | @@ -41,6 +41,7 @@ import { |
| 41 | 41 | ProviderRuntimeIngestionService, |
| 42 | 42 | type ProviderRuntimeIngestionShape, |
| 43 | 43 | } from "../Services/ProviderRuntimeIngestion.ts"; |
| 44 | import { projectActivityPayload } from "../ActivityPayloadProjection.ts"; | |
| 44 | 45 | import { forkParked } from "../../serverActivation.ts"; |
| 45 | 46 | import { ServerSettingsService } from "../../serverSettings.ts"; |
| 46 | 47 | import { canReplaceThreadTitle } from "../threadTitles.ts"; |
| ... | ... | @@ -786,8 +787,15 @@ export function runtimeEventToActivities( |
| 786 | 787 | if (!isToolLifecycleItemType(event.payload.itemType)) { |
| 787 | 788 | return []; |
| 788 | 789 | } |
| 790 | // A streaming update's `data` carries the full tool output accumulated | |
| 791 | // so far (adapters merge state forward), and a new activity is emitted | |
| 792 | // per chunk, so persisting `data` verbatim writes O(N²) bytes per tool | |
| 793 | // call into both the event store and the projection table. No reader | |
| 794 | // needs it: ws.ts and http.ts apply `projectActivityPayload` before any | |
| 795 | // payload reaches a client. Persist the projected form for non-terminal | |
| 796 | // updates; `item.completed` below still persists the full payload. | |
| 789 | 797 | return [ |
| 790 | { | |
| 798 | projectActivityPayload({ | |
| 791 | 799 | id: event.eventId, |
| 792 | 800 | createdAt: event.createdAt, |
| 793 | 801 | tone: "tool", |
| ... | ... | @@ -805,7 +813,7 @@ export function runtimeEventToActivities( |
| 805 | 813 | }, |
| 806 | 814 | turnId: toTurnId(event.turnId) ?? null, |
| 807 | 815 | ...maybeSequence, |
| 808 | }, | |
| 816 | }), | |
| 809 | 817 | ]; |
| 810 | 818 | } |
| 811 | 819 |