authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-08-06 02:20:21-07:00
committergravatar for me@theo.liTheo Browne <me@theo.li> 2026-08-06 08:24:08-04:00
logb7d1981b57f1c30908808d1939fd4edbc781de12
treecfe3df24301be7b147f6ac9b5b3ca841859e9dbb
parent64a3cd6d7daab21f6bfbc00faf0104b6fd361267

fix(server): drop superseded tool updates from snapshots

A tool.updated row is the in-flight snapshot of a call; once the call completes, the tool.completed row carries the final state and both clients fold every matching update into it. Shipping the updates buys nothing: 47k such rows exist in one real database, and a single thread carries 3,291 of them. Filter them out of thread snapshots, mirroring the existing context-window dedup. Matching is per turn and only against a LATER completion, so a revert that discards the completing turn cannot leave a call unrepresented, and a later update under the same identity (the next call, still in flight) survives. Live events are untouched. Rows are matched on the same identity the clients collapse by: an explicit data.toolCallId when the adapter emits one, otherwise the itemType/title/detail triple. No tool lifecycle row in the real db carries a toolCallId, so the fallback does the work. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

2 files changed, 242 insertions(+), 3 deletions(-)

apps/server/src/orchestration/ActivityPayloadProjection.ts+91-3
......@@ -365,6 +365,94 @@ function dropStaleContextWindowActivities(
365365 );
366366}
367367
368/**
369 * Identity both clients use to fold a tool lifecycle row into the call it
370 * belongs to (`deriveToolLifecycleCollapseKey` in web's `session-logic` and
371 * mobile's `threadActivity`): an explicit `data.toolCallId` when the adapter
372 * emits one, otherwise the itemType/title/detail triple. Returns null for rows
373 * with no identity at all — those never collapse on the client either, so they
374 * must not be dropped here.
375 */
376function toolLifecycleIdentity(activity: OrchestrationThreadActivity): string | null {
377 const payload = asRecord(activity.payload);
378 if (!payload) {
379 return null;
380 }
381
382 const toolCallId = asTrimmedString(asRecord(payload.data)?.toolCallId);
383 if (toolCallId) {
384 return `id:${toolCallId}`;
385 }
386
387 const itemType = asTrimmedString(payload.itemType) ?? "";
388 // Mirrors the clients' `normalizeCompactToolLabel`: a completion's title may
389 // gain a trailing "complete"/"completed" the in-flight updates lack.
390 const label = (asTrimmedString(payload.title) ?? activity.summary)
391 .replace(/\s+(?:complete|completed)\s*$/iu, "")
392 .trim();
393 const detail = asTrimmedString(payload.detail) ?? "";
394 if (itemType.length === 0 && label.length === 0 && detail.length === 0) {
395 return null;
396 }
397 return [itemType, label, detail].join("");
398}
399
400/**
401 * Drops `tool.updated` rows a `tool.completed` row already supersedes. An
402 * update is the in-flight snapshot of a call; once the call completes, the
403 * completion carries the final state and the clients fold every matching
404 * update into it, so shipping the updates buys nothing — 47k such rows exist
405 * in one real database, and a single thread carries 2,291 of them totalling
406 * ~1MB post-slimming.
407 *
408 * Matching is per turn for the same reason `dropStaleContextWindowActivities`
409 * retains per turn: a live `thread.reverted` makes the client discard whole
410 * turns, so a completion in a different turn could vanish and leave the
411 * dropped update unrepresented. The completion must also come *after* the
412 * update within the turn — a later update belongs to a subsequent call that
413 * reuses the same identity and is still in flight. Rows without a lifecycle
414 * identity pass through, matching the clients, which never collapse them.
415 * Live `thread.activity-appended` events are untouched: updates still stream
416 * in real time and the completion supersedes them on the client as before.
417 */
418function dropSupersededToolUpdatedActivities(
419 activities: ReadonlyArray<OrchestrationThreadActivity>,
420): ReadonlyArray<OrchestrationThreadActivity> {
421 const completionIndicesByKey = new Map<string, number[]>();
422 for (let index = 0; index < activities.length; index += 1) {
423 const activity = activities[index]!;
424 if (activity.kind !== "tool.completed") {
425 continue;
426 }
427 const identity = toolLifecycleIdentity(activity);
428 if (!identity) {
429 continue;
430 }
431 const key = `${activity.turnId ?? ""}${identity}`;
432 const indices = completionIndicesByKey.get(key);
433 if (indices) {
434 indices.push(index);
435 } else {
436 completionIndicesByKey.set(key, [index]);
437 }
438 }
439 if (completionIndicesByKey.size === 0) {
440 return activities;
441 }
442
443 return activities.filter((activity, index) => {
444 if (activity.kind !== "tool.updated") {
445 return true;
446 }
447 const identity = toolLifecycleIdentity(activity);
448 if (!identity) {
449 return true;
450 }
451 const indices = completionIndicesByKey.get(`${activity.turnId ?? ""}${identity}`);
452 return !indices?.some((completionIndex) => completionIndex > index);
453 });
454}
455
368456export function projectThreadDetailSnapshot(
369457 snapshot: OrchestrationThreadDetailSnapshot,
370458): OrchestrationThreadDetailSnapshot {
......@@ -372,9 +460,9 @@ export function projectThreadDetailSnapshot(
372460 ...snapshot,
373461 thread: {
374462 ...snapshot.thread,
375 activities: dropStaleContextWindowActivities(snapshot.thread.activities).map(
376 projectActivityPayload,
377 ),
463 activities: dropSupersededToolUpdatedActivities(
464 dropStaleContextWindowActivities(snapshot.thread.activities),
465 ).map(projectActivityPayload),
378466 },
379467 };
380468}
apps/server/test/ActivityPayloadProjection.test.ts+151
......@@ -257,6 +257,157 @@ describe("projectActivityPayload", () => {
257257 });
258258});
259259
260describe("superseded tool.updated snapshot dedup", () => {
261 function makeToolLifecycleActivity(
262 id: string,
263 kind: "tool.updated" | "tool.completed",
264 options: {
265 readonly turn?: string;
266 readonly title?: string;
267 readonly detail?: string;
268 readonly toolCallId?: string;
269 } = {},
270 ): OrchestrationThreadActivity {
271 const { turn = "turn-a", title = "File change", detail, toolCallId } = options;
272 return {
273 id: EventId.make(id),
274 tone: "tool",
275 kind,
276 summary: title,
277 payload: {
278 itemType: "file_change",
279 title,
280 ...(detail ? { detail } : {}),
281 data: {
282 ...(toolCallId ? { toolCallId } : {}),
283 toolName: "Edit",
284 input: { file_path: "src/app.ts" },
285 },
286 },
287 turnId: TurnId.make(turn),
288 createdAt: "2026-07-27T00:00:00.000Z",
289 };
290 }
291
292 function projectedIds(activities: ReadonlyArray<OrchestrationThreadActivity>) {
293 return projectThreadDetailSnapshot({
294 snapshotSequence: 7,
295 thread: makeThread(activities),
296 }).thread.activities.map((activity) => activity.id);
297 }
298
299 it("drops updates a later completion supersedes in the same turn", () => {
300 const update1 = makeToolLifecycleActivity("upd-1", "tool.updated");
301 const update2 = makeToolLifecycleActivity("upd-2", "tool.updated");
302 const completed = makeToolLifecycleActivity("done-1", "tool.completed");
303
304 expect(projectedIds([update1, update2, completed])).toEqual([completed.id]);
305 });
306
307 it("matches on toolCallId when the adapter emits one", () => {
308 const otherCall = makeToolLifecycleActivity("upd-other", "tool.updated", {
309 toolCallId: "call-b",
310 });
311 const update = makeToolLifecycleActivity("upd-a", "tool.updated", { toolCallId: "call-a" });
312 const completed = makeToolLifecycleActivity("done-a", "tool.completed", {
313 toolCallId: "call-a",
314 });
315
316 // Same itemType/title, different call: only call-a's update is superseded.
317 expect(projectedIds([otherCall, update, completed])).toEqual([otherCall.id, completed.id]);
318 });
319
320 it("keeps updates with no matching completion", () => {
321 const inFlight = makeToolLifecycleActivity("upd-live", "tool.updated", { title: "Running" });
322 const other = makeToolLifecycleActivity("upd-other", "tool.updated", { title: "Reading" });
323 const completed = makeToolLifecycleActivity("done-other", "tool.completed", {
324 title: "Reading",
325 });
326
327 expect(projectedIds([inFlight, other, completed])).toEqual([inFlight.id, completed.id]);
328 });
329
330 it("keeps an update whose completion lives in another turn", () => {
331 // A live thread.reverted can discard the completing turn while keeping
332 // the updating one, which would leave the call unrepresented.
333 const update = makeToolLifecycleActivity("upd-kept", "tool.updated", { turn: "turn-kept" });
334 const completed = makeToolLifecycleActivity("done-later", "tool.completed", {
335 turn: "turn-reverted",
336 });
337
338 expect(projectedIds([update, completed])).toEqual([update.id, completed.id]);
339 });
340
341 it("keeps an update that follows its completion", () => {
342 // A later update under the same identity is the next call, still in flight.
343 const completed = makeToolLifecycleActivity("done-first", "tool.completed");
344 const nextCall = makeToolLifecycleActivity("upd-next", "tool.updated");
345
346 expect(projectedIds([completed, nextCall])).toEqual([completed.id, nextCall.id]);
347 });
348
349 it("keeps identity-less rows the clients never collapse", () => {
350 const anonymous: OrchestrationThreadActivity = {
351 id: EventId.make("upd-anon"),
352 tone: "tool",
353 kind: "tool.updated",
354 summary: " ",
355 payload: { data: { toolName: "Edit" } },
356 turnId: TurnId.make("turn-a"),
357 createdAt: "2026-07-27T00:00:00.000Z",
358 };
359 const completed: OrchestrationThreadActivity = {
360 ...anonymous,
361 id: EventId.make("done-anon"),
362 kind: "tool.completed",
363 };
364
365 expect(projectedIds([anonymous, completed])).toEqual([anonymous.id, completed.id]);
366 });
367
368 it("does not filter live activity-appended events", () => {
369 const update = makeToolLifecycleActivity("upd-live-event", "tool.updated");
370 const event = {
371 sequence: 11,
372 eventId: EventId.make("event-tool-updated"),
373 aggregateKind: "thread",
374 aggregateId: ThreadId.make("thread-projection"),
375 occurredAt: "2026-07-27T00:00:03.000Z",
376 commandId: null,
377 causationEventId: null,
378 correlationId: null,
379 metadata: {},
380 type: "thread.activity-appended",
381 payload: {
382 threadId: ThreadId.make("thread-projection"),
383 activity: update,
384 },
385 } satisfies Extract<OrchestrationEvent, { type: "thread.activity-appended" }>;
386
387 const projected = projectActivityEvent(event);
388 expect(
389 projected.type === "thread.activity-appended" ? projected.payload.activity.id : undefined,
390 ).toEqual(update.id);
391 });
392
393 it("leaves the collapsed work log identical to the full history", () => {
394 const activities = [
395 makeToolLifecycleActivity("upd-1", "tool.updated", { detail: "writing" }),
396 makeToolLifecycleActivity("upd-2", "tool.updated", { detail: "writing" }),
397 makeToolLifecycleActivity("done-1", "tool.completed", { detail: "writing" }),
398 ];
399 const projected = projectThreadDetailSnapshot({
400 snapshotSequence: 7,
401 thread: makeThread(activities),
402 });
403
404 const before = deriveWorkLogEntries(activities);
405 const after = deriveWorkLogEntries(projected.thread.activities);
406 expect(after).toHaveLength(before.length);
407 expect(after.map((entry) => entry.label)).toEqual(before.map((entry) => entry.label));
408 });
409});
410
260411describe("context-window snapshot dedup", () => {
261412 function makeContextWindowActivity(
262413 id: string,