From 228a4a3e5e5fb4adf1b160af6d1dbec610aab422 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Mon, 5 Oct 2026 15:28:58 -0700 Subject: [PATCH] perf: cheaper shell refreshes, one copy of Codex streaming text, no MCP wait polling Co-Authored-By: Claude Opus 5.5 (1M context) --- apps/server/src/mcp/OrchestratorMcpService.ts | 47 +++++++-- .../Adapters/CodexAdapterV2.test.ts | 54 ++++++++++- .../Adapters/CodexAdapterV2.ts | 15 ++- .../src/orchestration-v2/Orchestrator.ts | 2 + .../orchestration-v2/ProjectionStore.test.ts | 15 +++ .../src/orchestration-v2/ProjectionStore.ts | 53 ++++++++--- .../src/orchestration-v2/ShellStream.test.ts | 95 ++++++++++++++++++- .../src/orchestration-v2/ShellStream.ts | 58 ++++++++++- .../ThreadManagementService.test.ts | 80 +++++++++++++--- .../ThreadManagementService.ts | 70 ++++++++++---- apps/server/src/ws.ts | 2 + .../ThreadNotificationCoordinator.tsx | 33 +++++-- apps/web/src/components/chat/ChatComposer.tsx | 22 +++-- .../src/components/chat/MessagesTimeline.tsx | 13 +-- .../chat/ThreadAutomationsPanel.tsx | 2 +- apps/web/src/index.css | 29 ------ apps/web/src/lib/composerPathSearchState.ts | 12 ++- apps/web/src/state/entities.ts | 11 ++- apps/web/src/state/queries.ts | 4 +- .../src/state/shellReducer.test.ts | 15 +++ .../client-runtime/src/state/shellReducer.ts | 21 +++- 21 files changed, 525 insertions(+), 128 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index cc34256b97da..7a18455c4d70 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -63,6 +63,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; import { @@ -81,7 +82,14 @@ import { const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000; const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000; -const TASK_POLL_INTERVAL_MS = 50; +// Events that can make a delegated task terminal: the parent's task record, +// and the child's runs, nested tasks, and pending provider background work. +const TASK_WAKE_EVENTS = [ + { thread: "parent", eventType: "subagent.updated" }, + { thread: "child", eventType: "run.updated" }, + { thread: "child", eventType: "subagent.updated" }, + { thread: "child", eventType: "provider-thread.updated" }, +] as const; const DEFAULT_THREAD_LIST_LIMIT = 50; const DEFAULT_THREAD_READ_LIMIT = 50; const DEFAULT_THREAD_RUN_LIMIT = 10; @@ -1298,14 +1306,39 @@ const make = Effect.gen(function* () { return response; }); + // Re-read the task only when an event on the parent or child thread can + // change its status, instead of polling the projections every 50 ms. const waitForTask = (scope: McpThreadInvocationScope, taskId: NodeId, timeoutMs: number) => Effect.gen(function* () { - while (true) { - const result = yield* readTask(scope, taskId, false, true); - if (isTerminalTaskStatus(result.status)) return result; - yield* Effect.sleep(Duration.millis(TASK_POLL_INTERVAL_MS)); - } - }).pipe(Effect.timeoutOption(Duration.millis(timeoutMs))); + const streamError = (error: unknown) => + failure( + "orchestration_error", + `Unable to watch delegated task ${taskId}: ${errorMessage(error)}`, + ); + // Sequences are global, so one cursor taken before the first read + // replays anything either thread records after it. + const afterSequence = yield* threadManagement + .getThreadEventSequence(scope.thread.threadId) + .pipe(Effect.mapError(streamError)); + const initial = yield* readTask(scope, taskId, false, true); + if (isTerminalTaskStatus(initial.status)) return Option.some(initial); + // One stream per event type, so transcript events never fill a buffer. + return yield* Stream.mergeAll( + TASK_WAKE_EVENTS.map(({ thread, eventType }) => + threadManagement.streamStoredEventsFrom({ + threadId: thread === "parent" ? scope.thread.threadId : initial.childThreadId, + afterSequence, + eventType, + }), + ), + { concurrency: "unbounded" }, + ).pipe( + Stream.mapError(streamError), + Stream.mapEffect(() => readTask(scope, taskId, false, true)), + Stream.filter((result) => isTerminalTaskStatus(result.status)), + Stream.runHead, + ); + }).pipe(Effect.timeoutOption(Duration.millis(timeoutMs)), Effect.map(Option.flatten)); /** * A scheduled task the caller may change: one whose modes are no broader diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 89f5b7ae1a26..d20c06a87737 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2291,6 +2291,12 @@ describe("CodexAdapterV2 post-settle continuation", () => { (event): event is Extract => event.type === "message.updated" && event.message.role === "assistant", ); + const assistantTurnItems = (events: ReadonlyArray) => + events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [event.turnItem] + : [], + ); it.effect("keeps an asynchronous Codex question actionable after the turn completes", () => Effect.scoped( @@ -3427,6 +3433,49 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("streams text on the turn item and sends the message once, when it completes", () => + Effect.scoped( + Effect.gen(function* () { + const transcript = finalAnswerTranscript("codex-streamed-message-once", [ + { id: "answer", text: "CODEX_RECOVERY_OK", streamed: true, completionDelayMs: 100 }, + ]); + const harness = yield* makeCodexReplayHarness(transcript); + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-codex-streamed-message-once"), + text: "Reply with the requested recovery marker.", + }), + ); + yield* Effect.yieldNow; + yield* TestClock.adjust("50 millis"); + yield* awaitUntil( + () => assistantTurnItems(harness.events).length === 1, + "streamed turn item", + ); + assert.deepEqual( + assistantTurnItems(harness.events).map(({ text, streaming }) => ({ text, streaming })), + [{ text: "CODEX_RECOVERY_OK", streaming: true }], + ); + assert.deepEqual(assistantMessages(harness.events), []); + + yield* TestClock.adjust("50 millis"); + yield* awaitUntil(() => harness.terminalEvents().length === 1, "root turn terminal"); + assert.deepEqual( + assistantMessages(harness.events).map(({ message }) => ({ + text: message.text, + streaming: message.streaming, + })), + [{ text: "CODEX_RECOVERY_OK", streaming: false }], + ); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + it.effect("suppresses a later streamed duplicate final answer", () => Effect.scoped( Effect.gen(function* () { @@ -3578,10 +3627,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { yield* TestClock.adjust("50 millis"); yield* Effect.yieldNow; - assert.equal( - new Set(assistantMessages(harness.events).map((event) => event.message.id)).size, - 1, - ); + assert.equal(new Set(assistantTurnItems(harness.events).map((item) => item.id)).size, 1); yield* TestClock.adjust("50 millis"); yield* awaitUntil(() => harness.terminalEvents().length === 1, "root turn terminal"); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 6ef64f685bf3..ed9843f30d78 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -3180,11 +3180,16 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi driver: CODEX_PROVIDER, node: artifacts.node, }); - yield* emitProviderEvent({ - type: "message.updated", - driver: CODEX_PROVIDER, - message: artifacts.message, - }); + // Clients render streaming text from the turn item, so the + // message only carries the final text. flushTurn completes every + // buffered item when a turn ends, interrupted or not. + if (update.completed) { + yield* emitProviderEvent({ + type: "message.updated", + driver: CODEX_PROVIDER, + message: artifacts.message, + }); + } yield* emitProviderEvent({ type: "turn_item.updated", driver: CODEX_PROVIDER, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index ca772485b1af..7116741f8e2a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -316,6 +316,8 @@ export interface OrchestratorV2Shape { readonly streamStoredEventsFrom: (input?: { readonly threadId?: ThreadId; readonly afterSequence?: number; + /** Keep only this type, before the bounded buffer retains anything. */ + readonly eventType?: OrchestrationV2DomainEvent["type"]; }) => Stream.Stream; readonly streamDomainEvents: Stream.Stream; } diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index d93132005566..efe5a0577f23 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -3762,6 +3762,21 @@ it.layer(TestLayer)("ProjectionStoreV2", (it) => { (yield* projectionStore.getThreadProjection(sourceThreadId)).messages, ); const targetAfterRollback = yield* projectionStore.getThreadProjection(targetThreadId); + // Both shell reads drop the rolled-back run from the source count, while + // the fork keeps the prefix it inherited. + const shellSnapshot = yield* projectionStore.getShellSnapshot(); + for (const shell of [ + yield* projectionStore.getThreadShell(sourceThreadId), + shellSnapshot.threads.find((thread) => thread.id === sourceThreadId), + ]) { + assert.equal(shell?.itemCount, 2); + } + for (const shell of [ + yield* projectionStore.getThreadShell(targetThreadId), + shellSnapshot.threads.find((thread) => thread.id === targetThreadId), + ]) { + assert.equal(shell?.visibleItemCount, targetAfterRollback.visibleTurnItems.length); + } const forwardPage = yield* projectionStore.getTimelinePage(targetThreadId, { view: "activity", limit: 2, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 23563a2d69ff..87a16b586602 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1550,6 +1550,25 @@ function itemCountThroughRun(input: { return count; } +/** + * Threads that some shell row forks from at a run. Only these need run + * ordinals and per-run item counts, to count the inherited prefix, so a thread + * nobody forked from never pays for a scan of its run history. + */ +function shellForkSourceIds( + rows: ReadonlyArray<{ readonly forked_from_run_source_thread_id: string | null }>, +): ReadonlyArray { + return [ + ...new Set( + rows.flatMap((row) => + row.forked_from_run_source_thread_id === null + ? [] + : [ThreadId.make(row.forked_from_run_source_thread_id)], + ), + ), + ]; +} + function visibleItemCountForShell(input: { readonly threadId: ThreadId; readonly statesByThreadId: ReadonlyMap; @@ -4938,13 +4957,19 @@ export const layer: Layer.Layer = AND plan.kind = 'proposed_plan' AND plan.status = 'active' ) AS has_actionable_proposed_plan, + -- Count per run on the covering (thread_id, run_id) index, then + -- look up each run once, instead of one run lookup per item. ( - SELECT COUNT(*) - FROM orchestration_v2_projection_turn_items i + SELECT COALESCE(SUM(per_run.item_count), 0) + FROM ( + SELECT i.run_id, COUNT(*) AS item_count + FROM orchestration_v2_projection_turn_items i + WHERE i.thread_id = t.thread_id + GROUP BY i.run_id + ) per_run LEFT JOIN orchestration_v2_projection_runs r - ON r.run_id = i.run_id - WHERE i.thread_id = t.thread_id - AND (i.run_id IS NULL OR r.status <> 'rolled_back') + ON r.run_id = per_run.run_id + WHERE per_run.run_id IS NULL OR r.status <> 'rolled_back' ) AS item_count, ( SELECT COUNT(*) @@ -5426,14 +5451,15 @@ export const layer: Layer.Layer = } const threadRows = [...rowsByThreadId.values()]; const threadIds = [...rowsByThreadId.keys()]; + const forkSourceIds = shellForkSourceIds(threadRows); const readForThreadIds = ( read: (ids: ReadonlyArray) => Effect.Effect, unknown>, - ) => - threadIds.length === 0 ? Effect.succeed([] as ReadonlyArray) : read(threadIds); + ids: ReadonlyArray = threadIds, + ) => (ids.length === 0 ? Effect.succeed([] as ReadonlyArray) : read(ids)); const [runRows, itemCountRows, sequenceRows, providerThreadRows, pendingTurnItemRows] = yield* Effect.all([ - readForThreadIds(selectShellRunRows), - readForThreadIds(selectShellRunItemCounts), + readForThreadIds(selectShellRunRows, forkSourceIds), + readForThreadIds(selectShellRunItemCounts, forkSourceIds), sql<{ readonly snapshot_sequence: number | null }>` SELECT MAX(sequence) AS snapshot_sequence FROM orchestration_events @@ -5520,10 +5546,15 @@ export const layer: Layer.Layer = } const threadIds = [...rowsByThreadId.keys()]; + const forkSourceIds = shellForkSourceIds([...rowsByThreadId.values()]); const [runRows, itemCountRows, providerThreadRows, pendingTurnItemRows] = yield* Effect.all([ - selectShellRunRows(threadIds), - selectShellRunItemCounts(threadIds), + forkSourceIds.length === 0 + ? Effect.succeed([] as ReadonlyArray) + : selectShellRunRows(forkSourceIds), + forkSourceIds.length === 0 + ? Effect.succeed([] as ReadonlyArray) + : selectShellRunItemCounts(forkSourceIds), selectShellProviderThreadRows(threadIds), selectShellPendingTurnItemRows(threadIds), ]); diff --git a/apps/server/src/orchestration-v2/ShellStream.test.ts b/apps/server/src/orchestration-v2/ShellStream.test.ts index 3435468a41d3..b9c818c3f4c7 100644 --- a/apps/server/src/orchestration-v2/ShellStream.test.ts +++ b/apps/server/src/orchestration-v2/ShellStream.test.ts @@ -2,12 +2,16 @@ import { describe, expect, it } from "@effect/vitest"; import type { ApplicationStoredEvent, OrchestrationV2ShellSnapshot, + OrchestrationV2ShellStreamItem, OrchestrationV2StoredEvent, OrchestrationV2ThreadShell, } from "@t3tools/contracts"; -import { ProjectId, ThreadId } from "@t3tools/contracts"; +import { ProjectId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { archivedShellStreamItemFromThreadShell, @@ -20,6 +24,7 @@ import { shellStreamItemFromThreadShell, shellStreamItemsFromInitialSnapshot, shellStreamItemsFromResumeSnapshot, + skipUnchangedThreadShells, } from "./ShellStream.ts"; function project(sequence: number, id: string): ApplicationStoredEvent { @@ -493,3 +498,91 @@ describe("dedupeShellEnrichment", () => { }), ); }); + +describe("skipUnchangedThreadShells", () => { + const at = (ms: number) => DateTime.makeUnsafe(Date.parse("2026-10-01T00:00:00.000Z") + ms); + const shell = ( + id: string, + overrides: Partial = {}, + ): OrchestrationV2ThreadShell => ({ + id: ThreadId.make(id), + projectId: ProjectId.make("project-a"), + title: "Thread", + providerInstanceId: ProviderInstanceId.make("codex"), + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + worktreePath: null, + activeProviderThreadId: null, + lineage: { rootThreadId: ThreadId.make(id), parentThreadId: null, relationshipToParent: null }, + forkedFrom: null, + createdBy: "user", + creationSource: "web", + activeRunId: null, + latestVisibleMessage: null, + hasActionableProposedPlan: false, + itemCount: 1, + visibleItemCount: 1, + branch: null, + status: "running", + pendingRuntimeRequest: null, + pendingBackgroundTasks: [], + latestRunId: null, + latestUserMessageAt: null, + createdAt: at(0), + updatedAt: at(0), + archivedAt: null, + settledOverride: null, + settledAt: null, + deletedAt: null, + ...overrides, + }); + type Step = + | { readonly advanceMs: number } + | Exclude; + const updated = (sequence: number, thread: OrchestrationV2ThreadShell): Step => ({ + kind: "thread.updated", + sequence, + location: "active", + thread, + }); + const run = (steps: ReadonlyArray) => + Stream.fromIterable(steps).pipe( + Stream.mapEffect((step) => + "advanceMs" in step + ? TestClock.adjust(Duration.millis(step.advanceMs)).pipe(Effect.as(null)) + : Effect.succeed(step), + ), + Stream.filter((item) => item !== null), + skipUnchangedThreadShells, + Stream.runCollect, + Effect.map((items) => Array.from(items, (item) => ("sequence" in item ? item.sequence : -1))), + ); + + it.effect("drops shells that only moved updatedAt, and resends them after a while", () => + Effect.gen(function* () { + const sent = yield* run([ + updated(1, shell("a")), + updated(2, shell("a", { updatedAt: at(100) })), + updated(3, shell("a", { updatedAt: at(200), itemCount: 2, visibleItemCount: 2 })), + updated(4, shell("a", { updatedAt: at(300), itemCount: 2, visibleItemCount: 2 })), + updated(5, shell("b")), + // The resend window is 5 s. + { advanceMs: 5_000 }, + updated(6, shell("a", { updatedAt: at(5_300), itemCount: 2, visibleItemCount: 2 })), + ]); + expect(sent).toEqual([1, 3, 5, 6]); + }), + ); + + it.effect("sends the next update after a removal", () => + Effect.gen(function* () { + const sent = yield* run([ + updated(1, shell("a")), + { kind: "thread.removed", sequence: 2, location: "active", threadId: ThreadId.make("a") }, + updated(3, shell("a")), + ]); + expect(sent).toEqual([1, 2, 3]); + }), + ); +}); diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index 31f556ce6dec..1464f7a211a9 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -8,7 +8,12 @@ import type { OrchestrationV2ShellStreamItem, OrchestrationV2StoredEvent, } from "@t3tools/contracts"; -import { OrchestrationProjectShell as ProjectShellSchema } from "@t3tools/contracts"; +import { + OrchestrationProjectShell as ProjectShellSchema, + OrchestrationV2ThreadShell as ThreadShellSchema, +} from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -188,6 +193,57 @@ export function shellStreamItemsFromResumeSnapshot(input: { ); } +const sameThreadShell = Schema.toEquivalence(ThreadShellSchema); + +/** How long an unchanged thread shell may go unsent on a live subscription. */ +const UNCHANGED_THREAD_SHELL_RESEND_MS = 5_000; + +/** + * Drop live `thread.updated` deltas whose shell matches the last one this + * subscription sent, apart from `updatedAt`. Every thread event bumps + * `updatedAt`, so tool output and streaming text alone used to send the full, + * otherwise unchanged shell once per batch. + * + * A skipped delta leaves the client cursor behind, and a resume replays the + * events after that cursor. So an unchanged shell is still sent once its last + * send is `UNCHANGED_THREAD_SHELL_RESEND_MS` old. That bounds both the replay + * and how stale the client's `updatedAt` can get. + */ +export function skipUnchangedThreadShells( + stream: Stream.Stream, +): Stream.Stream { + return Stream.suspend(() => { + const lastSent = new Map< + OrchestrationV2ThreadShell["id"], + { readonly thread: OrchestrationV2ThreadShell; readonly sentAt: number } + >(); + return stream.pipe( + Stream.filterEffect((item) => + Effect.map(Clock.currentTimeMillis, (now) => { + if (item.kind === "thread.removed") { + lastSent.delete(item.threadId); + return true; + } + if (item.kind !== "thread.updated") return true; + const previous = lastSent.get(item.thread.id); + if ( + previous !== undefined && + now - previous.sentAt < UNCHANGED_THREAD_SHELL_RESEND_MS && + sameThreadShell( + { ...item.thread, updatedAt: previous.thread.updatedAt }, + previous.thread, + ) + ) { + return false; + } + lastSent.set(item.thread.id, { thread: item.thread, sentAt: now }); + return true; + }), + ), + ); + }); +} + /** Keep only the newest stored event per thread within a coalescing window. */ export function coalesceStoredThreadEvents( events: ReadonlyArray, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 7ff557c338dd..1c09b96bb305 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -5,6 +5,7 @@ import { NodeId, type OrchestrationV2Command, type OrchestrationV2Run, + type OrchestrationV2StoredEvent, type OrchestrationV2ThreadProjection, ProjectId, ProviderInstanceId, @@ -16,6 +17,8 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts"; @@ -357,7 +360,7 @@ it.effect.each([ const projectId = ProjectId.make("project:thread-management:wait-timeout"); const threadId = ThreadId.make("thread:thread-management:wait-timeout"); const runId = RunId.make("run:thread-management:wait-timeout"); - const loopRead = yield* Deferred.make(); + const subscribed = yield* Deferred.make(); let reads = 0; const projection = (status: OrchestrationV2Run["status"] | "missing") => ({ @@ -367,19 +370,18 @@ it.effect.each([ const testLayer = ThreadManagementService.layer.pipe( Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + // No run update arrives, so the timeout path runs while a final + // projection read can still observe a terminal run. + streamStoredEventsFrom: () => + Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.never), + ), getThreadRecords: () => - Effect.gen(function* () { + Effect.sync(() => { reads += 1; - if (reads === 1) { - return projection("running"); - } - if (reads === 2) { - // Park inside the wait loop so the timeout path runs while a - // final projection read can still observe a terminal run. - yield* Deferred.succeed(loopRead, undefined); - return yield* Effect.never; - } - return projection(scenario.finalStatus); + return projection(reads === 1 ? "running" : scenario.finalStatus); }), }), ), @@ -395,7 +397,7 @@ it.effect.each([ timeoutMs: 1, }) .pipe(Effect.result, Effect.forkChild); - yield* Deferred.await(loopRead); + yield* Deferred.await(subscribed); yield* TestClock.adjust(Duration.millis(1)); const result = yield* Fiber.join(fiber); @@ -419,3 +421,55 @@ it.effect.each([ } }), ); + +it.effect("waitForThread reads the run again only when the run updates", () => + Effect.gen(function* () { + const projectId = ProjectId.make("project:thread-management:wait-event"); + const threadId = ThreadId.make("thread:thread-management:wait-event"); + const runId = RunId.make("run:thread-management:wait-event"); + const subscribed = yield* Deferred.make(); + const events = yield* Queue.unbounded(); + let status: OrchestrationV2Run["status"] = "running"; + let reads = 0; + const stored = (sequence: number, event: object) => + ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent; + const testLayer = ThreadManagementService.layer.pipe( + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + // Only the run.updated stream carries events in this test. + streamStoredEventsFrom: (input) => + input?.eventType === "run.updated" + ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.fromQueue(events)), + ) + : Stream.never, + getThreadRecords: () => + Effect.sync(() => { + reads += 1; + return { + thread: { id: threadId, projectId, deletedAt: null }, + runs: [{ id: runId, status }], + } as unknown as OrchestrationV2ThreadProjection; + }), + }), + ), + ); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide(testLayer), + ); + const fiber = yield* service + .waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 }) + .pipe(Effect.forkChild); + yield* Deferred.await(subscribed); + status = "completed"; + yield* Queue.offer(events, stored(1, { type: "run.updated", payload: { id: "other-run" } })); + yield* Queue.offer(events, stored(2, { type: "run.updated", payload: { id: runId, status } })); + const result = yield* Fiber.join(fiber); + + expect(result).toMatchObject({ timedOut: false, run: { id: runId, status: "completed" } }); + // The first read plus one for this run's update. The other run caused none. + expect(reads).toBe(2); + }), +); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 2909ea2435aa..df4dc2cf246a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -32,6 +32,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; import * as Orchestrator from "./Orchestrator.ts"; import { projectTurnItemForDetail } from "./WireProjection.ts"; @@ -131,7 +132,6 @@ export interface ThreadManagementWaitInput { readonly threadId: ThreadId; readonly runId?: RunId; readonly timeoutMs: number; - readonly pollIntervalMs?: number; } export interface ThreadManagementWaitResult { @@ -632,6 +632,17 @@ const make = Effect.gen(function* () { const waitForThread: ThreadManagementServiceShape["waitForThread"] = (input) => Effect.gen(function* () { + const loadError = (cause: unknown) => + new ThreadManagementProjectionLoadError({ + projectId: input.projectId, + threadId: input.threadId, + cause, + }); + // Taken before the first read, so a run update between that read and + // the subscription below still replays. + const afterSequence = yield* orchestrator + .getThreadEventSequence(input.threadId) + .pipe(Effect.mapError(loadError)); const target = yield* getProjectThreadRecords(input, ["runs"]); const selectedRun = input.runId === undefined @@ -650,23 +661,46 @@ const make = Effect.gen(function* () { return { threadId: input.threadId, run: selectedRun, timedOut: false }; } - const wait = Effect.gen(function* () { - while (true) { - const current = yield* getProjectThreadRecords(input, ["runs"], { - runIds: [selectedRun.id], - }); - const run = current.runs.find((candidate) => candidate.id === selectedRun.id); - if (run === undefined) { - return yield* new ThreadManagementRunNotFoundError({ - threadId: input.threadId, - runId: selectedRun.id, - }); - } - if (isTerminalRunStatus(run.status)) return run; - yield* Effect.sleep(Duration.millis(Math.max(1, input.pollIntervalMs ?? 250))); - } - }).pipe(Effect.timeoutOption(Duration.millis(Math.max(1, input.timeoutMs)))); - const waited = yield* wait; + // Re-read the run only when the thread records a run update or its + // deletion, instead of polling the projection for up to an hour. Each + // stream keeps one event type, so transcript events never fill its buffer. + const wait = Stream.merge( + orchestrator.streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + eventType: "run.updated", + }), + orchestrator.streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + eventType: "thread.deleted", + }), + ).pipe( + Stream.mapError(loadError), + Stream.filter( + (stored) => + stored.event.type !== "run.updated" || stored.event.payload.id === selectedRun.id, + ), + Stream.mapEffect(() => + getProjectThreadRecords(input, ["runs"], { runIds: [selectedRun.id] }).pipe( + Effect.flatMap((current) => { + const run = current.runs.find((candidate) => candidate.id === selectedRun.id); + return run === undefined + ? Effect.fail( + new ThreadManagementRunNotFoundError({ + threadId: input.threadId, + runId: selectedRun.id, + }), + ) + : Effect.succeed(run); + }), + ), + ), + Stream.filter((run) => isTerminalRunStatus(run.status)), + Stream.runHead, + Effect.timeoutOption(Duration.millis(Math.max(1, input.timeoutMs))), + ); + const waited = Option.flatten(yield* wait); if (Option.isSome(waited)) { return { threadId: input.threadId, run: waited.value, timedOut: false }; } diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index a2a0d677efb2..a9c2e73933a7 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -132,6 +132,7 @@ import { shellStreamItemFromThreadShell, shellStreamItemsFromInitialSnapshot, shellStreamItemsFromResumeSnapshot, + skipUnchangedThreadShells, toShellApplicationEvent, type ShellApplicationEvent, } from "./orchestration-v2/ShellStream.ts"; @@ -1030,6 +1031,7 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc Stream.groupedWithin(512, Duration.millis(50)), Stream.mapEffect((events) => projectShellItems(Array.from(events))), Stream.flatMap(Stream.fromIterable), + skipUnchangedThreadShells, ); const liveFrom = (afterSequence: number) => diff --git a/apps/web/src/components/ThreadNotificationCoordinator.tsx b/apps/web/src/components/ThreadNotificationCoordinator.tsx index 741e850ca0d6..509483d3bccd 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.tsx +++ b/apps/web/src/components/ThreadNotificationCoordinator.tsx @@ -1,7 +1,7 @@ import { presentThreadShell } from "@t3tools/client-runtime/state/models"; import { useAtomValue } from "@effect/atom-react"; import { useNavigate, useParams } from "@tanstack/react-router"; -import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import type { EnvironmentId, OrchestrationV2ThreadShell, ThreadId } from "@t3tools/contracts"; import * as Option from "effect/Option"; import { CircleAlertIcon, @@ -88,6 +88,12 @@ export function ThreadNotificationCoordinator() { )); } +interface NotificationState { + readonly raw: OrchestrationV2ThreadShell; + readonly attention: string | null; + readonly completion: number | null; +} + function EnvironmentNotifications({ environmentId, onNotification, @@ -96,6 +102,10 @@ function EnvironmentNotifications({ onNotification: (environmentId: EnvironmentId, notification: Notification) => void; }) { const shell = useAtomValue(environmentShell.stateValueAtom(environmentId)); + // The shell reducer keeps the thread list and unchanged thread objects + // stable, so this only rescans when a thread actually changed. + const threads = + shell.status === "live" && Option.isSome(shell.snapshot) ? shell.snapshot.value.threads : null; const mode = useClientSettings((settings) => settings.notificationMode); const inAppNotificationsEnabled = useClientSettings( (settings) => settings.inAppNotificationsEnabled, @@ -104,22 +114,25 @@ function EnvironmentNotifications({ const { environmentId: activeEnvironmentId, threadId: activeThreadId } = useParams({ strict: false, }); - const previous = useRef( - new Map(), - ); + const previous = useRef(new Map()); useEffect(() => { - if (shell.status !== "live" || Option.isNone(shell.snapshot)) { + if (threads === null) { previous.current.clear(); return; } - const next = new Map(); - for (const rawThread of shell.snapshot.value.threads) { + const next = new Map(); + for (const rawThread of threads) { if (rawThread.lineage.relationshipToParent === "subagent") continue; + const prior = previous.current.get(rawThread.id); + // The same object cannot produce a new notification. + if (prior?.raw === rawThread) { + next.set(rawThread.id, prior); + continue; + } const thread = presentThreadShell(environmentId, rawThread); let status = resolveSidebarThreadStatus(thread); if (status === "ready" && thread.latestRun?.status === "failed") status = "failed"; - const prior = previous.current.get(thread.id); const attention = status === "input" || status === "approval" || status === "failed" || status === "limited" ? `${thread.latestRun?.runId ?? ""}:${status}` @@ -132,7 +145,7 @@ function EnvironmentNotifications({ Number.isFinite(completedAt) ? completedAt : (prior?.completion ?? null); - next.set(thread.id, { attention, completion }); + next.set(thread.id, { raw: rawThread, attention, completion }); if (!prior || thread.archivedAt !== null) continue; const kind = attention && attention !== prior.attention @@ -227,7 +240,7 @@ function EnvironmentNotifications({ mode, navigate, onNotification, - shell, + threads, ]); return null; diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index 36b2b2a6a271..0ee5a5fc1402 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -2170,12 +2170,19 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) [selectedProviderEntry], ); const compactCommandAvailable = providerSupportsManualCompaction(selectedProviderEntry); - const selectedProviderSkills = selectedProviderStatus - ? resolveProviderSkillsForCwd(selectedProviderStatus, gitCwd) - : []; - const selectedProviderSlashCommands = selectedProviderStatus - ? resolveProviderSlashCommandsForCwd(selectedProviderStatus, gitCwd) - : []; + // Memoized so the composer menu memo below can cache between renders. + const selectedProviderSkills = useMemo( + () => + selectedProviderStatus ? resolveProviderSkillsForCwd(selectedProviderStatus, gitCwd) : [], + [gitCwd, selectedProviderStatus], + ); + const selectedProviderSlashCommands = useMemo( + () => + selectedProviderStatus + ? resolveProviderSlashCommandsForCwd(selectedProviderStatus, gitCwd) + : [], + [gitCwd, selectedProviderStatus], + ); const refreshProviders = useAtomCommand(serverEnvironment.refreshProviders, { reportFailure: false, }); @@ -2526,7 +2533,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const settledPullRequestTextQuery = pullRequestTextQuery === debouncedPullRequestTextQuery ? pullRequestTextQuery : null; const isPathTrigger = composerTriggerKind === "path"; - const environmentThreadShells = useThreadShells(); + // Thread shells only feed `@` thread matches, so skip shell updates otherwise. + const environmentThreadShells = useThreadShells(isPathTrigger); const workspaceEntries = useComposerPathSearch({ environmentId, cwd: isPathTrigger ? gitCwd : null, diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 33f51752fbca..32d19c6472f2 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -3525,7 +3525,7 @@ function LiveActivityRow({ const showShimmer = animated && shimmer; return (
+ {label} } @@ -3733,7 +3727,6 @@ function WorkGroupHeader(props: { }) { return ( } label={ - + {props.label} } diff --git a/apps/web/src/components/chat/ThreadAutomationsPanel.tsx b/apps/web/src/components/chat/ThreadAutomationsPanel.tsx index d4434fffbf18..cd1202d1e429 100644 --- a/apps/web/src/components/chat/ThreadAutomationsPanel.tsx +++ b/apps/web/src/components/chat/ThreadAutomationsPanel.tsx @@ -25,7 +25,7 @@ import { const STATUS_DOT_CLASS: Record = { never: "bg-muted-foreground/40", - running: "animate-pulse bg-sky-500", + running: "bg-sky-500", succeeded: "bg-emerald-500", failed: "bg-destructive", }; diff --git a/apps/web/src/index.css b/apps/web/src/index.css index 46b57325a3ff..a4b80193dce2 100644 --- a/apps/web/src/index.css +++ b/apps/web/src/index.css @@ -499,35 +499,6 @@ html[data-mobile-composer-route-transition="true"]::view-transition-old(t3-mobil } } -@keyframes live-tool-shine { - from { - background-position: - -4.5rem 0, - 0 0; - } - to { - background-position: - calc(100% + 4.5rem) 0, - 0 0; - } -} - -@utility live-tool-shine { - @media (prefers-reduced-motion: no-preference) and (forced-colors: none) { - color: transparent; - background-image: - linear-gradient(to right, transparent, var(--contrast-foreground), transparent), - linear-gradient(var(--contrast-secondary-label), var(--contrast-secondary-label)); - background-size: - 4.5rem 100%, - 100% 100%; - background-repeat: no-repeat; - background-clip: text; - animation: live-tool-shine 2.2s steps(30) infinite; - animation-play-state: var(--visible-animation-state, paused); - } -} - @utility visible-animate-spin { animation: var(--animate-spin); animation-play-state: var(--visible-animation-state, paused); diff --git a/apps/web/src/lib/composerPathSearchState.ts b/apps/web/src/lib/composerPathSearchState.ts index a2ad55c67752..6a2472e30272 100644 --- a/apps/web/src/lib/composerPathSearchState.ts +++ b/apps/web/src/lib/composerPathSearchState.ts @@ -3,15 +3,19 @@ import { type ComposerPathSearchTarget, } from "@t3tools/client-runtime/state/threads"; +import { useMemo } from "react"; + import { useComposerPathSearch as useComposerPathSearchQuery } from "../state/queries"; export function useComposerPathSearch(target: ComposerPathSearchTarget): ComposerPathSearchState { const state = useComposerPathSearchQuery(target); + // A stable list lets the composer menu memo cache between renders. + const entries = useMemo( + () => state.entries.map((entry) => ({ path: entry.path, kind: entry.kind })), + [state.entries], + ); return { - entries: state.entries.map((entry) => ({ - path: entry.path, - kind: entry.kind, - })), + entries, error: state.error, isPending: state.isPending, }; diff --git a/apps/web/src/state/entities.ts b/apps/web/src/state/entities.ts index 00fcbd23cd7b..ad20c3f3dd83 100644 --- a/apps/web/src/state/entities.ts +++ b/apps/web/src/state/entities.ts @@ -31,6 +31,9 @@ const EMPTY_PROJECT_ATOM = Atom.make(null).pipe( const EMPTY_THREAD_REFS_ATOM = Atom.make(EMPTY_THREAD_REFS).pipe( Atom.withLabel("web-thread-refs:empty"), ); +const EMPTY_THREAD_SHELLS_ATOM = Atom.make>( + Object.freeze([]), +).pipe(Atom.withLabel("web-thread-shells:empty")); const EMPTY_THREAD_SHELL_ATOM = Atom.make(null).pipe( Atom.withLabel("web-thread-shell:empty"), ); @@ -82,8 +85,12 @@ export function useServerConfigs(): ReadonlyMap { return useAtomValue(environmentServerConfigsAtom); } -export function useThreadShells(): ReadonlyArray { - return useAtomValue(environmentThreadShells.threadShellsAtom); +/** Every thread shell. Pass `enabled: false` to read a stable empty list and + skip re-rendering on each shell update while the caller does not need them. */ +export function useThreadShells(enabled = true): ReadonlyArray { + return useAtomValue( + enabled ? environmentThreadShells.threadShellsAtom : EMPTY_THREAD_SHELLS_ATOM, + ); } export function useAllEnvironmentShellsBootstrapped(): boolean { diff --git a/apps/web/src/state/queries.ts b/apps/web/src/state/queries.ts index 0cb413df91d0..2b5a96ab4cc9 100644 --- a/apps/web/src/state/queries.ts +++ b/apps/web/src/state/queries.ts @@ -12,6 +12,7 @@ import { type VcsRefTarget } from "@t3tools/client-runtime/state/vcs"; import type { EnvironmentId, ProjectContentMatch, + ProjectEntry, ProjectEntryKind, ThreadId, TurnItemId, @@ -37,6 +38,7 @@ const PROJECT_CONTENT_SEARCH_LIMIT = 500; const THREAD_SEARCH_DEBOUNCE_MS = 200; const VCS_REF_LIST_LIMIT = 100; const EMPTY_REFS: ReadonlyArray = []; +const EMPTY_PROJECT_ENTRIES: ReadonlyArray = []; const EMPTY_CONTENT_MATCHES: ReadonlyArray = []; const INITIAL_BRANCH_CURSORS = [undefined] as const; const EMPTY_THREAD_SEARCH_MATCHES: ReadonlyArray = Object.freeze([]); @@ -254,7 +256,7 @@ export function useProjectPathSearch( ); return { - entries: result.data?.entries ?? [], + entries: result.data?.entries ?? EMPTY_PROJECT_ENTRIES, error: result.error, isPending: !areProjectPathSearchTargetsEqual(normalizedTarget, debouncedTarget) || result.isPending, diff --git a/packages/client-runtime/src/state/shellReducer.test.ts b/packages/client-runtime/src/state/shellReducer.test.ts index eefaebb0a112..7a3bd4c75254 100644 --- a/packages/client-runtime/src/state/shellReducer.test.ts +++ b/packages/client-runtime/src/state/shellReducer.test.ts @@ -41,6 +41,21 @@ describe("applyShellStreamEvent", () => { expect(next.threads[2]).toBe(threads[2]); }); + it("keeps the thread list and object when an update changes nothing", () => { + const threads = ["a", "b"].map((id) => ({ ...v2ThreadShell, id: ThreadId.make(id) })); + const next = applyShellStreamEvent( + { ...v2ShellSnapshot, threads }, + { + kind: "thread.updated", + sequence: 1, + location: "active", + thread: { ...threads[1]! }, + }, + ); + expect(next.threads).toBe(threads); + expect(next.snapshotSequence).toBe(1); + }); + it("ignores stale project updates without mutating the snapshot", () => { const snapshotWithProject = { ...v2ShellSnapshot, diff --git a/packages/client-runtime/src/state/shellReducer.ts b/packages/client-runtime/src/state/shellReducer.ts index 752934da5fc0..331b271d6eb7 100644 --- a/packages/client-runtime/src/state/shellReducer.ts +++ b/packages/client-runtime/src/state/shellReducer.ts @@ -1,8 +1,12 @@ -import type { - OrchestrationProjectShell, - OrchestrationV2ShellSnapshot, - OrchestrationV2ShellStreamItem, +import { + OrchestrationV2ThreadShell, + type OrchestrationProjectShell, + type OrchestrationV2ShellSnapshot, + type OrchestrationV2ShellStreamItem, } from "@t3tools/contracts"; +import * as Schema from "effect/Schema"; + +const sameThreadShell = Schema.toEquivalence(OrchestrationV2ThreadShell); function upsertById( items: ReadonlyArray, @@ -123,6 +127,15 @@ export function applyShellStreamEvent( snapshotSequence: event.sequence, }; case "thread.updated": { + // An unchanged shell keeps its object and the list, so subscribers that + // compare by reference skip the update. Only the cursor moves. + const existing = + event.location === "active" + ? snapshot.threads.find((thread) => thread.id === event.thread.id) + : undefined; + if (existing !== undefined && sameThreadShell(existing, event.thread)) { + return { ...snapshot, snapshotSequence: event.sequence }; + } const withoutThread = (threads: OrchestrationV2ShellSnapshot["threads"]) => threads.filter((thread) => thread.id !== event.thread.id); return {