authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-05 15:39:24-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-05 15:39:24-07:00
logc138163c8cc39d4a98b2d3d796bcef183013bac5
tree645bbb34158fe2b743152e4db4725541a9053921
parent2afe87ea4bb94f4dbbd1660a556cbffefd7498a2
signature Signed by PGP key B5690EEEBB952194

perf: cheaper shell refreshes, one copy of Codex streaming text, no MCP wait polling (#15033)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

21 files changed, 525 insertions(+), 128 deletions(-)

apps/server/src/mcp/OrchestratorMcpService.ts+40-7
...@@ -63,6 +63,7 @@ import * as Effect from "effect/Effect";...@@ -63,6 +63,7 @@ import * as Effect from "effect/Effect";
63import * as Layer from "effect/Layer";63import * as Layer from "effect/Layer";
64import * as Option from "effect/Option";64import * as Option from "effect/Option";
65import * as Schema from "effect/Schema";65import * as Schema from "effect/Schema";
66import * as Stream from "effect/Stream";
6667
67import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";68import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
68import {69import {
...@@ -81,7 +82,14 @@ import {...@@ -81,7 +82,14 @@ import {
8182
82const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000;83const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000;
83const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000;84const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000;
84const TASK_POLL_INTERVAL_MS = 50;85// Events that can make a delegated task terminal: the parent's task record,
86// and the child's runs, nested tasks, and pending provider background work.
87const TASK_WAKE_EVENTS = [
88 { thread: "parent", eventType: "subagent.updated" },
89 { thread: "child", eventType: "run.updated" },
90 { thread: "child", eventType: "subagent.updated" },
91 { thread: "child", eventType: "provider-thread.updated" },
92] as const;
85const DEFAULT_THREAD_LIST_LIMIT = 50;93const DEFAULT_THREAD_LIST_LIMIT = 50;
86const DEFAULT_THREAD_READ_LIMIT = 50;94const DEFAULT_THREAD_READ_LIMIT = 50;
87const DEFAULT_THREAD_RUN_LIMIT = 10;95const DEFAULT_THREAD_RUN_LIMIT = 10;
...@@ -1298,14 +1306,39 @@ const make = Effect.gen(function* () {...@@ -1298,14 +1306,39 @@ const make = Effect.gen(function* () {
1298 return response;1306 return response;
1299 });1307 });
13001308
1309 // Re-read the task only when an event on the parent or child thread can
1310 // change its status, instead of polling the projections every 50 ms.
1301 const waitForTask = (scope: McpThreadInvocationScope, taskId: NodeId, timeoutMs: number) =>1311 const waitForTask = (scope: McpThreadInvocationScope, taskId: NodeId, timeoutMs: number) =>
1302 Effect.gen(function* () {1312 Effect.gen(function* () {
1303 while (true) {1313 const streamError = (error: unknown) =>
1304 const result = yield* readTask(scope, taskId, false, true);1314 failure(
1305 if (isTerminalTaskStatus(result.status)) return result;1315 "orchestration_error",
1306 yield* Effect.sleep(Duration.millis(TASK_POLL_INTERVAL_MS));1316 `Unable to watch delegated task ${taskId}: ${errorMessage(error)}`,
1307 }1317 );
1308 }).pipe(Effect.timeoutOption(Duration.millis(timeoutMs)));1318 // Sequences are global, so one cursor taken before the first read
1319 // replays anything either thread records after it.
1320 const afterSequence = yield* threadManagement
1321 .getThreadEventSequence(scope.thread.threadId)
1322 .pipe(Effect.mapError(streamError));
1323 const initial = yield* readTask(scope, taskId, false, true);
1324 if (isTerminalTaskStatus(initial.status)) return Option.some(initial);
1325 // One stream per event type, so transcript events never fill a buffer.
1326 return yield* Stream.mergeAll(
1327 TASK_WAKE_EVENTS.map(({ thread, eventType }) =>
1328 threadManagement.streamStoredEventsFrom({
1329 threadId: thread === "parent" ? scope.thread.threadId : initial.childThreadId,
1330 afterSequence,
1331 eventType,
1332 }),
1333 ),
1334 { concurrency: "unbounded" },
1335 ).pipe(
1336 Stream.mapError(streamError),
1337 Stream.mapEffect(() => readTask(scope, taskId, false, true)),
1338 Stream.filter((result) => isTerminalTaskStatus(result.status)),
1339 Stream.runHead,
1340 );
1341 }).pipe(Effect.timeoutOption(Duration.millis(timeoutMs)), Effect.map(Option.flatten));
13091342
1310 /**1343 /**
1311 * A scheduled task the caller may change: one whose modes are no broader1344 * A scheduled task the caller may change: one whose modes are no broader
apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts+50-4
...@@ -2291,6 +2291,12 @@ describe("CodexAdapterV2 post-settle continuation", () => {...@@ -2291,6 +2291,12 @@ describe("CodexAdapterV2 post-settle continuation", () => {
2291 (event): event is Extract<ProviderAdapterV2Event, { type: "message.updated" }> =>2291 (event): event is Extract<ProviderAdapterV2Event, { type: "message.updated" }> =>
2292 event.type === "message.updated" && event.message.role === "assistant",2292 event.type === "message.updated" && event.message.role === "assistant",
2293 );2293 );
2294 const assistantTurnItems = (events: ReadonlyArray<ProviderAdapterV2Event>) =>
2295 events.flatMap((event) =>
2296 event.type === "turn_item.updated" && event.turnItem.type === "assistant_message"
2297 ? [event.turnItem]
2298 : [],
2299 );
22942300
2295 it.effect("keeps an asynchronous Codex question actionable after the turn completes", () =>2301 it.effect("keeps an asynchronous Codex question actionable after the turn completes", () =>
2296 Effect.scoped(2302 Effect.scoped(
...@@ -3427,6 +3433,49 @@ describe("CodexAdapterV2 post-settle continuation", () => {...@@ -3427,6 +3433,49 @@ describe("CodexAdapterV2 post-settle continuation", () => {
3427 ),3433 ),
3428 );3434 );
34293435
3436 it.effect("streams text on the turn item and sends the message once, when it completes", () =>
3437 Effect.scoped(
3438 Effect.gen(function* () {
3439 const transcript = finalAnswerTranscript("codex-streamed-message-once", [
3440 { id: "answer", text: "CODEX_RECOVERY_OK", streamed: true, completionDelayMs: 100 },
3441 ]);
3442 const harness = yield* makeCodexReplayHarness(transcript);
3443 const now = yield* DateTime.now;
3444
3445 yield* harness.runtime.startTurn(
3446 makeCodexTestTurnInput({
3447 threadId: harness.threadId,
3448 providerThread: harness.providerThread,
3449 now,
3450 attemptId: RunAttemptId.make("attempt-codex-streamed-message-once"),
3451 text: "Reply with the requested recovery marker.",
3452 }),
3453 );
3454 yield* Effect.yieldNow;
3455 yield* TestClock.adjust("50 millis");
3456 yield* awaitUntil(
3457 () => assistantTurnItems(harness.events).length === 1,
3458 "streamed turn item",
3459 );
3460 assert.deepEqual(
3461 assistantTurnItems(harness.events).map(({ text, streaming }) => ({ text, streaming })),
3462 [{ text: "CODEX_RECOVERY_OK", streaming: true }],
3463 );
3464 assert.deepEqual(assistantMessages(harness.events), []);
3465
3466 yield* TestClock.adjust("50 millis");
3467 yield* awaitUntil(() => harness.terminalEvents().length === 1, "root turn terminal");
3468 assert.deepEqual(
3469 assistantMessages(harness.events).map(({ message }) => ({
3470 text: message.text,
3471 streaming: message.streaming,
3472 })),
3473 [{ text: "CODEX_RECOVERY_OK", streaming: false }],
3474 );
3475 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
3476 ),
3477 );
3478
3430 it.effect("suppresses a later streamed duplicate final answer", () =>3479 it.effect("suppresses a later streamed duplicate final answer", () =>
3431 Effect.scoped(3480 Effect.scoped(
3432 Effect.gen(function* () {3481 Effect.gen(function* () {
...@@ -3578,10 +3627,7 @@ describe("CodexAdapterV2 post-settle continuation", () => {...@@ -3578,10 +3627,7 @@ describe("CodexAdapterV2 post-settle continuation", () => {
3578 yield* TestClock.adjust("50 millis");3627 yield* TestClock.adjust("50 millis");
3579 yield* Effect.yieldNow;3628 yield* Effect.yieldNow;
35803629
3581 assert.equal(3630 assert.equal(new Set(assistantTurnItems(harness.events).map((item) => item.id)).size, 1);
3582 new Set(assistantMessages(harness.events).map((event) => event.message.id)).size,
3583 1,
3584 );
35853631
3586 yield* TestClock.adjust("50 millis");3632 yield* TestClock.adjust("50 millis");
3587 yield* awaitUntil(() => harness.terminalEvents().length === 1, "root turn terminal");3633 yield* awaitUntil(() => harness.terminalEvents().length === 1, "root turn terminal");
apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts+10-5
...@@ -3180,11 +3180,16 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi...@@ -3180,11 +3180,16 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
3180 driver: CODEX_PROVIDER,3180 driver: CODEX_PROVIDER,
3181 node: artifacts.node,3181 node: artifacts.node,
3182 });3182 });
3183 yield* emitProviderEvent({3183 // Clients render streaming text from the turn item, so the
3184 type: "message.updated",3184 // message only carries the final text. flushTurn completes every
3185 driver: CODEX_PROVIDER,3185 // buffered item when a turn ends, interrupted or not.
3186 message: artifacts.message,3186 if (update.completed) {
3187 });3187 yield* emitProviderEvent({
3188 type: "message.updated",
3189 driver: CODEX_PROVIDER,
3190 message: artifacts.message,
3191 });
3192 }
3188 yield* emitProviderEvent({3193 yield* emitProviderEvent({
3189 type: "turn_item.updated",3194 type: "turn_item.updated",
3190 driver: CODEX_PROVIDER,3195 driver: CODEX_PROVIDER,
apps/server/src/orchestration-v2/Orchestrator.ts+2
...@@ -316,6 +316,8 @@ export interface OrchestratorV2Shape {...@@ -316,6 +316,8 @@ export interface OrchestratorV2Shape {
316 readonly streamStoredEventsFrom: (input?: {316 readonly streamStoredEventsFrom: (input?: {
317 readonly threadId?: ThreadId;317 readonly threadId?: ThreadId;
318 readonly afterSequence?: number;318 readonly afterSequence?: number;
319 /** Keep only this type, before the bounded buffer retains anything. */
320 readonly eventType?: OrchestrationV2DomainEvent["type"];
319 }) => Stream.Stream<OrchestrationV2StoredEvent, OrchestratorV2Error>;321 }) => Stream.Stream<OrchestrationV2StoredEvent, OrchestratorV2Error>;
320 readonly streamDomainEvents: Stream.Stream<OrchestrationV2DomainEvent, OrchestratorV2Error>;322 readonly streamDomainEvents: Stream.Stream<OrchestrationV2DomainEvent, OrchestratorV2Error>;
321}323}
apps/server/src/orchestration-v2/ProjectionStore.test.ts+15
...@@ -3762,6 +3762,21 @@ it.layer(TestLayer)("ProjectionStoreV2", (it) => {...@@ -3762,6 +3762,21 @@ it.layer(TestLayer)("ProjectionStoreV2", (it) => {
3762 (yield* projectionStore.getThreadProjection(sourceThreadId)).messages,3762 (yield* projectionStore.getThreadProjection(sourceThreadId)).messages,
3763 );3763 );
3764 const targetAfterRollback = yield* projectionStore.getThreadProjection(targetThreadId);3764 const targetAfterRollback = yield* projectionStore.getThreadProjection(targetThreadId);
3765 // Both shell reads drop the rolled-back run from the source count, while
3766 // the fork keeps the prefix it inherited.
3767 const shellSnapshot = yield* projectionStore.getShellSnapshot();
3768 for (const shell of [
3769 yield* projectionStore.getThreadShell(sourceThreadId),
3770 shellSnapshot.threads.find((thread) => thread.id === sourceThreadId),
3771 ]) {
3772 assert.equal(shell?.itemCount, 2);
3773 }
3774 for (const shell of [
3775 yield* projectionStore.getThreadShell(targetThreadId),
3776 shellSnapshot.threads.find((thread) => thread.id === targetThreadId),
3777 ]) {
3778 assert.equal(shell?.visibleItemCount, targetAfterRollback.visibleTurnItems.length);
3779 }
3765 const forwardPage = yield* projectionStore.getTimelinePage(targetThreadId, {3780 const forwardPage = yield* projectionStore.getTimelinePage(targetThreadId, {
3766 view: "activity",3781 view: "activity",
3767 limit: 2,3782 limit: 2,
apps/server/src/orchestration-v2/ProjectionStore.ts+42-11
...@@ -1550,6 +1550,25 @@ function itemCountThroughRun(input: {...@@ -1550,6 +1550,25 @@ function itemCountThroughRun(input: {
1550 return count;1550 return count;
1551}1551}
15521552
1553/**
1554 * Threads that some shell row forks from at a run. Only these need run
1555 * ordinals and per-run item counts, to count the inherited prefix, so a thread
1556 * nobody forked from never pays for a scan of its run history.
1557 */
1558function shellForkSourceIds(
1559 rows: ReadonlyArray<{ readonly forked_from_run_source_thread_id: string | null }>,
1560): ReadonlyArray<ThreadId> {
1561 return [
1562 ...new Set(
1563 rows.flatMap((row) =>
1564 row.forked_from_run_source_thread_id === null
1565 ? []
1566 : [ThreadId.make(row.forked_from_run_source_thread_id)],
1567 ),
1568 ),
1569 ];
1570}
1571
1553function visibleItemCountForShell(input: {1572function visibleItemCountForShell(input: {
1554 readonly threadId: ThreadId;1573 readonly threadId: ThreadId;
1555 readonly statesByThreadId: ReadonlyMap<ThreadId, ShellThreadState>;1574 readonly statesByThreadId: ReadonlyMap<ThreadId, ShellThreadState>;
...@@ -4938,13 +4957,19 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -4938,13 +4957,19 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
4938 AND plan.kind = 'proposed_plan'4957 AND plan.kind = 'proposed_plan'
4939 AND plan.status = 'active'4958 AND plan.status = 'active'
4940 ) AS has_actionable_proposed_plan,4959 ) AS has_actionable_proposed_plan,
4960 -- Count per run on the covering (thread_id, run_id) index, then
4961 -- look up each run once, instead of one run lookup per item.
4941 (4962 (
4942 SELECT COUNT(*)4963 SELECT COALESCE(SUM(per_run.item_count), 0)
4943 FROM orchestration_v2_projection_turn_items i4964 FROM (
4965 SELECT i.run_id, COUNT(*) AS item_count
4966 FROM orchestration_v2_projection_turn_items i
4967 WHERE i.thread_id = t.thread_id
4968 GROUP BY i.run_id
4969 ) per_run
4944 LEFT JOIN orchestration_v2_projection_runs r4970 LEFT JOIN orchestration_v2_projection_runs r
4945 ON r.run_id = i.run_id4971 ON r.run_id = per_run.run_id
4946 WHERE i.thread_id = t.thread_id4972 WHERE per_run.run_id IS NULL OR r.status <> 'rolled_back'
4947 AND (i.run_id IS NULL OR r.status <> 'rolled_back')
4948 ) AS item_count,4973 ) AS item_count,
4949 (4974 (
4950 SELECT COUNT(*)4975 SELECT COUNT(*)
...@@ -5426,14 +5451,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -5426,14 +5451,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
5426 }5451 }
5427 const threadRows = [...rowsByThreadId.values()];5452 const threadRows = [...rowsByThreadId.values()];
5428 const threadIds = [...rowsByThreadId.keys()];5453 const threadIds = [...rowsByThreadId.keys()];
5454 const forkSourceIds = shellForkSourceIds(threadRows);
5429 const readForThreadIds = <A>(5455 const readForThreadIds = <A>(
5430 read: (ids: ReadonlyArray<ThreadId>) => Effect.Effect<ReadonlyArray<A>, unknown>,5456 read: (ids: ReadonlyArray<ThreadId>) => Effect.Effect<ReadonlyArray<A>, unknown>,
5431 ) =>5457 ids: ReadonlyArray<ThreadId> = threadIds,
5432 threadIds.length === 0 ? Effect.succeed([] as ReadonlyArray<A>) : read(threadIds);5458 ) => (ids.length === 0 ? Effect.succeed([] as ReadonlyArray<A>) : read(ids));
5433 const [runRows, itemCountRows, sequenceRows, providerThreadRows, pendingTurnItemRows] =5459 const [runRows, itemCountRows, sequenceRows, providerThreadRows, pendingTurnItemRows] =
5434 yield* Effect.all([5460 yield* Effect.all([
5435 readForThreadIds(selectShellRunRows),5461 readForThreadIds(selectShellRunRows, forkSourceIds),
5436 readForThreadIds(selectShellRunItemCounts),5462 readForThreadIds(selectShellRunItemCounts, forkSourceIds),
5437 sql<{ readonly snapshot_sequence: number | null }>`5463 sql<{ readonly snapshot_sequence: number | null }>`
5438 SELECT MAX(sequence) AS snapshot_sequence5464 SELECT MAX(sequence) AS snapshot_sequence
5439 FROM orchestration_events5465 FROM orchestration_events
...@@ -5520,10 +5546,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -5520,10 +5546,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
5520 }5546 }
55215547
5522 const threadIds = [...rowsByThreadId.keys()];5548 const threadIds = [...rowsByThreadId.keys()];
5549 const forkSourceIds = shellForkSourceIds([...rowsByThreadId.values()]);
5523 const [runRows, itemCountRows, providerThreadRows, pendingTurnItemRows] =5550 const [runRows, itemCountRows, providerThreadRows, pendingTurnItemRows] =
5524 yield* Effect.all([5551 yield* Effect.all([
5525 selectShellRunRows(threadIds),5552 forkSourceIds.length === 0
5526 selectShellRunItemCounts(threadIds),5553 ? Effect.succeed([] as ReadonlyArray<ShellRunRow>)
5554 : selectShellRunRows(forkSourceIds),
5555 forkSourceIds.length === 0
5556 ? Effect.succeed([] as ReadonlyArray<ShellRunItemCountRow>)
5557 : selectShellRunItemCounts(forkSourceIds),
5527 selectShellProviderThreadRows(threadIds),5558 selectShellProviderThreadRows(threadIds),
5528 selectShellPendingTurnItemRows(threadIds),5559 selectShellPendingTurnItemRows(threadIds),
5529 ]);5560 ]);
apps/server/src/orchestration-v2/ShellStream.test.ts+94-1
...@@ -2,12 +2,16 @@ import { describe, expect, it } from "@effect/vitest";...@@ -2,12 +2,16 @@ import { describe, expect, it } from "@effect/vitest";
2import type {2import type {
3 ApplicationStoredEvent,3 ApplicationStoredEvent,
4 OrchestrationV2ShellSnapshot,4 OrchestrationV2ShellSnapshot,
5 OrchestrationV2ShellStreamItem,
5 OrchestrationV2StoredEvent,6 OrchestrationV2StoredEvent,
6 OrchestrationV2ThreadShell,7 OrchestrationV2ThreadShell,
7} from "@t3tools/contracts";8} from "@t3tools/contracts";
8import { ProjectId, ThreadId } from "@t3tools/contracts";9import { ProjectId, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
10import * as DateTime from "effect/DateTime";
11import * as Duration from "effect/Duration";
9import * as Effect from "effect/Effect";12import * as Effect from "effect/Effect";
10import * as Stream from "effect/Stream";13import * as Stream from "effect/Stream";
14import * as TestClock from "effect/testing/TestClock";
1115
12import {16import {
13 archivedShellStreamItemFromThreadShell,17 archivedShellStreamItemFromThreadShell,
...@@ -20,6 +24,7 @@ import {...@@ -20,6 +24,7 @@ import {
20 shellStreamItemFromThreadShell,24 shellStreamItemFromThreadShell,
21 shellStreamItemsFromInitialSnapshot,25 shellStreamItemsFromInitialSnapshot,
22 shellStreamItemsFromResumeSnapshot,26 shellStreamItemsFromResumeSnapshot,
27 skipUnchangedThreadShells,
23} from "./ShellStream.ts";28} from "./ShellStream.ts";
2429
25function project(sequence: number, id: string): ApplicationStoredEvent {30function project(sequence: number, id: string): ApplicationStoredEvent {
...@@ -493,3 +498,91 @@ describe("dedupeShellEnrichment", () => {...@@ -493,3 +498,91 @@ describe("dedupeShellEnrichment", () => {
493 }),498 }),
494 );499 );
495});500});
501
502describe("skipUnchangedThreadShells", () => {
503 const at = (ms: number) => DateTime.makeUnsafe(Date.parse("2026-10-01T00:00:00.000Z") + ms);
504 const shell = (
505 id: string,
506 overrides: Partial<OrchestrationV2ThreadShell> = {},
507 ): OrchestrationV2ThreadShell => ({
508 id: ThreadId.make(id),
509 projectId: ProjectId.make("project-a"),
510 title: "Thread",
511 providerInstanceId: ProviderInstanceId.make("codex"),
512 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
513 runtimeMode: "full-access",
514 interactionMode: "default",
515 worktreePath: null,
516 activeProviderThreadId: null,
517 lineage: { rootThreadId: ThreadId.make(id), parentThreadId: null, relationshipToParent: null },
518 forkedFrom: null,
519 createdBy: "user",
520 creationSource: "web",
521 activeRunId: null,
522 latestVisibleMessage: null,
523 hasActionableProposedPlan: false,
524 itemCount: 1,
525 visibleItemCount: 1,
526 branch: null,
527 status: "running",
528 pendingRuntimeRequest: null,
529 pendingBackgroundTasks: [],
530 latestRunId: null,
531 latestUserMessageAt: null,
532 createdAt: at(0),
533 updatedAt: at(0),
534 archivedAt: null,
535 settledOverride: null,
536 settledAt: null,
537 deletedAt: null,
538 ...overrides,
539 });
540 type Step =
541 | { readonly advanceMs: number }
542 | Exclude<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" | "synchronized" }>;
543 const updated = (sequence: number, thread: OrchestrationV2ThreadShell): Step => ({
544 kind: "thread.updated",
545 sequence,
546 location: "active",
547 thread,
548 });
549 const run = (steps: ReadonlyArray<Step>) =>
550 Stream.fromIterable(steps).pipe(
551 Stream.mapEffect((step) =>
552 "advanceMs" in step
553 ? TestClock.adjust(Duration.millis(step.advanceMs)).pipe(Effect.as(null))
554 : Effect.succeed(step),
555 ),
556 Stream.filter((item) => item !== null),
557 skipUnchangedThreadShells,
558 Stream.runCollect,
559 Effect.map((items) => Array.from(items, (item) => ("sequence" in item ? item.sequence : -1))),
560 );
561
562 it.effect("drops shells that only moved updatedAt, and resends them after a while", () =>
563 Effect.gen(function* () {
564 const sent = yield* run([
565 updated(1, shell("a")),
566 updated(2, shell("a", { updatedAt: at(100) })),
567 updated(3, shell("a", { updatedAt: at(200), itemCount: 2, visibleItemCount: 2 })),
568 updated(4, shell("a", { updatedAt: at(300), itemCount: 2, visibleItemCount: 2 })),
569 updated(5, shell("b")),
570 // The resend window is 5 s.
571 { advanceMs: 5_000 },
572 updated(6, shell("a", { updatedAt: at(5_300), itemCount: 2, visibleItemCount: 2 })),
573 ]);
574 expect(sent).toEqual([1, 3, 5, 6]);
575 }),
576 );
577
578 it.effect("sends the next update after a removal", () =>
579 Effect.gen(function* () {
580 const sent = yield* run([
581 updated(1, shell("a")),
582 { kind: "thread.removed", sequence: 2, location: "active", threadId: ThreadId.make("a") },
583 updated(3, shell("a")),
584 ]);
585 expect(sent).toEqual([1, 2, 3]);
586 }),
587 );
588});
apps/server/src/orchestration-v2/ShellStream.ts+57-1
...@@ -8,7 +8,12 @@ import type {...@@ -8,7 +8,12 @@ import type {
8 OrchestrationV2ShellStreamItem,8 OrchestrationV2ShellStreamItem,
9 OrchestrationV2StoredEvent,9 OrchestrationV2StoredEvent,
10} from "@t3tools/contracts";10} from "@t3tools/contracts";
11import { OrchestrationProjectShell as ProjectShellSchema } from "@t3tools/contracts";11import {
12 OrchestrationProjectShell as ProjectShellSchema,
13 OrchestrationV2ThreadShell as ThreadShellSchema,
14} from "@t3tools/contracts";
15import * as Clock from "effect/Clock";
16import * as Effect from "effect/Effect";
12import * as Schema from "effect/Schema";17import * as Schema from "effect/Schema";
13import * as Stream from "effect/Stream";18import * as Stream from "effect/Stream";
1419
...@@ -188,6 +193,57 @@ export function shellStreamItemsFromResumeSnapshot(input: {...@@ -188,6 +193,57 @@ export function shellStreamItemsFromResumeSnapshot(input: {
188 );193 );
189}194}
190195
196const sameThreadShell = Schema.toEquivalence(ThreadShellSchema);
197
198/** How long an unchanged thread shell may go unsent on a live subscription. */
199const UNCHANGED_THREAD_SHELL_RESEND_MS = 5_000;
200
201/**
202 * Drop live `thread.updated` deltas whose shell matches the last one this
203 * subscription sent, apart from `updatedAt`. Every thread event bumps
204 * `updatedAt`, so tool output and streaming text alone used to send the full,
205 * otherwise unchanged shell once per batch.
206 *
207 * A skipped delta leaves the client cursor behind, and a resume replays the
208 * events after that cursor. So an unchanged shell is still sent once its last
209 * send is `UNCHANGED_THREAD_SHELL_RESEND_MS` old. That bounds both the replay
210 * and how stale the client's `updatedAt` can get.
211 */
212export function skipUnchangedThreadShells<A extends OrchestrationV2ShellStreamItem, E, R>(
213 stream: Stream.Stream<A, E, R>,
214): Stream.Stream<A, E, R> {
215 return Stream.suspend(() => {
216 const lastSent = new Map<
217 OrchestrationV2ThreadShell["id"],
218 { readonly thread: OrchestrationV2ThreadShell; readonly sentAt: number }
219 >();
220 return stream.pipe(
221 Stream.filterEffect((item) =>
222 Effect.map(Clock.currentTimeMillis, (now) => {
223 if (item.kind === "thread.removed") {
224 lastSent.delete(item.threadId);
225 return true;
226 }
227 if (item.kind !== "thread.updated") return true;
228 const previous = lastSent.get(item.thread.id);
229 if (
230 previous !== undefined &&
231 now - previous.sentAt < UNCHANGED_THREAD_SHELL_RESEND_MS &&
232 sameThreadShell(
233 { ...item.thread, updatedAt: previous.thread.updatedAt },
234 previous.thread,
235 )
236 ) {
237 return false;
238 }
239 lastSent.set(item.thread.id, { thread: item.thread, sentAt: now });
240 return true;
241 }),
242 ),
243 );
244 });
245}
246
191/** Keep only the newest stored event per thread within a coalescing window. */247/** Keep only the newest stored event per thread within a coalescing window. */
192export function coalesceStoredThreadEvents(248export function coalesceStoredThreadEvents(
193 events: ReadonlyArray<OrchestrationV2StoredEvent>,249 events: ReadonlyArray<OrchestrationV2StoredEvent>,
apps/server/src/orchestration-v2/ThreadManagementService.test.ts+67-13
...@@ -5,6 +5,7 @@ import {...@@ -5,6 +5,7 @@ import {
5 NodeId,5 NodeId,
6 type OrchestrationV2Command,6 type OrchestrationV2Command,
7 type OrchestrationV2Run,7 type OrchestrationV2Run,
8 type OrchestrationV2StoredEvent,
8 type OrchestrationV2ThreadProjection,9 type OrchestrationV2ThreadProjection,
9 ProjectId,10 ProjectId,
10 ProviderInstanceId,11 ProviderInstanceId,
...@@ -16,6 +17,8 @@ import * as Duration from "effect/Duration";...@@ -16,6 +17,8 @@ import * as Duration from "effect/Duration";
16import * as Effect from "effect/Effect";17import * as Effect from "effect/Effect";
17import * as Fiber from "effect/Fiber";18import * as Fiber from "effect/Fiber";
18import * as Layer from "effect/Layer";19import * as Layer from "effect/Layer";
20import * as Queue from "effect/Queue";
21import * as Stream from "effect/Stream";
19import * as TestClock from "effect/testing/TestClock";22import * as TestClock from "effect/testing/TestClock";
2023
21import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";24import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";
...@@ -357,7 +360,7 @@ it.effect.each([...@@ -357,7 +360,7 @@ it.effect.each([
357 const projectId = ProjectId.make("project:thread-management:wait-timeout");360 const projectId = ProjectId.make("project:thread-management:wait-timeout");
358 const threadId = ThreadId.make("thread:thread-management:wait-timeout");361 const threadId = ThreadId.make("thread:thread-management:wait-timeout");
359 const runId = RunId.make("run:thread-management:wait-timeout");362 const runId = RunId.make("run:thread-management:wait-timeout");
360 const loopRead = yield* Deferred.make<void>();363 const subscribed = yield* Deferred.make<void>();
361 let reads = 0;364 let reads = 0;
362 const projection = (status: OrchestrationV2Run["status"] | "missing") =>365 const projection = (status: OrchestrationV2Run["status"] | "missing") =>
363 ({366 ({
...@@ -367,19 +370,18 @@ it.effect.each([...@@ -367,19 +370,18 @@ it.effect.each([
367 const testLayer = ThreadManagementService.layer.pipe(370 const testLayer = ThreadManagementService.layer.pipe(
368 Layer.provide(371 Layer.provide(
369 Layer.mock(Orchestrator.OrchestratorV2)({372 Layer.mock(Orchestrator.OrchestratorV2)({
373 getThreadEventSequence: () => Effect.succeed(0),
374 // No run update arrives, so the timeout path runs while a final
375 // projection read can still observe a terminal run.
376 streamStoredEventsFrom: () =>
377 Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
378 Stream.drain,
379 Stream.concat(Stream.never),
380 ),
370 getThreadRecords: () =>381 getThreadRecords: () =>
371 Effect.gen(function* () {382 Effect.sync(() => {
372 reads += 1;383 reads += 1;
373 if (reads === 1) {384 return projection(reads === 1 ? "running" : scenario.finalStatus);
374 return projection("running");
375 }
376 if (reads === 2) {
377 // Park inside the wait loop so the timeout path runs while a
378 // final projection read can still observe a terminal run.
379 yield* Deferred.succeed(loopRead, undefined);
380 return yield* Effect.never;
381 }
382 return projection(scenario.finalStatus);
383 }),385 }),
384 }),386 }),
385 ),387 ),
...@@ -395,7 +397,7 @@ it.effect.each([...@@ -395,7 +397,7 @@ it.effect.each([
395 timeoutMs: 1,397 timeoutMs: 1,
396 })398 })
397 .pipe(Effect.result, Effect.forkChild);399 .pipe(Effect.result, Effect.forkChild);
398 yield* Deferred.await(loopRead);400 yield* Deferred.await(subscribed);
399 yield* TestClock.adjust(Duration.millis(1));401 yield* TestClock.adjust(Duration.millis(1));
400 const result = yield* Fiber.join(fiber);402 const result = yield* Fiber.join(fiber);
401403
...@@ -419,3 +421,55 @@ it.effect.each([...@@ -419,3 +421,55 @@ it.effect.each([
419 }421 }
420 }),422 }),
421);423);
424
425it.effect("waitForThread reads the run again only when the run updates", () =>
426 Effect.gen(function* () {
427 const projectId = ProjectId.make("project:thread-management:wait-event");
428 const threadId = ThreadId.make("thread:thread-management:wait-event");
429 const runId = RunId.make("run:thread-management:wait-event");
430 const subscribed = yield* Deferred.make<void>();
431 const events = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
432 let status: OrchestrationV2Run["status"] = "running";
433 let reads = 0;
434 const stored = (sequence: number, event: object) =>
435 ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent;
436 const testLayer = ThreadManagementService.layer.pipe(
437 Layer.provide(
438 Layer.mock(Orchestrator.OrchestratorV2)({
439 getThreadEventSequence: () => Effect.succeed(0),
440 // Only the run.updated stream carries events in this test.
441 streamStoredEventsFrom: (input) =>
442 input?.eventType === "run.updated"
443 ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
444 Stream.drain,
445 Stream.concat(Stream.fromQueue(events)),
446 )
447 : Stream.never,
448 getThreadRecords: () =>
449 Effect.sync(() => {
450 reads += 1;
451 return {
452 thread: { id: threadId, projectId, deletedAt: null },
453 runs: [{ id: runId, status }],
454 } as unknown as OrchestrationV2ThreadProjection;
455 }),
456 }),
457 ),
458 );
459 const service = yield* ThreadManagementService.ThreadManagementService.pipe(
460 Effect.provide(testLayer),
461 );
462 const fiber = yield* service
463 .waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 })
464 .pipe(Effect.forkChild);
465 yield* Deferred.await(subscribed);
466 status = "completed";
467 yield* Queue.offer(events, stored(1, { type: "run.updated", payload: { id: "other-run" } }));
468 yield* Queue.offer(events, stored(2, { type: "run.updated", payload: { id: runId, status } }));
469 const result = yield* Fiber.join(fiber);
470
471 expect(result).toMatchObject({ timedOut: false, run: { id: runId, status: "completed" } });
472 // The first read plus one for this run's update. The other run caused none.
473 expect(reads).toBe(2);
474 }),
475);
apps/server/src/orchestration-v2/ThreadManagementService.ts+52-18
...@@ -32,6 +32,7 @@ import * as Effect from "effect/Effect";...@@ -32,6 +32,7 @@ import * as Effect from "effect/Effect";
32import * as Layer from "effect/Layer";32import * as Layer from "effect/Layer";
33import * as Option from "effect/Option";33import * as Option from "effect/Option";
34import * as Schema from "effect/Schema";34import * as Schema from "effect/Schema";
35import * as Stream from "effect/Stream";
3536
36import * as Orchestrator from "./Orchestrator.ts";37import * as Orchestrator from "./Orchestrator.ts";
37import { projectTurnItemForDetail } from "./WireProjection.ts";38import { projectTurnItemForDetail } from "./WireProjection.ts";
...@@ -131,7 +132,6 @@ export interface ThreadManagementWaitInput {...@@ -131,7 +132,6 @@ export interface ThreadManagementWaitInput {
131 readonly threadId: ThreadId;132 readonly threadId: ThreadId;
132 readonly runId?: RunId;133 readonly runId?: RunId;
133 readonly timeoutMs: number;134 readonly timeoutMs: number;
134 readonly pollIntervalMs?: number;
135}135}
136136
137export interface ThreadManagementWaitResult {137export interface ThreadManagementWaitResult {
...@@ -632,6 +632,17 @@ const make = Effect.gen(function* () {...@@ -632,6 +632,17 @@ const make = Effect.gen(function* () {
632632
633 const waitForThread: ThreadManagementServiceShape["waitForThread"] = (input) =>633 const waitForThread: ThreadManagementServiceShape["waitForThread"] = (input) =>
634 Effect.gen(function* () {634 Effect.gen(function* () {
635 const loadError = (cause: unknown) =>
636 new ThreadManagementProjectionLoadError({
637 projectId: input.projectId,
638 threadId: input.threadId,
639 cause,
640 });
641 // Taken before the first read, so a run update between that read and
642 // the subscription below still replays.
643 const afterSequence = yield* orchestrator
644 .getThreadEventSequence(input.threadId)
645 .pipe(Effect.mapError(loadError));
635 const target = yield* getProjectThreadRecords(input, ["runs"]);646 const target = yield* getProjectThreadRecords(input, ["runs"]);
636 const selectedRun =647 const selectedRun =
637 input.runId === undefined648 input.runId === undefined
...@@ -650,23 +661,46 @@ const make = Effect.gen(function* () {...@@ -650,23 +661,46 @@ const make = Effect.gen(function* () {
650 return { threadId: input.threadId, run: selectedRun, timedOut: false };661 return { threadId: input.threadId, run: selectedRun, timedOut: false };
651 }662 }
652663
653 const wait = Effect.gen(function* () {664 // Re-read the run only when the thread records a run update or its
654 while (true) {665 // deletion, instead of polling the projection for up to an hour. Each
655 const current = yield* getProjectThreadRecords(input, ["runs"], {666 // stream keeps one event type, so transcript events never fill its buffer.
656 runIds: [selectedRun.id],667 const wait = Stream.merge(
657 });668 orchestrator.streamStoredEventsFrom({
658 const run = current.runs.find((candidate) => candidate.id === selectedRun.id);669 threadId: input.threadId,
659 if (run === undefined) {670 afterSequence,
660 return yield* new ThreadManagementRunNotFoundError({671 eventType: "run.updated",
661 threadId: input.threadId,672 }),
662 runId: selectedRun.id,673 orchestrator.streamStoredEventsFrom({
663 });674 threadId: input.threadId,
664 }675 afterSequence,
665 if (isTerminalRunStatus(run.status)) return run;676 eventType: "thread.deleted",
666 yield* Effect.sleep(Duration.millis(Math.max(1, input.pollIntervalMs ?? 250)));677 }),
667 }678 ).pipe(
668 }).pipe(Effect.timeoutOption(Duration.millis(Math.max(1, input.timeoutMs))));679 Stream.mapError(loadError),
669 const waited = yield* wait;680 Stream.filter(
681 (stored) =>
682 stored.event.type !== "run.updated" || stored.event.payload.id === selectedRun.id,
683 ),
684 Stream.mapEffect(() =>
685 getProjectThreadRecords(input, ["runs"], { runIds: [selectedRun.id] }).pipe(
686 Effect.flatMap((current) => {
687 const run = current.runs.find((candidate) => candidate.id === selectedRun.id);
688 return run === undefined
689 ? Effect.fail(
690 new ThreadManagementRunNotFoundError({
691 threadId: input.threadId,
692 runId: selectedRun.id,
693 }),
694 )
695 : Effect.succeed(run);
696 }),
697 ),
698 ),
699 Stream.filter((run) => isTerminalRunStatus(run.status)),
700 Stream.runHead,
701 Effect.timeoutOption(Duration.millis(Math.max(1, input.timeoutMs))),
702 );
703 const waited = Option.flatten(yield* wait);
670 if (Option.isSome(waited)) {704 if (Option.isSome(waited)) {
671 return { threadId: input.threadId, run: waited.value, timedOut: false };705 return { threadId: input.threadId, run: waited.value, timedOut: false };
672 }706 }
apps/server/src/ws.ts+2
...@@ -132,6 +132,7 @@ import {...@@ -132,6 +132,7 @@ import {
132 shellStreamItemFromThreadShell,132 shellStreamItemFromThreadShell,
133 shellStreamItemsFromInitialSnapshot,133 shellStreamItemsFromInitialSnapshot,
134 shellStreamItemsFromResumeSnapshot,134 shellStreamItemsFromResumeSnapshot,
135 skipUnchangedThreadShells,
135 toShellApplicationEvent,136 toShellApplicationEvent,
136 type ShellApplicationEvent,137 type ShellApplicationEvent,
137} from "./orchestration-v2/ShellStream.ts";138} from "./orchestration-v2/ShellStream.ts";
...@@ -1030,6 +1031,7 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc...@@ -1030,6 +1031,7 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc
1030 Stream.groupedWithin(512, Duration.millis(50)),1031 Stream.groupedWithin(512, Duration.millis(50)),
1031 Stream.mapEffect((events) => projectShellItems(Array.from(events))),1032 Stream.mapEffect((events) => projectShellItems(Array.from(events))),
1032 Stream.flatMap(Stream.fromIterable),1033 Stream.flatMap(Stream.fromIterable),
1034 skipUnchangedThreadShells,
1033 );1035 );
10341036
1035 const liveFrom = (afterSequence: number) =>1037 const liveFrom = (afterSequence: number) =>
apps/web/src/components/ThreadNotificationCoordinator.tsx+23-10
...@@ -1,7 +1,7 @@...@@ -1,7 +1,7 @@
1import { presentThreadShell } from "@t3tools/client-runtime/state/models";1import { presentThreadShell } from "@t3tools/client-runtime/state/models";
2import { useAtomValue } from "@effect/atom-react";2import { useAtomValue } from "@effect/atom-react";
3import { useNavigate, useParams } from "@tanstack/react-router";3import { useNavigate, useParams } from "@tanstack/react-router";
4import type { EnvironmentId, ThreadId } from "@t3tools/contracts";4import type { EnvironmentId, OrchestrationV2ThreadShell, ThreadId } from "@t3tools/contracts";
5import * as Option from "effect/Option";5import * as Option from "effect/Option";
6import {6import {
7 CircleAlertIcon,7 CircleAlertIcon,
...@@ -88,6 +88,12 @@ export function ThreadNotificationCoordinator() {...@@ -88,6 +88,12 @@ export function ThreadNotificationCoordinator() {
88 ));88 ));
89}89}
9090
91interface NotificationState {
92 readonly raw: OrchestrationV2ThreadShell;
93 readonly attention: string | null;
94 readonly completion: number | null;
95}
96
91function EnvironmentNotifications({97function EnvironmentNotifications({
92 environmentId,98 environmentId,
93 onNotification,99 onNotification,
...@@ -96,6 +102,10 @@ function EnvironmentNotifications({...@@ -96,6 +102,10 @@ function EnvironmentNotifications({
96 onNotification: (environmentId: EnvironmentId, notification: Notification) => void;102 onNotification: (environmentId: EnvironmentId, notification: Notification) => void;
97}) {103}) {
98 const shell = useAtomValue(environmentShell.stateValueAtom(environmentId));104 const shell = useAtomValue(environmentShell.stateValueAtom(environmentId));
105 // The shell reducer keeps the thread list and unchanged thread objects
106 // stable, so this only rescans when a thread actually changed.
107 const threads =
108 shell.status === "live" && Option.isSome(shell.snapshot) ? shell.snapshot.value.threads : null;
99 const mode = useClientSettings((settings) => settings.notificationMode);109 const mode = useClientSettings((settings) => settings.notificationMode);
100 const inAppNotificationsEnabled = useClientSettings(110 const inAppNotificationsEnabled = useClientSettings(
101 (settings) => settings.inAppNotificationsEnabled,111 (settings) => settings.inAppNotificationsEnabled,
...@@ -104,22 +114,25 @@ function EnvironmentNotifications({...@@ -104,22 +114,25 @@ function EnvironmentNotifications({
104 const { environmentId: activeEnvironmentId, threadId: activeThreadId } = useParams({114 const { environmentId: activeEnvironmentId, threadId: activeThreadId } = useParams({
105 strict: false,115 strict: false,
106 });116 });
107 const previous = useRef(117 const previous = useRef(new Map<ThreadId, NotificationState>());
108 new Map<ThreadId, { attention: string | null; completion: number | null }>(),
109 );
110118
111 useEffect(() => {119 useEffect(() => {
112 if (shell.status !== "live" || Option.isNone(shell.snapshot)) {120 if (threads === null) {
113 previous.current.clear();121 previous.current.clear();
114 return;122 return;
115 }123 }
116 const next = new Map<ThreadId, { attention: string | null; completion: number | null }>();124 const next = new Map<ThreadId, NotificationState>();
117 for (const rawThread of shell.snapshot.value.threads) {125 for (const rawThread of threads) {
118 if (rawThread.lineage.relationshipToParent === "subagent") continue;126 if (rawThread.lineage.relationshipToParent === "subagent") continue;
127 const prior = previous.current.get(rawThread.id);
128 // The same object cannot produce a new notification.
129 if (prior?.raw === rawThread) {
130 next.set(rawThread.id, prior);
131 continue;
132 }
119 const thread = presentThreadShell(environmentId, rawThread);133 const thread = presentThreadShell(environmentId, rawThread);
120 let status = resolveSidebarThreadStatus(thread);134 let status = resolveSidebarThreadStatus(thread);
121 if (status === "ready" && thread.latestRun?.status === "failed") status = "failed";135 if (status === "ready" && thread.latestRun?.status === "failed") status = "failed";
122 const prior = previous.current.get(thread.id);
123 const attention =136 const attention =
124 status === "input" || status === "approval" || status === "failed" || status === "limited"137 status === "input" || status === "approval" || status === "failed" || status === "limited"
125 ? `${thread.latestRun?.runId ?? ""}:${status}`138 ? `${thread.latestRun?.runId ?? ""}:${status}`
...@@ -132,7 +145,7 @@ function EnvironmentNotifications({...@@ -132,7 +145,7 @@ function EnvironmentNotifications({
132 Number.isFinite(completedAt)145 Number.isFinite(completedAt)
133 ? completedAt146 ? completedAt
134 : (prior?.completion ?? null);147 : (prior?.completion ?? null);
135 next.set(thread.id, { attention, completion });148 next.set(thread.id, { raw: rawThread, attention, completion });
136 if (!prior || thread.archivedAt !== null) continue;149 if (!prior || thread.archivedAt !== null) continue;
137 const kind =150 const kind =
138 attention && attention !== prior.attention151 attention && attention !== prior.attention
...@@ -227,7 +240,7 @@ function EnvironmentNotifications({...@@ -227,7 +240,7 @@ function EnvironmentNotifications({
227 mode,240 mode,
228 navigate,241 navigate,
229 onNotification,242 onNotification,
230 shell,243 threads,
231 ]);244 ]);
232245
233 return null;246 return null;
apps/web/src/components/chat/ChatComposer.tsx+15-7
...@@ -2170,12 +2170,19 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)...@@ -2170,12 +2170,19 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
2170 [selectedProviderEntry],2170 [selectedProviderEntry],
2171 );2171 );
2172 const compactCommandAvailable = providerSupportsManualCompaction(selectedProviderEntry);2172 const compactCommandAvailable = providerSupportsManualCompaction(selectedProviderEntry);
2173 const selectedProviderSkills = selectedProviderStatus2173 // Memoized so the composer menu memo below can cache between renders.
2174 ? resolveProviderSkillsForCwd(selectedProviderStatus, gitCwd)2174 const selectedProviderSkills = useMemo(
2175 : [];2175 () =>
2176 const selectedProviderSlashCommands = selectedProviderStatus2176 selectedProviderStatus ? resolveProviderSkillsForCwd(selectedProviderStatus, gitCwd) : [],
2177 ? resolveProviderSlashCommandsForCwd(selectedProviderStatus, gitCwd)2177 [gitCwd, selectedProviderStatus],
2178 : [];2178 );
2179 const selectedProviderSlashCommands = useMemo(
2180 () =>
2181 selectedProviderStatus
2182 ? resolveProviderSlashCommandsForCwd(selectedProviderStatus, gitCwd)
2183 : [],
2184 [gitCwd, selectedProviderStatus],
2185 );
2179 const refreshProviders = useAtomCommand(serverEnvironment.refreshProviders, {2186 const refreshProviders = useAtomCommand(serverEnvironment.refreshProviders, {
2180 reportFailure: false,2187 reportFailure: false,
2181 });2188 });
...@@ -2526,7 +2533,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)...@@ -2526,7 +2533,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
2526 const settledPullRequestTextQuery =2533 const settledPullRequestTextQuery =
2527 pullRequestTextQuery === debouncedPullRequestTextQuery ? pullRequestTextQuery : null;2534 pullRequestTextQuery === debouncedPullRequestTextQuery ? pullRequestTextQuery : null;
2528 const isPathTrigger = composerTriggerKind === "path";2535 const isPathTrigger = composerTriggerKind === "path";
2529 const environmentThreadShells = useThreadShells();2536 // Thread shells only feed `@` thread matches, so skip shell updates otherwise.
2537 const environmentThreadShells = useThreadShells(isPathTrigger);
2530 const workspaceEntries = useComposerPathSearch({2538 const workspaceEntries = useComposerPathSearch({
2531 environmentId,2539 environmentId,
2532 cwd: isPathTrigger ? gitCwd : null,2540 cwd: isPathTrigger ? gitCwd : null,
apps/web/src/components/chat/MessagesTimeline.tsx+3-10
...@@ -3525,7 +3525,7 @@ function LiveActivityRow({...@@ -3525,7 +3525,7 @@ function LiveActivityRow({
3525 const showShimmer = animated && shimmer;3525 const showShimmer = animated && shimmer;
3526 return (3526 return (
3527 <div3527 <div
3528 ref={animated ? observeVisibleAnimation : undefined}3528 ref={showShimmer ? observeVisibleAnimation : undefined}
3529 className="relative min-h-6 w-fit max-w-full min-w-0 overflow-hidden rounded-md text-sm leading-relaxed"3529 className="relative min-h-6 w-fit max-w-full min-w-0 overflow-hidden rounded-md text-sm leading-relaxed"
3530 >3530 >
3531 <LiveActivityContent3531 <LiveActivityContent
...@@ -3591,13 +3591,7 @@ function LiveActivityContent({...@@ -3591,13 +3591,7 @@ function LiveActivityContent({
3591 ) : null3591 ) : null
3592 }3592 }
3593 label={3593 label={
3594 <span3594 <span className={cn("block truncate", (highlighted || active) && "text-foreground")}>
3595 className={cn(
3596 "block truncate",
3597 highlighted && "text-foreground",
3598 active && "live-tool-shine",
3599 )}
3600 >
3601 {label}3595 {label}
3602 </span>3596 </span>
3603 }3597 }
...@@ -3733,7 +3727,6 @@ function WorkGroupHeader(props: {...@@ -3733,7 +3727,6 @@ function WorkGroupHeader(props: {
3733}) {3727}) {
3734 return (3728 return (
3735 <WorkLogButton3729 <WorkLogButton
3736 ref={props.active && !props.failed ? observeVisibleAnimation : undefined}
3737 aria-label={props.failed ? `${props.label}, tool call failed` : props.label}3730 aria-label={props.failed ? `${props.label}, tool call failed` : props.label}
3738 aria-expanded={props.expanded}3731 aria-expanded={props.expanded}
3739 onClick={props.onToggle}3732 onClick={props.onToggle}
...@@ -3746,7 +3739,7 @@ function WorkGroupHeader(props: {...@@ -3746,7 +3739,7 @@ function WorkGroupHeader(props: {
3746 />3739 />
3747 }3740 }
3748 label={3741 label={
3749 <span className={cn("block truncate", props.active && !props.failed && "live-tool-shine")}>3742 <span className={cn("block truncate", props.active && !props.failed && "text-foreground")}>
3750 {props.label}3743 {props.label}
3751 </span>3744 </span>
3752 }3745 }
apps/web/src/components/chat/ThreadAutomationsPanel.tsx+1-1
...@@ -25,7 +25,7 @@ import {...@@ -25,7 +25,7 @@ import {
2525
26const STATUS_DOT_CLASS: Record<ScheduledTask["lastRunStatus"], string> = {26const STATUS_DOT_CLASS: Record<ScheduledTask["lastRunStatus"], string> = {
27 never: "bg-muted-foreground/40",27 never: "bg-muted-foreground/40",
28 running: "animate-pulse bg-sky-500",28 running: "bg-sky-500",
29 succeeded: "bg-emerald-500",29 succeeded: "bg-emerald-500",
30 failed: "bg-destructive",30 failed: "bg-destructive",
31};31};
apps/web/src/index.css-29
...@@ -499,35 +499,6 @@ html[data-mobile-composer-route-transition="true"]::view-transition-old(t3-mobil...@@ -499,35 +499,6 @@ html[data-mobile-composer-route-transition="true"]::view-transition-old(t3-mobil
499 }499 }
500}500}
501501
502@keyframes live-tool-shine {
503 from {
504 background-position:
505 -4.5rem 0,
506 0 0;
507 }
508 to {
509 background-position:
510 calc(100% + 4.5rem) 0,
511 0 0;
512 }
513}
514
515@utility live-tool-shine {
516 @media (prefers-reduced-motion: no-preference) and (forced-colors: none) {
517 color: transparent;
518 background-image:
519 linear-gradient(to right, transparent, var(--contrast-foreground), transparent),
520 linear-gradient(var(--contrast-secondary-label), var(--contrast-secondary-label));
521 background-size:
522 4.5rem 100%,
523 100% 100%;
524 background-repeat: no-repeat;
525 background-clip: text;
526 animation: live-tool-shine 2.2s steps(30) infinite;
527 animation-play-state: var(--visible-animation-state, paused);
528 }
529}
530
531@utility visible-animate-spin {502@utility visible-animate-spin {
532 animation: var(--animate-spin);503 animation: var(--animate-spin);
533 animation-play-state: var(--visible-animation-state, paused);504 animation-play-state: var(--visible-animation-state, paused);
apps/web/src/lib/composerPathSearchState.ts+8-4
...@@ -3,15 +3,19 @@ import {...@@ -3,15 +3,19 @@ import {
3 type ComposerPathSearchTarget,3 type ComposerPathSearchTarget,
4} from "@t3tools/client-runtime/state/threads";4} from "@t3tools/client-runtime/state/threads";
55
6import { useMemo } from "react";
7
6import { useComposerPathSearch as useComposerPathSearchQuery } from "../state/queries";8import { useComposerPathSearch as useComposerPathSearchQuery } from "../state/queries";
79
8export function useComposerPathSearch(target: ComposerPathSearchTarget): ComposerPathSearchState {10export function useComposerPathSearch(target: ComposerPathSearchTarget): ComposerPathSearchState {
9 const state = useComposerPathSearchQuery(target);11 const state = useComposerPathSearchQuery(target);
12 // A stable list lets the composer menu memo cache between renders.
13 const entries = useMemo(
14 () => state.entries.map((entry) => ({ path: entry.path, kind: entry.kind })),
15 [state.entries],
16 );
10 return {17 return {
11 entries: state.entries.map((entry) => ({18 entries,
12 path: entry.path,
13 kind: entry.kind,
14 })),
15 error: state.error,19 error: state.error,
16 isPending: state.isPending,20 isPending: state.isPending,
17 };21 };
apps/web/src/state/entities.ts+9-2
...@@ -31,6 +31,9 @@ const EMPTY_PROJECT_ATOM = Atom.make<EnvironmentProject | null>(null).pipe(...@@ -31,6 +31,9 @@ const EMPTY_PROJECT_ATOM = Atom.make<EnvironmentProject | null>(null).pipe(
31const EMPTY_THREAD_REFS_ATOM = Atom.make(EMPTY_THREAD_REFS).pipe(31const EMPTY_THREAD_REFS_ATOM = Atom.make(EMPTY_THREAD_REFS).pipe(
32 Atom.withLabel("web-thread-refs:empty"),32 Atom.withLabel("web-thread-refs:empty"),
33);33);
34const EMPTY_THREAD_SHELLS_ATOM = Atom.make<ReadonlyArray<EnvironmentThreadShell>>(
35 Object.freeze([]),
36).pipe(Atom.withLabel("web-thread-shells:empty"));
34const EMPTY_THREAD_SHELL_ATOM = Atom.make<EnvironmentThreadShell | null>(null).pipe(37const EMPTY_THREAD_SHELL_ATOM = Atom.make<EnvironmentThreadShell | null>(null).pipe(
35 Atom.withLabel("web-thread-shell:empty"),38 Atom.withLabel("web-thread-shell:empty"),
36);39);
...@@ -82,8 +85,12 @@ export function useServerConfigs(): ReadonlyMap<EnvironmentId, ServerConfig> {...@@ -82,8 +85,12 @@ export function useServerConfigs(): ReadonlyMap<EnvironmentId, ServerConfig> {
82 return useAtomValue(environmentServerConfigsAtom);85 return useAtomValue(environmentServerConfigsAtom);
83}86}
8487
85export function useThreadShells(): ReadonlyArray<EnvironmentThreadShell> {88/** Every thread shell. Pass `enabled: false` to read a stable empty list and
86 return useAtomValue(environmentThreadShells.threadShellsAtom);89 skip re-rendering on each shell update while the caller does not need them. */
90export function useThreadShells(enabled = true): ReadonlyArray<EnvironmentThreadShell> {
91 return useAtomValue(
92 enabled ? environmentThreadShells.threadShellsAtom : EMPTY_THREAD_SHELLS_ATOM,
93 );
87}94}
8895
89export function useAllEnvironmentShellsBootstrapped(): boolean {96export function useAllEnvironmentShellsBootstrapped(): boolean {
apps/web/src/state/queries.ts+3-1
...@@ -12,6 +12,7 @@ import { type VcsRefTarget } from "@t3tools/client-runtime/state/vcs";...@@ -12,6 +12,7 @@ import { type VcsRefTarget } from "@t3tools/client-runtime/state/vcs";
12import type {12import type {
13 EnvironmentId,13 EnvironmentId,
14 ProjectContentMatch,14 ProjectContentMatch,
15 ProjectEntry,
15 ProjectEntryKind,16 ProjectEntryKind,
16 ThreadId,17 ThreadId,
17 TurnItemId,18 TurnItemId,
...@@ -37,6 +38,7 @@ const PROJECT_CONTENT_SEARCH_LIMIT = 500;...@@ -37,6 +38,7 @@ const PROJECT_CONTENT_SEARCH_LIMIT = 500;
37const THREAD_SEARCH_DEBOUNCE_MS = 200;38const THREAD_SEARCH_DEBOUNCE_MS = 200;
38const VCS_REF_LIST_LIMIT = 100;39const VCS_REF_LIST_LIMIT = 100;
39const EMPTY_REFS: ReadonlyArray<VcsRef> = [];40const EMPTY_REFS: ReadonlyArray<VcsRef> = [];
41const EMPTY_PROJECT_ENTRIES: ReadonlyArray<ProjectEntry> = [];
40const EMPTY_CONTENT_MATCHES: ReadonlyArray<ProjectContentMatch> = [];42const EMPTY_CONTENT_MATCHES: ReadonlyArray<ProjectContentMatch> = [];
41const INITIAL_BRANCH_CURSORS = [undefined] as const;43const INITIAL_BRANCH_CURSORS = [undefined] as const;
42const EMPTY_THREAD_SEARCH_MATCHES: ReadonlyArray<EnvironmentThreadSearchMatch> = Object.freeze([]);44const EMPTY_THREAD_SEARCH_MATCHES: ReadonlyArray<EnvironmentThreadSearchMatch> = Object.freeze([]);
...@@ -254,7 +256,7 @@ export function useProjectPathSearch(...@@ -254,7 +256,7 @@ export function useProjectPathSearch(
254 );256 );
255257
256 return {258 return {
257 entries: result.data?.entries ?? [],259 entries: result.data?.entries ?? EMPTY_PROJECT_ENTRIES,
258 error: result.error,260 error: result.error,
259 isPending:261 isPending:
260 !areProjectPathSearchTargetsEqual(normalizedTarget, debouncedTarget) || result.isPending,262 !areProjectPathSearchTargetsEqual(normalizedTarget, debouncedTarget) || result.isPending,
packages/client-runtime/src/state/shellReducer.test.ts+15
...@@ -41,6 +41,21 @@ describe("applyShellStreamEvent", () => {...@@ -41,6 +41,21 @@ describe("applyShellStreamEvent", () => {
41 expect(next.threads[2]).toBe(threads[2]);41 expect(next.threads[2]).toBe(threads[2]);
42 });42 });
4343
44 it("keeps the thread list and object when an update changes nothing", () => {
45 const threads = ["a", "b"].map((id) => ({ ...v2ThreadShell, id: ThreadId.make(id) }));
46 const next = applyShellStreamEvent(
47 { ...v2ShellSnapshot, threads },
48 {
49 kind: "thread.updated",
50 sequence: 1,
51 location: "active",
52 thread: { ...threads[1]! },
53 },
54 );
55 expect(next.threads).toBe(threads);
56 expect(next.snapshotSequence).toBe(1);
57 });
58
44 it("ignores stale project updates without mutating the snapshot", () => {59 it("ignores stale project updates without mutating the snapshot", () => {
45 const snapshotWithProject = {60 const snapshotWithProject = {
46 ...v2ShellSnapshot,61 ...v2ShellSnapshot,
packages/client-runtime/src/state/shellReducer.ts+17-4
...@@ -1,8 +1,12 @@...@@ -1,8 +1,12 @@
1import type {1import {
2 OrchestrationProjectShell,2 OrchestrationV2ThreadShell,
3 OrchestrationV2ShellSnapshot,3 type OrchestrationProjectShell,
4 OrchestrationV2ShellStreamItem,4 type OrchestrationV2ShellSnapshot,
5 type OrchestrationV2ShellStreamItem,
5} from "@t3tools/contracts";6} from "@t3tools/contracts";
7import * as Schema from "effect/Schema";
8
9const sameThreadShell = Schema.toEquivalence(OrchestrationV2ThreadShell);
610
7function upsertById<T extends { readonly id: unknown }>(11function upsertById<T extends { readonly id: unknown }>(
8 items: ReadonlyArray<T>,12 items: ReadonlyArray<T>,
...@@ -123,6 +127,15 @@ export function applyShellStreamEvent(...@@ -123,6 +127,15 @@ export function applyShellStreamEvent(
123 snapshotSequence: event.sequence,127 snapshotSequence: event.sequence,
124 };128 };
125 case "thread.updated": {129 case "thread.updated": {
130 // An unchanged shell keeps its object and the list, so subscribers that
131 // compare by reference skip the update. Only the cursor moves.
132 const existing =
133 event.location === "active"
134 ? snapshot.threads.find((thread) => thread.id === event.thread.id)
135 : undefined;
136 if (existing !== undefined && sameThreadShell(existing, event.thread)) {
137 return { ...snapshot, snapshotSequence: event.sequence };
138 }
126 const withoutThread = (threads: OrchestrationV2ShellSnapshot["threads"]) =>139 const withoutThread = (threads: OrchestrationV2ShellSnapshot["threads"]) =>
127 threads.filter((thread) => thread.id !== event.thread.id);140 threads.filter((thread) => thread.id !== event.thread.id);
128 return {141 return {