diff --git a/apps/mobile/src/features/settings/SettingsRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsRouteScreen.tsx index 87537c2236c6..cbcc949f7f93 100644 --- a/apps/mobile/src/features/settings/SettingsRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsRouteScreen.tsx @@ -19,7 +19,6 @@ import { Platform, View } from "react-native"; import { useWorkspaceSurface } from "../workspace/useWorkspaceSurface"; import { useDeviceType } from "../../native/native-layout-metrics"; import { useCompactWorkspaceTabs } from "../workspace/useSystemWorkspaceTabBar"; -import { NativeStackScreenOptions } from "../../native/StackHeader"; import { deriveProjectGroupLabel } from "@t3tools/client-runtime/state/project-grouping"; import { useSafeAreaInsets } from "react-native-safe-area-context"; @@ -47,15 +46,6 @@ export function SettingsRouteScreen() { return ( - {Platform.OS === "ios" ? ( - - ) : null} {Platform.OS === "android" ? ( diff --git a/apps/mobile/src/features/settings/components/SettingsScreen.tsx b/apps/mobile/src/features/settings/components/SettingsScreen.tsx index 7eb807ec86c8..db596347e771 100644 --- a/apps/mobile/src/features/settings/components/SettingsScreen.tsx +++ b/apps/mobile/src/features/settings/components/SettingsScreen.tsx @@ -8,7 +8,6 @@ import { ScreenContent as SettingsScreenContent } from "../../../components/ui/S import { useWorkspaceTabs } from "../../workspace/WorkspaceTabsContext"; import { useWorkspaceSurface } from "../../workspace/useWorkspaceSurface"; import { useDeviceType } from "../../../native/native-layout-metrics"; -import { NativeStackScreenOptions } from "../../../native/StackHeader"; import { useCompactWorkspaceTabs } from "../../workspace/useSystemWorkspaceTabBar"; import { ScreenScrollViewBottomInset } from "../../../components/ui/ScreenScrollView"; @@ -47,15 +46,6 @@ export function SettingsScreen( onBack={workspaceRoot ? undefined : () => navigation.goBack()} hideBottomBorder={!props.formSheet} /> - {Platform.OS === "ios" && (workspaceRoot || cover) ? ( - - ) : null} {props.formSheet ? ( props.children diff --git a/apps/mobile/src/features/usage/UsageRouteScreen.tsx b/apps/mobile/src/features/usage/UsageRouteScreen.tsx index 5e9ca9f9afa9..a9c806f7ff58 100644 --- a/apps/mobile/src/features/usage/UsageRouteScreen.tsx +++ b/apps/mobile/src/features/usage/UsageRouteScreen.tsx @@ -56,6 +56,7 @@ import { UsageLimitsSection } from "./UsageLimitsPooled"; import { Menu } from "../../components/ui/Menu"; import { SymbolView } from "../../components/ui/AppSymbol"; import { NativeStackScreenOptions } from "../../native/StackHeader"; +import { withNativeGlassHeaderItem } from "../layout/native-glass-header-items"; import type { UsageChartMetric } from "./usageChartData"; import { PROVIDER_LABEL, useProviderColors, useUsageMixColors } from "./usageProviders"; @@ -245,8 +246,31 @@ function UsageScreen(props: { {Platform.OS === "ios" ? ( environmentFilter }} + optionsVersion={JSON.stringify({ showEnvironmentFilter, filterIcon, environmentActions })} + options={{ + // A native menu item, unlike a custom view, moves into the Duo's vertical bar. + unstable_headerRightItems: () => + showEnvironmentFilter + ? [ + withNativeGlassHeaderItem({ + accessibilityLabel: "Filter usage environments", + icon: { name: filterIcon, type: "sfSymbol" }, + label: "", + type: "menu", + menu: { + title: "Environments", + items: environmentActions.map((action) => ({ + type: "action" as const, + label: action.title, + description: action.subtitle, + state: action.state === "on" ? ("on" as const) : undefined, + onPress: () => selectEnvironment(action.id), + })), + }, + }), + ] + : [], + }} /> ) : null} - + ); diff --git a/apps/mobile/src/features/workspace/WorkspaceRootScreen.tsx b/apps/mobile/src/features/workspace/WorkspaceRootScreen.tsx index 4e9c0d1d137c..2ca3fc7cdf8c 100644 --- a/apps/mobile/src/features/workspace/WorkspaceRootScreen.tsx +++ b/apps/mobile/src/features/workspace/WorkspaceRootScreen.tsx @@ -93,7 +93,6 @@ function WorkspaceRootContent() { const keyboardHeight = useKeyboardState((keyboard) => keyboard.height); const workspace = useWorkspaceTabs()!; const { state } = workspace; - const solidHeader = deviceType === "duo-folded" || state.selectedTab === "usage"; const renderHeaderTitle = useCallback( () => ( [] } : {}), diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 5c6c4fece44b..2e0d16c535c8 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -816,9 +816,10 @@ const handoffTargetIds = ( }; /** - * Runs a tool registration against a server that records one anonymous - * `mcp.tool.invoked` event per call: the tool, the calling agent's provider, - * the outcome with its failure code, and the duration. A handoff tool also + * Runs a tool registration against a server that records an anonymous + * `mcp.tool.invoked` event for every handoff call and every failed call of + * other tools: the tool, the calling agent's provider, the outcome with its + * failure code, and the duration. A handoff tool also * reports the caller's model, origin, and delegation depth, the settings the * agent chose, and the provider, model, and modes of each thread that received * the work. Ids, prompts, and free text are never recorded. @@ -882,6 +883,11 @@ const withToolAnalytics = result: McpSchema.CallToolResult | undefined, durationMs: number, ) { + const handoff = HANDOFF_TOOLS.has(tool); + const outcome = toolOutcome(result); + // Agents call status, read, and preview tools in tight loops, so an + // event per successful call floods analytics. Only failures count there. + if (!handoff && outcome.outcome === "ok") return; // Client credentials (no calling thread) report an unknown caller. const invocation = Option.getOrUndefined( yield* Effect.serviceOption(McpInvocationContext.McpInvocationContext), @@ -889,7 +895,6 @@ const withToolAnalytics = const providers = Option.isSome(registry) ? yield* registry.value.getProviders : []; // Only handoff tools pay for thread lookups; the rest report the // caller's provider from the credential. - const handoff = HANDOFF_TOOLS.has(tool); const caller = handoff && invocation !== undefined ? ((yield* shellOf(invocation.threadId)) ?? undefined) @@ -917,7 +922,7 @@ const withToolAnalytics = ), }), targets: targets.filter((thread) => thread !== null), - outcome: toolOutcome(result), + outcome, ...(handoff ? { settings: handoffSettings(tool, args, result?.structuredContent) } : {}), diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 02216cec576d..1e4abf2062e4 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -18,6 +18,7 @@ import { type OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as TestClock from "effect/testing/TestClock"; import * as Layer from "effect/Layer"; @@ -48,6 +49,7 @@ import * as ServerSettings from "../../serverSettings.ts"; import * as McpHttpServer from "../McpHttpServer.ts"; import * as McpInvocationContext from "../McpInvocationContext.ts"; import * as McpToolAccessTestkit from "../McpToolAccess.testkit.ts"; +import * as AnalyticsService from "../../telemetry/AnalyticsService.ts"; import { dispatchFailure } from "../threadAccess.ts"; import * as EnvironmentHandlers from "./environment/handlers.ts"; import * as AttachmentHandlers from "./attachment/handlers.ts"; @@ -287,6 +289,61 @@ it.effect("returns an HTML render reference that Codex and Claude tool rows both ), ); +it.effect("records only failed calls of tools that hand off no work", () => { + // Recording runs detached from the tool response, so the first event is awaited. + const recorded = Deferred.makeUnsafe<{ event: string; properties: object }>(); + return Effect.gen(function* () { + const server = yield* McpServer.McpServer; + const organize = (capabilities: McpInvocationContext.McpInvocationScope["capabilities"]) => + server.callTool({ name: "t3_thread_organize", arguments: { action: "pin" } }).pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, { + ...scope, + capabilities, + }), + Effect.provideService(McpSchema.McpServerClient, client), + ); + expect((yield* organize(scope.capabilities)).isError).not.toBe(true); + expect(declaredFailure(yield* organize(new Set()))).toMatchObject({ + code: "capability_denied", + }); + // The successful call ran first, so it would be the first event if it were recorded. + expect(yield* Deferred.await(recorded)).toMatchObject({ + event: "mcp.tool.invoked", + properties: { tool: "t3_thread_organize", outcome: "error" }, + }); + }).pipe( + Effect.provide( + layerThreadToolkit.pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provide(NodeCrypto.layer), + Layer.provide( + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (id) => Effect.succeed(McpToolAccessTestkit.liveThreadShell(id)), + getProjectThreadRecords: ({ threadId: id }) => + Effect.succeed( + McpToolAccessTestkit.idleThreadProjection(McpToolAccessTestkit.liveThreadShell(id)), + ), + dispatch: () => Effect.succeed({ sequence: 1 } as never), + }), + ), + Layer.provide( + Layer.succeed( + AnalyticsService.AnalyticsService, + AnalyticsService.AnalyticsService.of({ + record: (event, properties) => + Deferred.succeed(recorded, { event, properties: properties ?? {} }).pipe( + Effect.asVoid, + ), + flush: Effect.void, + recordOwners: () => Effect.void, + }), + ), + ), + ), + ), + ); +}); + it.effect("keeps unexpected handler defects private through the production registration", () => Effect.gen(function* () { const server = yield* McpServer.McpServer; diff --git a/apps/server/src/orchestration-v2/ThreadLiveEventCoalescer.test.ts b/apps/server/src/orchestration-v2/ThreadLiveEventCoalescer.test.ts index de8fee88508b..05b3af1a36dd 100644 --- a/apps/server/src/orchestration-v2/ThreadLiveEventCoalescer.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLiveEventCoalescer.test.ts @@ -1,10 +1,14 @@ import { EventId, MessageId, + ProjectId, + ProviderInstanceId, + type OrchestrationV2ThreadProjection, ThreadId, RunId, TurnItemId, type OrchestrationV2ThreadStreamItem, + type OrchestrationV2TurnItem, } from "@t3tools/contracts"; import { it } from "@effect/vitest"; import * as DateTime from "effect/DateTime"; @@ -13,11 +17,13 @@ import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { describe, expect } from "vite-plus/test"; import { - coalesceLiveToolUpdatedEvents, + coalesceLiveUpdatedEvents, coalesceThreadLiveStream, makeThreadLiveEventCoalescer, } from "./ThreadLiveEventCoalescer.ts"; +import { applyOrchestrationV2ProjectionEvent } from "../../../../packages/client-runtime/src/state/orchestrationV2Projection.ts"; + type ThreadEvent = Extract; const threadId = ThreadId.make("thread-coalescer-test"); const runId = RunId.make("run-coalescer-test"); @@ -94,15 +100,69 @@ function makeMessage(sequence: number, text = "Still working"): ThreadEvent { }; } +function makeStreamingMessage( + sequence: number, + options: { + readonly id?: string; + readonly text?: string; + readonly streaming?: boolean; + } = {}, +): ThreadEvent { + const message = makeMessage(sequence, options.text ?? `text-${sequence}`); + if (message.event.type !== "message.updated") throw new Error("Expected message"); + return { + ...message, + event: { + ...message.event, + payload: { + ...message.event.payload, + id: MessageId.make(options.id ?? "message"), + streaming: options.streaming ?? true, + }, + }, + }; +} + +function makeTextItem( + sequence: number, + options: { + readonly id?: string; + readonly text?: string; + readonly type?: "assistant_message" | "reasoning"; + readonly status?: OrchestrationV2TurnItem["status"]; + readonly streaming?: boolean; + } = {}, +): ThreadEvent { + const tool = makeToolActivity(sequence); + if (tool.event.type !== "turn-item.updated" || tool.event.payload.type !== "command_execution") + throw new Error("Expected command item"); + const { input: _input, output: _output, ...base } = tool.event.payload; + return { + ...tool, + event: { + ...tool.event, + payload: { + ...base, + id: TurnItemId.make(options.id ?? "item"), + type: options.type ?? "assistant_message", + messageId: MessageId.make("message"), + text: options.text ?? `text-${sequence}`, + status: options.status ?? "running", + streaming: options.streaming ?? true, + }, + }, + }; +} + describe("ThreadLiveEventCoalescer", () => { - it("coalesces only calls with a stable toolCallId", () => { + it("preserves tool-call identity switches as insertion-order boundaries", () => { const events = [ makeToolActivity(1, { toolCallId: "call-a" }), makeToolActivity(2, { toolCallId: "call-b" }), makeToolActivity(3, { toolCallId: "call-a" }), ]; - expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3]); + expect(coalesceLiveUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2, 3]); }); it("does not coalesce stable tool calls across turns", () => { @@ -111,7 +171,7 @@ describe("ThreadLiveEventCoalescer", () => { makeToolActivity(2, { turnId: RunId.make("turn-new") }), ]; - expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2]); + expect(coalesceLiveUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2]); }); it("flushes a stable update run before a completion boundary", () => { @@ -122,9 +182,247 @@ describe("ThreadLiveEventCoalescer", () => { makeToolActivity(4), ]; - expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3, 4]); + expect(coalesceLiveUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3, 4]); + }); + + it("keeps the latest full text in sequence order and preserves state after every survivor cursor", () => { + const events = [ + makeStreamingMessage(1, { id: "shared", text: "H" }), + makeTextItem(2, { id: "shared", text: "H" }), + makeTextItem(3, { id: "reasoning", type: "reasoning", text: "Think" }), + makeToolActivity(4), + makeTextItem(5, { id: "unique", text: "Keep me" }), + makeStreamingMessage(6, { id: "shared", text: "Hello" }), + makeTextItem(7, { id: "shared", text: "Hello" }), + makeTextItem(8, { id: "reasoning", type: "reasoning", text: "Thinking done" }), + makeToolActivity(9), + makeStreamingMessage(10, { id: "shared", streaming: false, text: "Hello!" }), + makeTextItem(11, { id: "shared", streaming: false, status: "completed", text: "Hello!" }), + ]; + const survivors = coalesceLiveUpdatedEvents(events); + expect(survivors).toEqual(events); + + const latestStates = (input: ReadonlyArray) => + Object.fromEntries( + input.map(({ event }) => [ + `${event.type}:${"id" in event.payload ? event.payload.id : ""}`, + event.payload, + ]), + ); + for (let index = 0; index < survivors.length; index += 1) { + const received = survivors.slice(0, index + 1); + const cursor = received.at(-1)!.sequence; + const resumed = coalesceLiveUpdatedEvents(events.filter((event) => event.sequence > cursor)); + expect(latestStates([...received, ...resumed])).toEqual(latestStates(events)); + } }); + it("preserves assistant-message identity changes as insertion-order boundaries", () => { + const events = [ + makeStreamingMessage(1, { id: "message-a" }), + makeTextItem(2, { id: "item-a" }), + makeStreamingMessage(3, { id: "message-b" }), + makeTextItem(4, { id: "item-b" }), + makeStreamingMessage(5, { id: "message-a" }), + makeTextItem(6, { id: "item-a" }), + ]; + expect(coalesceLiveUpdatedEvents(events)).toEqual(events); + }); + + it.effect( + "preserves turn-item insertion and partial timeline state after every survivor cursor", + () => + Effect.gen(function* () { + const item = ( + sequence: number, + id: string, + text: string, + completed = false, + ): ThreadEvent => { + const input = makeTextItem(sequence, { + id, + text, + type: "reasoning", + status: completed ? "completed" : "running", + streaming: !completed, + }); + if (input.event.type !== "turn-item.updated") throw new Error("Expected turn item"); + return { + ...input, + event: { + ...input.event, + runId: undefined, + payload: { + ...input.event.payload, + runId: null, + ordinal: id === "a" ? 1 : 2, + completedAt: completed ? now : null, + }, + }, + }; + }; + const events = [ + item(1, "a", "A"), + item(2, "b", "B"), + item(3, "a", "AA"), + item(4, "a", "Final A", true), + item(5, "b", "Final B", true), + ]; + const emptyProjection = { + thread: { + id: threadId, + projectId: ProjectId.make("project-reducer"), + title: "Reducer", + providerInstanceId: ProviderInstanceId.make("codex"), + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null }, + forkedFrom: null, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + runs: [], + attempts: [], + nodes: [], + subagents: [], + providerSessions: [], + providerThreads: [], + providerTurns: [], + runtimeRequests: [], + messages: [], + plans: [], + turnItems: [], + checkpointScopes: [], + checkpoints: [], + contextHandoffs: [], + contextTransfers: [], + visibleTurnItems: [], + updatedAt: now, + } satisfies OrchestrationV2ThreadProjection; + const apply = (inputs: ReadonlyArray, partialTimeline: boolean) => { + let projection: OrchestrationV2ThreadProjection = emptyProjection; + let latestLocalTurnOrdinal = 0; + for (const { event } of inputs) { + const next = applyOrchestrationV2ProjectionEvent(projection, event, { + partialTimeline, + latestLocalTurnOrdinal, + }); + if (next === null || next === projection) continue; + projection = next; + if (partialTimeline && event.type === "turn-item.updated") { + latestLocalTurnOrdinal = Math.max(latestLocalTurnOrdinal, event.payload.ordinal); + } + } + return projection; + }; + const survivors = yield* Stream.fromIterable(events).pipe( + coalesceThreadLiveStream, + Stream.filter((input): input is ThreadEvent => input.kind === "event"), + Stream.runCollect, + ); + for (const partialTimeline of [true, false]) { + const expected = apply(events, partialTimeline); + expect( + expected.turnItems.map((item) => [item.id, "text" in item ? item.text : null]), + ).toEqual([ + ["a", "Final A"], + ["b", "Final B"], + ]); + expect(apply(survivors, partialTimeline)).toEqual(expected); + for (const [index, survivor] of survivors.entries()) { + const received = survivors.slice(0, index + 1); + const replay = yield* Stream.fromIterable( + events.filter((event) => event.sequence > survivor.sequence), + ).pipe( + coalesceThreadLiveStream, + Stream.filter((input): input is ThreadEvent => input.kind === "event"), + Stream.runCollect, + ); + expect(apply([...received, ...replay], partialTimeline)).toEqual(expected); + } + } + }), + ); + + it.effect("flushes text at synchronization and completion boundaries", () => + Effect.scoped( + Effect.gen(function* () { + const coalescer = yield* makeThreadLiveEventCoalescer(); + const events = [ + makeStreamingMessage(1), + makeTextItem(2, { type: "reasoning" }), + makeStreamingMessage(3), + makeTextItem(4, { type: "reasoning" }), + { kind: "synchronized" } as const, + makeStreamingMessage(5), + makeTextItem(6, { type: "reasoning" }), + makeStreamingMessage(7, { streaming: false, text: "Final answer" }), + makeTextItem(8, { + type: "reasoning", + streaming: false, + status: "completed", + text: "Final answer", + }), + makeStreamingMessage(9), + makeTextItem(10, { type: "reasoning" }), + ]; + yield* coalescer.offerAll(events); + yield* coalescer.end; + expect(yield* Stream.runCollect(coalescer.stream)).toEqual(events.slice(2)); + expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 }); + }), + ), + ); + + it.effect( + "flushes growing text on the fixed window and charges only unacknowledged survivors", + () => + Effect.scoped( + Effect.gen(function* () { + const coalescer = yield* makeThreadLiveEventCoalescer(); + const first = [makeStreamingMessage(1, { text: "é" }), makeTextItem(2, { text: "é" })]; + const latest = [makeStreamingMessage(3, { text: "éé" }), makeTextItem(4, { text: "éé" })]; + const serializedBytes = (events: ReadonlyArray) => + events.reduce((sum, event) => sum + Buffer.byteLength(encodeEvent(event)), 0); + yield* coalescer.offerAll(first); + yield* TestClock.adjust("40 millis"); + yield* coalescer.offerAll(latest); + expect(yield* coalescer.usage).toEqual({ + retainedItems: 4, + retainedSerializedBytes: serializedBytes([...first, ...latest]), + }); + yield* TestClock.adjust("10 millis"); + const expectedUsage = { + retainedItems: 2, + retainedSerializedBytes: serializedBytes(latest), + }; + expect(yield* coalescer.usage).toEqual(expectedUsage); + const pull = yield* Stream.toPull(coalescer.stream); + expect(yield* pull).toEqual(latest); + expect(yield* coalescer.usage).toEqual(expectedUsage); + + const next = [makeStreamingMessage(5), makeTextItem(6)]; + yield* coalescer.offerAll(next); + yield* TestClock.adjust("50 millis"); + expect(yield* pull).toEqual(next); + yield* coalescer.end; + yield* pull.pipe(Effect.result); + expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 }); + }), + ), + ); + it.effect("keeps the flush timer alive when an offer's shorter scope closes", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ThreadLiveEventCoalescer.ts b/apps/server/src/orchestration-v2/ThreadLiveEventCoalescer.ts index 00340e2e06fd..8356838b1500 100644 --- a/apps/server/src/orchestration-v2/ThreadLiveEventCoalescer.ts +++ b/apps/server/src/orchestration-v2/ThreadLiveEventCoalescer.ts @@ -21,63 +21,59 @@ const MAX_PENDING_UPDATES = 512; type ThreadLiveEvent = Extract; export type ThreadLiveInput = ThreadLiveEvent | { readonly kind: "synchronized" }; -function isToolUpdated(input: ThreadLiveEvent): boolean { - if (input.event.type !== "turn-item.updated" || input.event.payload.status !== "running") - return false; - switch (input.event.payload.type) { +function coalescibleUpdateKey(input: ThreadLiveEvent): string | null { + const { event } = input; + const prefix = `${event.threadId}\u0000${event.runId ?? ""}\u0000${event.type}`; + if (event.type === "message.updated") { + return event.payload.role === "assistant" && event.payload.streaming + ? `${prefix}\u0000${event.payload.id}` + : null; + } + if (event.type !== "turn-item.updated" || event.payload.status !== "running") return null; + switch (event.payload.type) { + case "assistant_message": + case "reasoning": + if (!event.payload.streaming) return null; + break; case "command_execution": case "file_change": case "file_search": case "web_search": case "dynamic_tool": - return true; + break; default: - return false; + return null; } + return `${prefix}\u0000${event.payload.type}\u0000${event.payload.id}`; } -function stableToolCallIdentity(input: ThreadLiveEvent): string | null { - return input.event.type === "turn-item.updated" ? input.event.payload.id : null; -} - -export function coalesceLiveToolUpdatedEvents( +// These events replace whole entities. Keep each survivor at its original +// position/sequence so a resume cursor cannot skip an unrelated earlier update. +export function coalesceLiveUpdatedEvents( events: ReadonlyArray, ): ReadonlyArray { - const survivors: Array = []; - let pendingUpdates: Array = []; - - const flushUpdates = () => { - const seen = new Set(); - const latestUpdates: Array = []; - for (let index = pendingUpdates.length - 1; index >= 0; index -= 1) { - const event = pendingUpdates[index]!; - const identity = stableToolCallIdentity(event); - const key = identity - ? `${event.event.threadId}\u0000${event.event.runId ?? ""}\u0000${identity}` - : null; - if (key && seen.has(key)) { - continue; - } - if (key) { - seen.add(key); - } - latestUpdates.push(event); - } - latestUpdates.reverse(); - survivors.push(...latestUpdates); - pendingUpdates = []; - }; - - for (const event of events) { - if (isToolUpdated(event)) { - pendingUpdates.push(event); + const latestUpdates = new Map(); + const superseded = new Set(); + const pendingKeys = new Map(); + + for (const [index, event] of events.entries()) { + const key = coalescibleUpdateKey(event); + if (key === null) { + latestUpdates.clear(); + pendingKeys.clear(); continue; } - flushUpdates(); - survivors.push(event); + // Both entity families use insertion order; partial timelines also reject + // unseen items below their watermark. Keep identity switches within each + // family, while allowing paired message/item updates to coalesce. + const pendingKey = pendingKeys.get(event.event.type); + if (pendingKey !== undefined && pendingKey !== key) latestUpdates.clear(); + pendingKeys.set(event.event.type, key); + const previous = latestUpdates.get(key); + if (previous !== undefined) superseded.add(previous); + latestUpdates.set(key, index); } - flushUpdates(); - return survivors; + return events.filter((_, index) => !superseded.has(index)); } export const makeThreadLiveEventCoalescer = Effect.fn( @@ -116,7 +112,7 @@ export const makeThreadLiveEventCoalescer = Effect.fn( } const items = yield* budget.replace( pendingUpdates, - coalesceLiveToolUpdatedEvents(pendingUpdates.map((item) => item.value)), + coalesceLiveUpdatedEvents(pendingUpdates.map((item) => item.value)), ); pendingUpdates = []; yield* Queue.offerAll(output, items); @@ -156,7 +152,7 @@ export const makeThreadLiveEventCoalescer = Effect.fn( Effect.uninterruptible, ); } - if (input.kind === "event" && isToolUpdated(input)) { + if (input.kind === "event" && coalescibleUpdateKey(input) !== null) { if (pendingUpdates.length === 1) { const generation = ++windowGeneration; windowFiber = yield* Effect.forkIn(flushWindow(generation), coalescerScope);