From bc23ec507316e7e2b1aeff396b645255f4bedada Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 9 Jul 2026 04:09:08 -0700 Subject: [PATCH 1/4] perf: fix sustained-CPU hot paths from the July 2026 web performance audit Addresses the audit's findings with minimal, desktop-first changes (no payload deferral, no streaming-path work since streaming is off by default): - Sidebar: replace the 10-thread bulk detail prewarm with a single hover-based prewarm; at idle one shell stream + at most one detail stream - Server replay: thread catch-up reads only the thread's aggregate stream via the (aggregate_kind, stream_id, sequence) index instead of scanning the global log; shell catch-up is bounded (500 events) with a shell snapshot fallback for stale cursors - Shell summaries: thread.message-sent updates latestUserMessageAt incrementally; activity-appended triggers the history-wide summary rebuild only for approval/user-input activity kinds; streaming assistant chunks no longer fan out per-chunk shell upserts - Client runtime: O(1) activity appends when order is preserved; skip IndexedDB full-snapshot encodes while a turn is running (settle event and scope finalizer persist the final state) - Timeline derivation: share one ordered activities view across the four derivations (identity-cached, O(n) sorted check); k-way merge for the combined timeline instead of concat+sort - Composer drafts: debounce JSON serialization itself, and cache per-draft partialization by object identity so typing doesn't re-clone every retained draft - Shiki: negative-cache languages without grammars (sh, yaml) so those fences stop retrying and throwing on every render - Bootstrap polling: one shared change-detecting poller instead of one 2s interval per consumer; minimap only writes data-in-view on change Co-Authored-By: Claude Fable 5 --- .../Layers/OrchestrationEngine.test.ts | 21 +++ .../Layers/OrchestrationEngine.ts | 9 + .../Layers/ProjectionPipeline.test.ts | 101 ++++++++++ .../Layers/ProjectionPipeline.ts | 63 ++++++- .../Services/OrchestrationEngine.ts | 12 ++ .../Layers/OrchestrationEventStore.test.ts | 45 +++++ .../Layers/OrchestrationEventStore.ts | 94 ++++++++-- .../Services/OrchestrationEventStore.ts | 20 ++ .../src/relay/AgentAwarenessRelay.test.ts | 2 + apps/server/src/server.test.ts | 1 + apps/server/src/serverRuntimeStartup.test.ts | 3 + apps/server/src/ws.ts | 74 ++++++-- apps/web/src/components/ChatMarkdown.tsx | 16 +- apps/web/src/components/Sidebar.logic.test.ts | 80 +++++++- apps/web/src/components/Sidebar.logic.ts | 73 +++++++- apps/web/src/components/Sidebar.tsx | 51 ++++-- .../src/components/chat/MessagesTimeline.tsx | 8 +- apps/web/src/composerDraftStore.ts | 172 ++++++++++-------- .../connection/useDesktopLocalBootstraps.ts | 76 ++++++-- apps/web/src/lib/storage.ts | 43 +++++ apps/web/src/session-logic.ts | 80 +++++++- .../client-runtime/src/state/threadReducer.ts | 21 ++- .../src/state/threads-sync.test.ts | 62 +++++++ packages/client-runtime/src/state/threads.ts | 9 +- 24 files changed, 963 insertions(+), 173 deletions(-) diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 405103450e8..3a974b69882 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -102,6 +102,7 @@ describe("OrchestrationEngine", () => { return savedEvent; }), readFromSequence: () => Stream.empty, + readAggregateFromSequence: () => Stream.empty, readAll: () => Stream.fail( new PersistenceSqlError({ @@ -772,6 +773,16 @@ describe("OrchestrationEngine", () => { readFromSequence(sequenceExclusive) { return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive)); }, + readAggregateFromSequence(aggregateKind, aggregateId, sequenceExclusive) { + return Stream.fromIterable( + events.filter( + (event) => + event.aggregateKind === aggregateKind && + event.aggregateId === aggregateId && + event.sequence > sequenceExclusive, + ), + ); + }, readAll() { return Stream.fromIterable(events); }, @@ -1004,6 +1015,16 @@ describe("OrchestrationEngine", () => { readFromSequence(sequenceExclusive) { return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive)); }, + readAggregateFromSequence(aggregateKind, aggregateId, sequenceExclusive) { + return Stream.fromIterable( + events.filter( + (event) => + event.aggregateKind === aggregateKind && + event.aggregateId === aggregateId && + event.sequence > sequenceExclusive, + ), + ); + }, readAll() { return Stream.fromIterable(events); }, diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 9598d1c3ef5..1d1c9e97703 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -309,6 +309,14 @@ const makeOrchestrationEngine = Effect.gen(function* () { const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) => eventStore.readFromSequence(fromSequenceExclusive, limit); + const readAggregateEvents: OrchestrationEngineShape["readAggregateEvents"] = ( + aggregateKind, + aggregateId, + fromSequenceExclusive, + limit, + ) => + eventStore.readAggregateFromSequence(aggregateKind, aggregateId, fromSequenceExclusive, limit); + const dispatch: OrchestrationEngineShape["dispatch"] = (command) => Effect.gen(function* () { const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>(); @@ -322,6 +330,7 @@ const makeOrchestrationEngine = Effect.gen(function* () { return { readEvents, + readAggregateEvents, dispatch, // Each access creates a fresh PubSub subscription so that multiple // consumers (wsServer, ProviderRuntimeIngestion, CheckpointReactor, etc.) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 0999000ed4f..df4475b94de 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -2013,6 +2013,107 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect("updates shell summary fields incrementally for message events", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); + + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.make("evt-incremental-1"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-incremental"), + occurredAt: "2026-02-26T14:00:00.000Z", + commandId: CommandId.make("cmd-incremental-1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-incremental-1"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-incremental"), + projectId: ProjectId.make("project-incremental"), + title: "Thread Incremental", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: "2026-02-26T14:00:00.000Z", + updatedAt: "2026-02-26T14:00:00.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.make("evt-incremental-2"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-incremental"), + occurredAt: "2026-02-26T14:00:01.000Z", + commandId: CommandId.make("cmd-incremental-2"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-incremental-2"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-incremental"), + messageId: MessageId.make("message-incremental-user"), + role: "user", + text: "Do the thing", + turnId: null, + streaming: false, + createdAt: "2026-02-26T14:00:01.000Z", + updatedAt: "2026-02-26T14:00:01.000Z", + }, + }); + + // A streaming assistant chunk must not move updatedAt (no shell fan-out). + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.make("evt-incremental-3"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-incremental"), + occurredAt: "2026-02-26T14:00:02.000Z", + commandId: CommandId.make("cmd-incremental-3"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-incremental-3"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-incremental"), + messageId: MessageId.make("message-incremental-assistant"), + role: "assistant", + text: "Work", + turnId: null, + streaming: true, + createdAt: "2026-02-26T14:00:02.000Z", + updatedAt: "2026-02-26T14:00:02.000Z", + }, + }); + + const threadRows = yield* sql<{ + readonly latestUserMessageAt: string | null; + readonly updatedAt: string; + }>` + SELECT + latest_user_message_at AS "latestUserMessageAt", + updated_at AS "updatedAt" + FROM projection_threads + WHERE thread_id = 'thread-incremental' + `; + assert.deepEqual(threadRows, [ + { + latestUserMessageAt: "2026-02-26T14:00:01.000Z", + updatedAt: "2026-02-26T14:00:01.000Z", + }, + ]); + }), + ); + it.effect("ignores non-stale provider approval response failures", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index f12df850941..aeee73a2557 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -176,6 +176,19 @@ function derivePendingUserInputCountFromActivities( return openRequestIds.size; } +// Activity kinds that can change the shell summary's pending-approval or +// pending-user-input counters. All other activity kinds (command output, +// file edits, streaming progress, ...) leave the summary untouched, so they +// must not trigger the history-wide summary rebuild. +const SHELL_SUMMARY_ACTIVITY_KINDS: ReadonlySet = new Set([ + "approval.requested", + "approval.resolved", + "provider.approval.respond.failed", + "user-input.requested", + "user-input.resolved", + "provider.user-input.respond.failed", +]); + function deriveHasActionableProposedPlan(input: { readonly latestTurnId: string | null; readonly proposedPlans: ReadonlyArray; @@ -713,9 +726,55 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } - case "thread.message-sent": + case "thread.message-sent": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) { + return; + } + // A message can only move latestUserMessageAt forward; the other + // summary fields depend on activities/plans/approvals, which this + // event cannot touch. Update incrementally instead of re-reading the + // whole message/plan/activity/approval history per (streamed) chunk. + const latestUserMessageAt = + event.payload.role === "user" && + (existingRow.value.latestUserMessageAt === null || + event.payload.createdAt > existingRow.value.latestUserMessageAt) + ? event.payload.createdAt + : existingRow.value.latestUserMessageAt; + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + latestUserMessageAt, + // Streaming assistant chunks arrive many times per second and + // change no shell-visible field; leaving updatedAt untouched lets + // the shell stream skip per-chunk upsert fan-out (toShellStreamEvent). + updatedAt: + event.payload.streaming && event.payload.role === "assistant" + ? existingRow.value.updatedAt + : event.occurredAt, + }); + return; + } + + case "thread.activity-appended": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) { + return; + } + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + updatedAt: event.occurredAt, + }); + if (SHELL_SUMMARY_ACTIVITY_KINDS.has(event.payload.activity.kind)) { + yield* refreshThreadShellSummary(event.payload.threadId); + } + return; + } + case "thread.proposed-plan-upserted": - case "thread.activity-appended": case "thread.approval-response-requested": case "thread.user-input-response-requested": { const existingRow = yield* projectionThreadRepository.getById({ diff --git a/apps/server/src/orchestration/Services/OrchestrationEngine.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts index dc1b5fd3003..f0b9a395436 100644 --- a/apps/server/src/orchestration/Services/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Services/OrchestrationEngine.ts @@ -37,6 +37,18 @@ export interface OrchestrationEngineShape { limit?: number, ) => Stream.Stream; + /** + * Replay a single aggregate's persisted events from an exclusive sequence + * cursor. Reads via the aggregate stream index, so per-thread catch-up does + * not scan the global event range after the cursor. + */ + readonly readAggregateEvents: ( + aggregateKind: OrchestrationEvent["aggregateKind"], + aggregateId: string, + fromSequenceExclusive: number, + limit?: number, + ) => Stream.Stream; + /** * Dispatch a validated orchestration command. * diff --git a/apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts b/apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts index 2bac5de920c..2473c2ed577 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts @@ -69,6 +69,51 @@ layer("OrchestrationEventStore", (it) => { }), ); + it.effect("replays only the requested aggregate's events after the cursor", () => + Effect.gen(function* () { + const eventStore = yield* OrchestrationEventStore; + const now = "2026-01-01T00:00:00.000Z"; + + const appendProjectCreated = (suffix: string) => + eventStore.append({ + type: "project.created", + eventId: EventId.make(`evt-aggregate-${suffix}`), + aggregateKind: "project", + aggregateId: ProjectId.make(`project-${suffix}`), + occurredAt: now, + commandId: CommandId.make(`cmd-aggregate-${suffix}`), + causationEventId: null, + correlationId: null, + metadata: {}, + payload: { + projectId: ProjectId.make(`project-${suffix}`), + title: `Project ${suffix}`, + workspaceRoot: `/tmp/project-${suffix}`, + defaultModelSelection: null, + scripts: [], + createdAt: now, + updatedAt: now, + }, + }); + + const first = yield* appendProjectCreated("a"); + yield* appendProjectCreated("b"); + + const replayed = yield* Stream.runCollect( + eventStore.readAggregateFromSequence("project", first.aggregateId, 0), + ).pipe(Effect.map((chunk) => Array.from(chunk))); + assert.deepStrictEqual( + replayed.map((event) => event.aggregateId), + [first.aggregateId], + ); + + const afterCursor = yield* Stream.runCollect( + eventStore.readAggregateFromSequence("project", first.aggregateId, first.sequence), + ).pipe(Effect.map((chunk) => Array.from(chunk))); + assert.deepStrictEqual(afterCursor, []); + }), + ); + it.effect("fails with PersistenceDecodeError when stored json is invalid", () => Effect.gen(function* () { const eventStore = yield* OrchestrationEventStore; diff --git a/apps/server/src/persistence/Layers/OrchestrationEventStore.ts b/apps/server/src/persistence/Layers/OrchestrationEventStore.ts index 18d0e9aa578..43765b8da6c 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEventStore.ts @@ -64,6 +64,12 @@ const ReadFromSequenceRequestSchema = Schema.Struct({ sequenceExclusive: NonNegativeInt, limit: Schema.Number, }); +const ReadAggregateFromSequenceRequestSchema = Schema.Struct({ + aggregateKind: OrchestrationAggregateKind, + aggregateId: Schema.String, + sequenceExclusive: NonNegativeInt, + limit: Schema.Number, +}); const DEFAULT_READ_FROM_SEQUENCE_LIMIT = 1_000; const READ_PAGE_SIZE = 500; @@ -181,6 +187,32 @@ const makeEventStore = Effect.gen(function* () { `, }); + const readAggregateEventRowsFromSequence = SqlSchema.findAll({ + Request: ReadAggregateFromSequenceRequestSchema, + Result: OrchestrationEventPersistedRowSchema, + execute: (request) => + sql` + SELECT + sequence, + event_id AS "eventId", + event_type AS "type", + aggregate_kind AS "aggregateKind", + stream_id AS "aggregateId", + occurred_at AS "occurredAt", + command_id AS "commandId", + causation_event_id AS "causationEventId", + correlation_id AS "correlationId", + payload_json AS "payload", + metadata_json AS "metadata" + FROM orchestration_events + WHERE aggregate_kind = ${request.aggregateKind} + AND stream_id = ${request.aggregateId} + AND sequence > ${request.sequenceExclusive} + ORDER BY sequence ASC + LIMIT ${request.limit} + `, + }); + const append: OrchestrationEventStoreShape["append"] = (event) => appendEventRow({ eventId: event.eventId, @@ -208,10 +240,18 @@ const makeEventStore = Effect.gen(function* () { ), ); - const readFromSequence: OrchestrationEventStoreShape["readFromSequence"] = ( - sequenceExclusive, - limit = DEFAULT_READ_FROM_SEQUENCE_LIMIT, - ) => { + const readPagedEvents = ( + operation: string, + readRows: ( + cursor: number, + pageLimit: number, + ) => Effect.Effect< + ReadonlyArray>, + E + >, + sequenceExclusive: number, + limit: number, + ): Stream.Stream => { const normalizedLimit = Math.max(0, Math.floor(limit)); if (normalizedLimit === 0) { return Stream.empty; @@ -221,22 +261,14 @@ const makeEventStore = Effect.gen(function* () { remaining: number, ): Stream.Stream => Stream.fromEffect( - readEventRowsFromSequence({ - sequenceExclusive: cursor, - limit: Math.min(remaining, READ_PAGE_SIZE), - }).pipe( + readRows(cursor, Math.min(remaining, READ_PAGE_SIZE)).pipe( Effect.mapError( - toPersistenceSqlOrDecodeError( - "OrchestrationEventStore.readFromSequence:query", - "OrchestrationEventStore.readFromSequence:decodeRows", - ), + toPersistenceSqlOrDecodeError(`${operation}:query`, `${operation}:decodeRows`), ), Effect.flatMap((rows) => Effect.forEach(rows, (row) => decodeEvent(row).pipe( - Effect.mapError( - toPersistenceDecodeError("OrchestrationEventStore.readFromSequence:rowToEvent"), - ), + Effect.mapError(toPersistenceDecodeError(`${operation}:rowToEvent`)), ), ), ), @@ -260,9 +292,41 @@ const makeEventStore = Effect.gen(function* () { return readPage(sequenceExclusive, normalizedLimit); }; + const readFromSequence: OrchestrationEventStoreShape["readFromSequence"] = ( + sequenceExclusive, + limit = DEFAULT_READ_FROM_SEQUENCE_LIMIT, + ) => + readPagedEvents( + "OrchestrationEventStore.readFromSequence", + (cursor, pageLimit) => + readEventRowsFromSequence({ sequenceExclusive: cursor, limit: pageLimit }), + sequenceExclusive, + limit, + ); + + const readAggregateFromSequence: OrchestrationEventStoreShape["readAggregateFromSequence"] = ( + aggregateKind, + aggregateId, + sequenceExclusive, + limit = Number.MAX_SAFE_INTEGER, + ) => + readPagedEvents( + "OrchestrationEventStore.readAggregateFromSequence", + (cursor, pageLimit) => + readAggregateEventRowsFromSequence({ + aggregateKind, + aggregateId, + sequenceExclusive: cursor, + limit: pageLimit, + }), + sequenceExclusive, + limit, + ); + return { append, readFromSequence, + readAggregateFromSequence, readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER), } satisfies OrchestrationEventStoreShape; }); diff --git a/apps/server/src/persistence/Services/OrchestrationEventStore.ts b/apps/server/src/persistence/Services/OrchestrationEventStore.ts index 8b465e7713e..f686f30d01d 100644 --- a/apps/server/src/persistence/Services/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/Services/OrchestrationEventStore.ts @@ -46,6 +46,26 @@ export interface OrchestrationEventStoreShape { limit?: number, ) => Stream.Stream; + /** + * Replay a single aggregate's events after the provided sequence. + * + * @param aggregateKind - Aggregate kind of the stream. + * @param aggregateId - Aggregate id of the stream. + * @param sequenceExclusive - Sequence cursor (exclusive). + * @param limit - Maximum number of events to emit. + * @returns Stream containing ordered events for that aggregate only. + * + * The filter runs in SQL against the (aggregate_kind, stream_id, sequence) + * index, so the cost scales with the aggregate's own events rather than the + * global event range after the cursor. + */ + readonly readAggregateFromSequence: ( + aggregateKind: OrchestrationEvent["aggregateKind"], + aggregateId: string, + sequenceExclusive: number, + limit?: number, + ) => Stream.Stream; + /** * Read all events from the beginning of the stream. * diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 40ed694723d..a45aaafd4bf 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -459,6 +459,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { const orchestrationEngine = { readEvents: () => Stream.empty, + readAggregateEvents: () => Stream.empty, dispatch: () => Effect.succeed({ sequence: 1 }), streamDomainEvents: Stream.fromQueue(events), } satisfies OrchestrationEngineShape; @@ -648,6 +649,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { }), Layer.succeed(OrchestrationEngineService, { readEvents: () => Stream.empty, + readAggregateEvents: () => Stream.empty, dispatch: () => Effect.succeed({ sequence: 1 }), streamDomainEvents: Stream.fromQueue(events), } satisfies OrchestrationEngineShape), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 347c2920792..1ed13bcd5bc 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -675,6 +675,7 @@ const buildAppUnderTest = (options?: { Layer.provide( Layer.mock(OrchestrationEngine.OrchestrationEngineService)({ readEvents: () => Stream.empty, + readAggregateEvents: () => Stream.empty, dispatch: () => Effect.succeed({ sequence: 0 }), streamDomainEvents: Stream.empty, ...options?.layers?.orchestrationEngine, diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index 1a331bc717f..e124dff62af 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -163,6 +163,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, + readAggregateEvents: () => Stream.empty, dispatch: (command) => Ref.update(dispatchCalls, (calls) => [...calls, command.type]).pipe( Effect.as({ sequence: 1 }), @@ -206,6 +207,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, + readAggregateEvents: () => Stream.empty, dispatch: (command) => Ref.update(dispatchCalls, (calls) => [...calls, command.type]).pipe( Effect.as({ sequence: 1 }), @@ -255,6 +257,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, + readAggregateEvents: () => Stream.empty, dispatch: (command) => Ref.update(dispatchCalls, (calls) => [...calls, command.type]).pipe( Effect.as({ sequence: 1 }), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index fafdbfa6814..67bb74225c4 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -276,6 +276,11 @@ function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< const PROVIDER_STATUS_DEBOUNCE_MS = 200; +// Shell catch-up replays the global event stream (shell relevance is only known +// after reading), so a stale cursor must not scan an unbounded gap. Past this +// many events it is cheaper to re-send the projection-backed shell snapshot. +const MAX_SHELL_CATCH_UP_EVENTS = 500; + const RPC_REQUIRED_SCOPE = new Map([ [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope], [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope], @@ -662,6 +667,16 @@ const makeWsRpcLayer = ( if (event.aggregateKind !== "thread") { return Effect.succeed(Option.none()); } + // Streaming assistant chunks change no shell-visible field (the + // projection leaves updatedAt untouched for them), so don't read + // and re-broadcast the thread shell row once per token. + if ( + event.type === "thread.message-sent" && + event.payload.streaming && + event.payload.role === "assistant" + ) { + return Effect.succeed(Option.none()); + } return projectionSnapshotQuery .getThreadShellById(ThreadId.make(event.aggregateId)) .pipe( @@ -1079,8 +1094,13 @@ const makeWsRpcLayer = ( // live subscription is attached (into a scope-bound buffer) before // draining the catch-up replay so no event published during the // replay window is lost; overlapping events are deduped by sequence - // on the client. The full range is read (not the store's default - // page limit) since the shell filter runs after reading. + // on the client. + // + // Shell events are filtered out of the global event stream after + // reading, so a stale cursor would otherwise scan every event + // appended since it. The catch-up read is therefore bounded: past + // the bound it is cheaper to re-send the projection-backed shell + // snapshot than to replay (and per-event enrich) the whole gap. if (input.afterSequence !== undefined) { const afterSequence = input.afterSequence; return Stream.unwrap( @@ -1089,22 +1109,41 @@ const makeWsRpcLayer = ( yield* Effect.forkScoped( liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), ); - const catchUpStream = orchestrationEngine - .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) - .pipe( + const catchUpEvents = yield* Stream.runCollect( + orchestrationEngine.readEvents(afterSequence, MAX_SHELL_CATCH_UP_EVENTS + 1), + ).pipe( + Effect.map((events) => Array.from(events)), + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to replay orchestration shell events", + cause, + }), + ), + ); + if (catchUpEvents.length <= MAX_SHELL_CATCH_UP_EVENTS) { + const catchUpStream = Stream.fromIterable(catchUpEvents).pipe( Stream.mapEffect(toShellStreamEvent), Stream.flatMap((event) => Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, ), - Stream.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: "Failed to replay orchestration shell events", - cause, - }), - ), ); - return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer)); + return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer)); + } + + const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe( + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to load orchestration shell snapshot", + cause, + }), + ), + ); + return Stream.concat( + Stream.make({ kind: "snapshot" as const, snapshot }), + Stream.fromQueue(liveBuffer), + ); }), ); } @@ -1179,10 +1218,9 @@ const makeWsRpcLayer = ( // catch-up followed by the buffered/ongoing live events. Overlapping // events are deduped by sequence on the client. // - // Read the full range after the cursor (not the store's default - // page-bounded limit): the range is normally tiny (a fresh HTTP - // snapshot sequence) and the per-thread filter runs after reading, - // so a global cap could otherwise omit this thread's events. + // The catch-up read targets this thread's aggregate stream (via + // the aggregate stream index), so a stale cursor replays only this + // thread's events instead of scanning the global range after it. if (input.afterSequence !== undefined) { const afterSequence = input.afterSequence; return Stream.unwrap( @@ -1192,7 +1230,7 @@ const makeWsRpcLayer = ( liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), ); const catchUpStream = orchestrationEngine - .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) + .readAggregateEvents("thread", input.threadId, afterSequence) .pipe( Stream.filter(isThisThreadDetailEvent), Stream.map((event) => ({ kind: "event" as const, event })), diff --git a/apps/web/src/components/ChatMarkdown.tsx b/apps/web/src/components/ChatMarkdown.tsx index b04e98a8a60..996b383c66e 100644 --- a/apps/web/src/components/ChatMarkdown.tsx +++ b/apps/web/src/components/ChatMarkdown.tsx @@ -140,6 +140,9 @@ const highlightedCodeCache = new LRUCache( MAX_HIGHLIGHT_CACHE_MEMORY_BYTES, ); const highlighterPromiseCache = new Map>(); +// Languages Shiki has no grammar for (e.g. `sh` fences). Without this negative +// cache every render of such a block retries the grammar load and throws again. +const fallbackHighlightLanguages = new Set(); function findTaskListMarkerOffset(markdown: string, listItemStart: number): number | null { const firstLineEnd = markdown.indexOf("\n", listItemStart); @@ -281,12 +284,15 @@ function getHighlighterPromise(language: string): Promise { langs: [language as SupportedLanguages], preferredHighlighter: "shiki-js", }).catch((err) => { - highlighterPromiseCache.delete(language); if (language === "text") { // "text" itself failed — Shiki cannot initialize at all, surface the error + highlighterPromiseCache.delete(language); throw err; } - // Language not supported by Shiki — fall back to "text" + // Language not supported by Shiki — remember that and fall back to "text", + // keeping the resolved promise cached so the grammar load isn't retried on + // every render of a block with that language. + fallbackHighlightLanguages.add(language); return getHighlighterPromise("text"); }); highlighterPromiseCache.set(language, promise); @@ -684,15 +690,17 @@ function UncachedShikiCodeBlock({ }: UncachedShikiCodeBlockProps) { const highlighter = use(getHighlighterPromise(language)); const highlightedHtml = useMemo(() => { + const effectiveLanguage = fallbackHighlightLanguages.has(language) ? "text" : language; try { - return highlighter.codeToHtml(code, { lang: language, theme: themeName }); + return highlighter.codeToHtml(code, { lang: effectiveLanguage, theme: themeName }); } catch (error) { // Log highlighting failures for debugging while falling back to plain text console.warn( `Code highlighting failed for language "${language}", falling back to plain text.`, error instanceof Error ? error.message : error, ); - // If highlighting fails for this language, render as plain text + // Remember the failure so subsequent renders skip the throwing path. + fallbackHighlightLanguages.add(language); return highlighter.codeToHtml(code, { lang: "text", theme: themeName }); } }, [code, highlighter, language, themeName]); diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index 574e33d4dab..04a984bcfd0 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -1,7 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; import { + createSidebarHoverPrewarmController, createThreadJumpHintVisibilityController, - getSidebarThreadIdsToPrewarm, getVisibleSidebarThreadIds, resolveAdjacentThreadId, getFallbackThreadIdAfterDelete, @@ -19,6 +19,7 @@ import { resolveThreadStatusPill, shouldClearThreadSelectionOnMouseDown, sortProjectsForSidebar, + SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS, THREAD_JUMP_HINT_SHOW_DELAY_MS, } from "./Sidebar.logic"; import { @@ -181,17 +182,80 @@ describe("createThreadJumpHintVisibilityController", () => { }); }); -describe("getSidebarThreadIdsToPrewarm", () => { - it("returns only the first visible thread ids up to the prewarm limit", () => { - expect(getSidebarThreadIdsToPrewarm(["t1", "t2", "t3"], 2)).toEqual(["t1", "t2"]); +describe("createSidebarHoverPrewarmController", () => { + beforeEach(() => { + vi.useFakeTimers(); }); - it("returns all visible thread ids when they fit within the limit", () => { - expect(getSidebarThreadIdsToPrewarm(["t1", "t2"], 10)).toEqual(["t1", "t2"]); + afterEach(() => { + vi.useRealTimers(); + }); + + const makeController = () => { + const targets: Array = []; + const controller = createSidebarHoverPrewarmController({ + delayMs: SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS, + onPrewarmTargetChange: (threadKey) => { + targets.push(threadKey); + }, + }); + return { controller, targets }; + }; + + it("prewarms a row only after the pointer rests on it", () => { + const { controller, targets } = makeController(); + + controller.hover("t1"); + vi.advanceTimersByTime(SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS - 1); + expect(targets).toEqual([]); + + vi.advanceTimersByTime(1); + expect(targets).toEqual(["t1"]); }); - it("returns no thread ids when the limit is zero", () => { - expect(getSidebarThreadIdsToPrewarm(["t1", "t2"], 0)).toEqual([]); + it("prewarms at most one thread while sweeping across rows", () => { + const { controller, targets } = makeController(); + + controller.hover("t1"); + vi.advanceTimersByTime(SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS / 2); + controller.hover("t2"); + vi.advanceTimersByTime(SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS / 2); + controller.hover("t3"); + vi.advanceTimersByTime(SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS); + + expect(targets).toEqual(["t3"]); + }); + + it("keeps an established prewarm when re-entering the same row", () => { + const { controller, targets } = makeController(); + + controller.hover("t1"); + vi.advanceTimersByTime(SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS); + controller.hover("t1"); + vi.advanceTimersByTime(SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS); + + expect(targets).toEqual(["t1"]); + }); + + it("releases the prewarm target when the pointer leaves thread rows", () => { + const { controller, targets } = makeController(); + + controller.hover("t1"); + vi.advanceTimersByTime(SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS); + controller.hover(null); + + expect(targets).toEqual(["t1", null]); + }); + + it("cancels a pending prewarm when the pointer leaves early", () => { + const { controller, targets } = makeController(); + + controller.hover("t1"); + vi.advanceTimersByTime(SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS / 2); + controller.hover(null); + vi.advanceTimersByTime(SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS); + + expect(targets).toEqual([]); }); }); diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index 4e7614ed551..abfdadda238 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -13,9 +13,9 @@ import { resolveServerBackedAppStageLabel } from "../branding.logic"; export const THREAD_SELECTION_SAFE_SELECTOR = "[data-thread-item], [data-thread-selection-safe]"; export const THREAD_JUMP_HINT_SHOW_DELAY_MS = 100; -// Visible sidebar rows are prewarmed into the thread-detail cache so opening a -// nearby thread usually reuses an already-hot subscription. -export const SIDEBAR_THREAD_PREWARM_LIMIT = 10; +// A pointer resting on a row signals intent to open it; sweeping past rows must +// not open a detail subscription per row. +export const SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS = 120; export type SidebarNewThreadEnvMode = "local" | "worktree"; type SidebarProject = { id: string; @@ -283,11 +283,68 @@ export function getVisibleSidebarThreadIds( ); } -export function getSidebarThreadIdsToPrewarm( - visibleThreadIds: readonly TThreadId[], - limit = SIDEBAR_THREAD_PREWARM_LIMIT, -): TThreadId[] { - return visibleThreadIds.slice(0, Math.max(0, limit)); +export interface SidebarHoverPrewarmController { + hover: (threadKey: string | null) => void; + dispose: () => void; +} + +// Prewarms at most one thread at a time: the row the pointer has rested on. +// The route mounts the active thread's detail subscription itself, so this is +// the only sidebar-driven detail stream. +export function createSidebarHoverPrewarmController(input: { + delayMs: number; + onPrewarmTargetChange: (threadKey: string | null) => void; + setTimeoutFn?: typeof globalThis.setTimeout; + clearTimeoutFn?: typeof globalThis.clearTimeout; +}): SidebarHoverPrewarmController { + const setTimeoutFn = input.setTimeoutFn ?? globalThis.setTimeout; + const clearTimeoutFn = input.clearTimeoutFn ?? globalThis.clearTimeout; + let target: string | null = null; + let pendingKey: string | null = null; + let timeoutId: NodeJS.Timeout | null = null; + + const clearPending = () => { + if (timeoutId === null) { + return; + } + clearTimeoutFn(timeoutId); + timeoutId = null; + pendingKey = null; + }; + + return { + hover: (threadKey) => { + if (threadKey === null) { + clearPending(); + if (target !== null) { + target = null; + input.onPrewarmTargetChange(null); + } + return; + } + + if (threadKey === target) { + clearPending(); + return; + } + + if (threadKey === pendingKey) { + return; + } + + clearPending(); + pendingKey = threadKey; + timeoutId = setTimeoutFn(() => { + timeoutId = null; + pendingKey = null; + target = threadKey; + input.onPrewarmTargetChange(threadKey); + }, input.delayMs); + }, + dispose: () => { + clearPending(); + }, + }; } export function resolveAdjacentThreadId(input: { diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 21525b56b77..f0eee3769f3 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -184,7 +184,8 @@ import { import { useThreadSelectionStore } from "../threadSelectionStore"; import { useOpenAddProjectCommandPalette } from "../commandPaletteContext"; import { - getSidebarThreadIdsToPrewarm, + createSidebarHoverPrewarmController, + SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS, resolveAdjacentThreadId, isContextMenuPointerDown, isTrailingDoubleClick, @@ -248,6 +249,37 @@ function SidebarThreadDetailPrewarmer({ threadRef }: { readonly threadRef: Scope return null; } +// Self-contained so hover changes re-render only this component, never the +// sidebar tree. Delegated `pointerover` avoids per-row handler props. +function SidebarHoverThreadPrewarmer() { + const [prewarmThreadKey, setPrewarmThreadKey] = useState(null); + + useEffect(() => { + const controller = createSidebarHoverPrewarmController({ + delayMs: SIDEBAR_THREAD_HOVER_PREWARM_DELAY_MS, + onPrewarmTargetChange: setPrewarmThreadKey, + }); + const onPointerOver = (event: globalThis.PointerEvent) => { + const target = event.target instanceof HTMLElement ? event.target : null; + controller.hover( + target?.closest("[data-thread-prewarm-key]")?.getAttribute("data-thread-prewarm-key") ?? + null, + ); + }; + window.addEventListener("pointerover", onPointerOver, { passive: true }); + return () => { + window.removeEventListener("pointerover", onPointerOver); + controller.dispose(); + }; + }, []); + + const prewarmThreadRef = useMemo( + () => (prewarmThreadKey === null ? null : parseScopedThreadKey(prewarmThreadKey)), + [prewarmThreadKey], + ); + return prewarmThreadRef ? : null; +} + function clampSidebarThreadPreviewCount(value: number): SidebarThreadPreviewCount { return Math.min( MAX_SIDEBAR_THREAD_PREVIEW_COUNT, @@ -673,6 +705,7 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr @@ -3491,18 +3524,6 @@ export default function Sidebar() { ? threadJumpLabelByKey : EMPTY_THREAD_JUMP_LABELS; const orderedSidebarThreadKeys = visibleSidebarThreadKeys; - const prewarmedSidebarThreadKeys = useMemo( - () => getSidebarThreadIdsToPrewarm(visibleSidebarThreadKeys), - [visibleSidebarThreadKeys], - ); - const prewarmedSidebarThreadRefs = useMemo( - () => - prewarmedSidebarThreadKeys.flatMap((threadKey) => { - const ref = parseScopedThreadKey(threadKey); - return ref ? [ref] : []; - }), - [prewarmedSidebarThreadKeys], - ); useEffect(() => { updateThreadJumpHintsVisibility(shouldShowThreadJumpHintsNow); @@ -3695,9 +3716,7 @@ export default function Sidebar() { return ( <> - {prewarmedSidebarThreadRefs.map((threadRef) => ( - - ))} + {isOnSettings ? ( diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 1a4dc6b6895..e1e90943347 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -373,7 +373,13 @@ export const MessagesTimeline = memo(function MessagesTimeline({ rowTop < scrollBottom && rowTop + Math.max(1, rowHeight ?? 1) > scrollTop; - strip.dataset.inView = inView ? "true" : "false"; + // Scroll fires per frame; unconditional writes would dirty every strip's + // attributes each frame even though only rows crossing the viewport edge + // change. + const nextInView = inView ? "true" : "false"; + if (strip.dataset.inView !== nextInView) { + strip.dataset.inView = nextInView; + } } }, [listRef, minimapItems, minimapStripMap, onIsAtEndChange]); diff --git a/apps/web/src/composerDraftStore.ts b/apps/web/src/composerDraftStore.ts index fdb8bfe7b18..81974009f7e 100644 --- a/apps/web/src/composerDraftStore.ts +++ b/apps/web/src/composerDraftStore.ts @@ -46,9 +46,9 @@ import { newElementContextId, } from "./lib/elementContext"; import { create } from "zustand"; -import { createJSONStorage, persist } from "zustand/middleware"; +import { persist } from "zustand/middleware"; import { useShallow } from "zustand/react/shallow"; -import { createDebouncedStorage, createMemoryStorage } from "./lib/storage"; +import { createDebouncedJsonStorage, createMemoryStorage } from "./lib/storage"; import { getDefaultServerModel } from "./providerModels"; import { UnifiedSettings } from "@t3tools/contracts/settings"; import { ReviewCommentContextSchema, type ReviewCommentContext } from "./reviewCommentContext"; @@ -66,7 +66,7 @@ export type DraftId = typeof DraftId.Type; const COMPOSER_PERSIST_DEBOUNCE_MS = 300; -const composerDebouncedStorage = createDebouncedStorage( +const composerDebouncedStorage = createDebouncedJsonStorage( typeof localStorage !== "undefined" ? localStorage : createMemoryStorage(), COMPOSER_PERSIST_DEBOUNCE_MS, ); @@ -1815,6 +1815,14 @@ function migratePersistedComposerDraftStoreState( }; } +// Runs on every store update (each prompt keystroke); drafts are immutable +// snapshots, so unchanged drafts reuse their previously partialized form +// instead of re-cloning nested contexts/annotations/comments per update. +const partializedDraftCache = new WeakMap< + ComposerThreadDraftState, + PersistedComposerThreadDraftState | null +>(); + function partializeComposerDraftStoreState( state: ComposerDraftStoreState, ): PersistedComposerDraftStoreState { @@ -1825,78 +1833,20 @@ function partializeComposerDraftStoreState( if (typeof threadKey !== "string" || threadKey.length === 0) { continue; } - const hasModelData = - Object.keys(draft.modelSelectionByProvider).length > 0 || draft.activeProvider !== null; - if ( - draft.prompt.length === 0 && - draft.persistedAttachments.length === 0 && - draft.terminalContexts.length === 0 && - draft.elementContexts.length === 0 && - draft.previewAnnotations.length === 0 && - draft.reviewComments.length === 0 && - !hasModelData && - draft.runtimeMode === null && - draft.interactionMode === null - ) { + const cached = partializedDraftCache.get(draft); + if (cached !== undefined) { + if (cached !== null) { + persistedDraftsByThreadKey[threadKey] = + cached as DeepMutable; + } continue; } - const persistedDraft: DeepMutable = { - prompt: draft.prompt, - attachments: draft.persistedAttachments, - ...(draft.terminalContexts.length > 0 - ? { - terminalContexts: draft.terminalContexts.map((context) => ({ - id: context.id, - threadId: context.threadId, - createdAt: context.createdAt, - terminalId: context.terminalId, - terminalLabel: context.terminalLabel, - lineStart: context.lineStart, - lineEnd: context.lineEnd, - })), - } - : {}), - ...(draft.elementContexts.length > 0 - ? { - elementContexts: draft.elementContexts.map((context) => ({ - id: context.id, - threadId: context.threadId, - pickedAt: context.pickedAt, - pageUrl: context.pageUrl, - pageTitle: context.pageTitle, - tagName: context.tagName, - selector: context.selector, - htmlPreview: context.htmlPreview, - componentName: context.componentName, - source: context.source, - styles: context.styles, - })), - } - : {}), - ...(draft.previewAnnotations.length > 0 - ? { - previewAnnotations: draft.previewAnnotations.map( - (annotation) => ({ ...annotation }) as DeepMutable, - ), - } - : {}), - ...(draft.reviewComments.length > 0 - ? { - reviewComments: draft.reviewComments.map((comment) => ({ ...comment })), - } - : {}), - ...(hasModelData - ? { - modelSelectionByProvider: compactModelSelectionByProvider( - draft.modelSelectionByProvider, - ), - activeProvider: draft.activeProvider, - } - : {}), - ...(draft.runtimeMode ? { runtimeMode: draft.runtimeMode } : {}), - ...(draft.interactionMode ? { interactionMode: draft.interactionMode } : {}), - }; - persistedDraftsByThreadKey[threadKey] = persistedDraft; + const persisted = partializeComposerThreadDraft(draft); + partializedDraftCache.set(draft, persisted); + if (persisted !== null) { + persistedDraftsByThreadKey[threadKey] = + persisted as DeepMutable; + } } return { draftsByThreadKey: persistedDraftsByThreadKey, @@ -1910,6 +1860,80 @@ function partializeComposerDraftStoreState( }; } +function partializeComposerThreadDraft( + draft: ComposerThreadDraftState, +): PersistedComposerThreadDraftState | null { + const hasModelData = + Object.keys(draft.modelSelectionByProvider).length > 0 || draft.activeProvider !== null; + if ( + draft.prompt.length === 0 && + draft.persistedAttachments.length === 0 && + draft.terminalContexts.length === 0 && + draft.elementContexts.length === 0 && + draft.previewAnnotations.length === 0 && + draft.reviewComments.length === 0 && + !hasModelData && + draft.runtimeMode === null && + draft.interactionMode === null + ) { + return null; + } + return { + prompt: draft.prompt, + attachments: draft.persistedAttachments, + ...(draft.terminalContexts.length > 0 + ? { + terminalContexts: draft.terminalContexts.map((context) => ({ + id: context.id, + threadId: context.threadId, + createdAt: context.createdAt, + terminalId: context.terminalId, + terminalLabel: context.terminalLabel, + lineStart: context.lineStart, + lineEnd: context.lineEnd, + })), + } + : {}), + ...(draft.elementContexts.length > 0 + ? { + elementContexts: draft.elementContexts.map((context) => ({ + id: context.id, + threadId: context.threadId, + pickedAt: context.pickedAt, + pageUrl: context.pageUrl, + pageTitle: context.pageTitle, + tagName: context.tagName, + selector: context.selector, + htmlPreview: context.htmlPreview, + componentName: context.componentName, + source: context.source, + styles: context.styles, + })), + } + : {}), + ...(draft.previewAnnotations.length > 0 + ? { + previewAnnotations: draft.previewAnnotations.map( + (annotation) => ({ ...annotation }) as DeepMutable, + ), + } + : {}), + ...(draft.reviewComments.length > 0 + ? { + reviewComments: draft.reviewComments.map((comment) => ({ ...comment })), + } + : {}), + ...(hasModelData + ? { + modelSelectionByProvider: compactModelSelectionByProvider(draft.modelSelectionByProvider), + activeProvider: draft.activeProvider, + } + : {}), + ...(draft.runtimeMode ? { runtimeMode: draft.runtimeMode } : {}), + ...(draft.interactionMode ? { interactionMode: draft.interactionMode } : {}), + }; +} + function normalizeCurrentPersistedComposerDraftStoreState( persistedState: unknown, ): PersistedComposerDraftStoreState { @@ -3341,7 +3365,7 @@ const composerDraftStore = create()( { name: COMPOSER_DRAFT_STORAGE_KEY, version: COMPOSER_DRAFT_STORAGE_VERSION, - storage: createJSONStorage(() => composerDebouncedStorage), + storage: composerDebouncedStorage, migrate: migratePersistedComposerDraftStoreState, partialize: partializeComposerDraftStoreState, merge: (persistedState, currentState) => { diff --git a/apps/web/src/connection/useDesktopLocalBootstraps.ts b/apps/web/src/connection/useDesktopLocalBootstraps.ts index 71642aecba4..3db1708d0f1 100644 --- a/apps/web/src/connection/useDesktopLocalBootstraps.ts +++ b/apps/web/src/connection/useDesktopLocalBootstraps.ts @@ -1,10 +1,71 @@ import type { DesktopEnvironmentBootstrap } from "@t3tools/contracts"; -import { useEffect, useState } from "react"; +import { useSyncExternalStore } from "react"; import { readDesktopSecondaryBootstraps } from "./desktopLocal"; const DESKTOP_LOCAL_BOOTSTRAP_POLL_MS = 2_000; +function bootstrapsEqual( + left: ReadonlyArray, + right: ReadonlyArray, +): boolean { + if (left.length !== right.length) { + return false; + } + return left.every((entry, index) => { + const other = right[index]; + return ( + other !== undefined && + entry.id === other.id && + entry.label === other.label && + entry.runningDistro === other.runningDistro && + entry.httpBaseUrl === other.httpBaseUrl && + entry.wsBaseUrl === other.wsBaseUrl && + entry.bootstrapToken === other.bootstrapToken + ); + }); +} + +// One shared poller for all consumers (sidebar, command palette, ...): a single +// interval runs only while someone is subscribed, and listeners are notified +// only when the topology actually changed — each poll returns a fresh array, +// so publishing it unconditionally would re-render every consumer per tick. +const listeners = new Set<() => void>(); +let currentSnapshot: ReadonlyArray | null = null; +let pollInterval: ReturnType | null = null; + +function poll(): void { + const next = readDesktopSecondaryBootstraps(); + if (currentSnapshot !== null && bootstrapsEqual(currentSnapshot, next)) { + return; + } + currentSnapshot = next; + for (const listener of listeners) { + listener(); + } +} + +function subscribe(listener: () => void): () => void { + listeners.add(listener); + if (pollInterval === null) { + poll(); + pollInterval = setInterval(poll, DESKTOP_LOCAL_BOOTSTRAP_POLL_MS); + } + return () => { + listeners.delete(listener); + if (listeners.size === 0 && pollInterval !== null) { + clearInterval(pollInterval); + pollInterval = null; + } + }; +} + +function getSnapshot(): ReadonlyArray { + // First read happens during render, before subscribe() has polled. + currentSnapshot ??= readDesktopSecondaryBootstraps(); + return currentSnapshot; +} + /** * Reactively track the desktop's secondary local backends (e.g. a parallel WSL * backend). The bridge exposes no change event, so we re-read on an interval; @@ -13,16 +74,5 @@ const DESKTOP_LOCAL_BOOTSTRAP_POLL_MS = 2_000; * renderer consumer reads the same topology. */ export function useDesktopLocalBootstraps(): ReadonlyArray { - const [bootstraps, setBootstraps] = useState>( - readDesktopSecondaryBootstraps, - ); - - useEffect(() => { - const read = () => setBootstraps(readDesktopSecondaryBootstraps()); - read(); - const interval = setInterval(read, DESKTOP_LOCAL_BOOTSTRAP_POLL_MS); - return () => clearInterval(interval); - }, []); - - return bootstraps; + return useSyncExternalStore(subscribe, getSnapshot, getSnapshot); } diff --git a/apps/web/src/lib/storage.ts b/apps/web/src/lib/storage.ts index a37c67064aa..618b98b1227 100644 --- a/apps/web/src/lib/storage.ts +++ b/apps/web/src/lib/storage.ts @@ -1,4 +1,5 @@ import { Debouncer } from "@tanstack/react-pacer"; +import type { PersistStorage, StorageValue } from "zustand/middleware"; export interface StateStorage { getItem: (name: string) => string | null | Promise; @@ -65,3 +66,45 @@ export function createDebouncedStorage( }, }; } + +export interface DebouncedPersistStorage extends PersistStorage { + flush: () => void; +} + +/** + * A zustand `PersistStorage` that debounces the JSON serialization as well as + * the storage write. `createJSONStorage` + a debounced `StateStorage` only + * delays the write — the full state is still stringified on every store + * update; here it happens once per debounce window instead. + */ +export function createDebouncedJsonStorage( + baseStorage: Partial | null | undefined, + debounceMs: number = 300, +): DebouncedPersistStorage { + const resolvedStorage = resolveStorage(baseStorage); + const debouncedSetItem = new Debouncer( + (name: string, value: StorageValue) => { + resolvedStorage.setItem(name, JSON.stringify(value)); + }, + { wait: debounceMs }, + ); + + return { + getItem: (name) => { + const raw = resolvedStorage.getItem(name); + const parse = (value: string | null): StorageValue | null => + value === null ? null : (JSON.parse(value) as StorageValue); + return raw instanceof Promise ? raw.then(parse) : parse(raw); + }, + setItem: (name, value) => { + debouncedSetItem.maybeExecute(name, value); + }, + removeItem: (name) => { + debouncedSetItem.cancel(); + resolvedStorage.removeItem(name); + }, + flush: () => { + debouncedSetItem.flush(); + }, + }; +} diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 5d5051f748e..bd8fd2f5e47 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -356,7 +356,7 @@ export function derivePendingApprovals( activities: ReadonlyArray, ): PendingApproval[] { const openByRequestId = new Map(); - const ordered = [...activities].toSorted(compareActivitiesByOrder); + const ordered = orderActivities(activities); for (const activity of ordered) { const payload = @@ -462,7 +462,7 @@ export function derivePendingUserInputs( activities: ReadonlyArray, ): PendingUserInput[] { const openByRequestId = new Map(); - const ordered = [...activities].toSorted(compareActivitiesByOrder); + const ordered = orderActivities(activities); for (const activity of ordered) { const payload = @@ -511,7 +511,7 @@ export function deriveActivePlanState( activities: ReadonlyArray, latestTurnId: TurnId | undefined, ): ActivePlanState | null { - const ordered = [...activities].toSorted(compareActivitiesByOrder); + const ordered = orderActivities(activities); const allPlanActivities = ordered.filter((activity) => activity.kind === "turn.plan.updated"); // Prefer plan from the current turn; fall back to the most recent plan from any turn // so that TodoWrite tasks persist across follow-up messages. @@ -627,7 +627,7 @@ export function hasActionableProposedPlan( export function deriveWorkLogEntries( activities: ReadonlyArray, ): WorkLogEntry[] { - const ordered = [...activities].toSorted(compareActivitiesByOrder); + const ordered = orderActivities(activities); const entries: DerivedWorkLogEntry[] = []; for (const activity of ordered) { if (activity.kind === "tool.started") continue; @@ -1296,6 +1296,35 @@ function extractChangedFiles(payload: Record | null): string[] return changedFiles; } +// Several independent derivations (work log, pending approvals, pending user +// inputs, active plan) order the same activities array per render. The reducer +// appends in order, so the input is almost always already sorted — detect that +// in O(n) and share the result per array identity instead of copying and +// re-sorting the whole history in each derivation. +const orderedActivitiesCache = new WeakMap< + ReadonlyArray, + ReadonlyArray +>(); + +function orderActivities( + activities: ReadonlyArray, +): ReadonlyArray { + const cached = orderedActivitiesCache.get(activities); + if (cached) { + return cached; + } + let isSorted = true; + for (let index = 1; index < activities.length; index += 1) { + if (compareActivitiesByOrder(activities[index - 1]!, activities[index]!) > 0) { + isSorted = false; + break; + } + } + const ordered = isSorted ? activities : [...activities].toSorted(compareActivitiesByOrder); + orderedActivitiesCache.set(activities, ordered); + return ordered; +} + function compareActivitiesByOrder( left: OrchestrationThreadActivity, right: OrchestrationThreadActivity, @@ -1337,6 +1366,42 @@ function compareActivityLifecycleRank(kind: string): number { return 1; } +function isSortedByCreatedAt(rows: ReadonlyArray): boolean { + for (let index = 1; index < rows.length; index += 1) { + if (rows[index - 1]!.createdAt.localeCompare(rows[index]!.createdAt) > 0) { + return false; + } + } + return true; +} + +// Stable k-way merge of per-kind rows that are each already in createdAt order +// (the reducers maintain that invariant). Equal timestamps keep the same order +// a stable sort of the concatenation would produce. +function mergeTimelineRows(sources: ReadonlyArray>): TimelineEntry[] { + const merged: TimelineEntry[] = []; + const cursors = sources.map(() => 0); + const total = sources.reduce((sum, rows) => sum + rows.length, 0); + while (merged.length < total) { + let nextSource = -1; + for (let index = 0; index < sources.length; index += 1) { + const candidate = sources[index]![cursors[index]!]; + if (candidate === undefined) { + continue; + } + if ( + nextSource === -1 || + candidate.createdAt.localeCompare(sources[nextSource]![cursors[nextSource]!]!.createdAt) < 0 + ) { + nextSource = index; + } + } + merged.push(sources[nextSource]![cursors[nextSource]!]!); + cursors[nextSource] = cursors[nextSource]! + 1; + } + return merged; +} + export function deriveTimelineEntries( messages: ReadonlyArray, proposedPlans: ReadonlyArray, @@ -1360,9 +1425,10 @@ export function deriveTimelineEntries( createdAt: entry.createdAt, entry, })); - return [...messageRows, ...proposedPlanRows, ...workRows].toSorted((a, b) => - a.createdAt.localeCompare(b.createdAt), - ); + const sources = [messageRows, proposedPlanRows, workRows]; + return sources.every(isSortedByCreatedAt) + ? mergeTimelineRows(sources) + : sources.flat().toSorted((a, b) => a.createdAt.localeCompare(b.createdAt)); } export function inferCheckpointTurnCountByTurnId( diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index 670540fee70..cf51d470c90 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -457,12 +457,21 @@ export function applyThreadDetailEvent( // ── Activities ────────────────────────────────────────────────── case "thread.activity-appended": { - const activities = pipe( - thread.activities, - Arr.filter((activity) => activity.id !== event.payload.activity.id), - Arr.append(event.payload.activity), - Arr.sort(activityOrder), - ); + const activity = event.payload.activity; + // Live activities arrive in order and are new: keep the sorted invariant + // with a single append instead of filter+append+sort over the (possibly + // very long) history on every event. + const lastActivity = thread.activities.at(-1); + const activities = + (lastActivity === undefined || activityOrder(lastActivity, activity) <= 0) && + !thread.activities.some((entry) => entry.id === activity.id) + ? Arr.append(thread.activities, activity) + : pipe( + thread.activities, + Arr.filter((entry) => entry.id !== activity.id), + Arr.append(activity), + Arr.sort(activityOrder), + ); return { kind: "updated", diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index b4548995d1f..502aebe476f 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -5,6 +5,7 @@ import { ProjectId, ProviderInstanceId, ThreadId, + TurnId, type OrchestrationThread, type OrchestrationThreadDetailSnapshot, type OrchestrationThreadStreamItem, @@ -304,6 +305,67 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("skips snapshot persistence while the session is running a turn", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_THREAD }); + yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD)); + const sessionSet = ( + status: "running" | "idle", + sequence: number, + ): OrchestrationThreadStreamItem => ({ + kind: "event", + event: { + eventId: EventId.make(`event-session-${sequence}`), + sequence, + occurredAt: "2026-04-01T01:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + aggregateKind: "thread", + aggregateId: THREAD_ID, + type: "thread.session-set", + payload: { + threadId: THREAD_ID, + session: { + threadId: THREAD_ID, + status, + providerName: null, + runtimeMode: "full-access", + activeTurnId: status === "running" ? TurnId.make("turn-1") : null, + lastError: null, + updatedAt: "2026-04-01T01:00:00.000Z", + }, + }, + }, + }); + + yield* Queue.offer(harness.inputs, sessionSet("running", 2)); + yield* Queue.offer(harness.inputs, titleUpdated("Streaming title", 3)); + yield* awaitThreadState( + harness.observed, + (value) => Option.isSome(value.data) && value.data.value.title === "Streaming title", + ); + yield* TestClock.adjust("500 millis"); + yield* Effect.yieldNow; + const savedWhileRunning = yield* Ref.get(harness.savedThreads); + expect(savedWhileRunning.some((saved) => saved.thread.title === "Streaming title")).toBe( + false, + ); + + yield* Queue.offer(harness.inputs, sessionSet("idle", 4)); + yield* awaitThreadState( + harness.observed, + (value) => Option.isSome(value.data) && value.data.value.session?.status === "idle", + ); + yield* TestClock.adjust("500 millis"); + yield* Effect.yieldNow; + const savedAfterSettle = yield* Ref.get(harness.savedThreads); + expect(savedAfterSettle.at(-1)?.thread.title).toBe("Streaming title"); + expect(savedAfterSettle.at(-1)?.snapshotSequence).toBe(4); + }), + ); + it.effect("seeds the thread from the HTTP snapshot and resumes live events", () => Effect.gen(function* () { const httpThread: OrchestrationThread = { ...BASE_THREAD, title: "HTTP title" }; diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index fd5b425fa2a..9b41492a229 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -129,7 +129,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make error: Option.none(), }); // Persist the thread together with the sequence it reflects so the next warm - // cache can resume from exactly here. + // cache can resume from exactly here. While a turn is actively running, + // events arrive many times per second and each snapshot encode covers the + // whole thread history — skip those; the session leaving "running" (and the + // scope finalizer) persists the settled state, and a stale cache only means + // resuming via a slightly longer `afterSequence` replay. + if (thread.session?.status === "running") { + return; + } const snapshotSequence = yield* SubscriptionRef.get(lastSequence); yield* Queue.offer(persistence, { snapshotSequence, thread }); }); From 05e623fee9e0173c0e579136e6bc46f31516caca Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 11 Jul 2026 17:01:44 -0700 Subject: [PATCH 2/4] refactor: drop createDebouncedStorage now superseded by the JSON variant The composer store was its only production consumer; port its tests to createDebouncedJsonStorage. Co-Authored-By: Claude Fable 5 --- apps/web/src/composerDraftStore.test.ts | 67 ++++++++++++++----------- apps/web/src/lib/storage.ts | 31 ------------ 2 files changed, 39 insertions(+), 59 deletions(-) diff --git a/apps/web/src/composerDraftStore.test.ts b/apps/web/src/composerDraftStore.test.ts index bc1b7107306..d5ae41ff50c 100644 --- a/apps/web/src/composerDraftStore.test.ts +++ b/apps/web/src/composerDraftStore.test.ts @@ -75,7 +75,7 @@ import { insertInlineTerminalContextPlaceholder, type TerminalContextDraft, } from "./lib/terminalContext"; -import { createDebouncedStorage } from "./lib/storage"; +import { createDebouncedJsonStorage } from "./lib/storage"; function makeImage(input: { id: string; @@ -1648,7 +1648,7 @@ describe("composerDraftStore runtime and interaction settings", () => { }); // --------------------------------------------------------------------------- -// createDebouncedStorage +// createDebouncedJsonStorage // --------------------------------------------------------------------------- function createMockStorage() { @@ -1664,7 +1664,11 @@ function createMockStorage() { }; } -describe("createDebouncedStorage", () => { +describe("createDebouncedJsonStorage", () => { + const v1 = { state: { value: "v1" }, version: 1 }; + const v2 = { state: { value: "v2" }, version: 1 }; + const v3 = { state: { value: "v3" }, version: 1 }; + beforeEach(() => { vi.useFakeTimers(); }); @@ -1673,47 +1677,54 @@ describe("createDebouncedStorage", () => { vi.useRealTimers(); }); - it("delegates getItem immediately", () => { + it("parses getItem results immediately", () => { const base = createMockStorage(); - base.getItem.mockReturnValueOnce("value"); - const storage = createDebouncedStorage(base); + base.getItem.mockReturnValueOnce(JSON.stringify(v1)); + const storage = createDebouncedJsonStorage(base); - expect(storage.getItem("key")).toBe("value"); + expect(storage.getItem("key")).toEqual(v1); expect(base.getItem).toHaveBeenCalledWith("key"); }); - it("does not write to base storage until the debounce fires", () => { + it("returns null from getItem when nothing is stored", () => { + const base = createMockStorage(); + const storage = createDebouncedJsonStorage(base); + + expect(storage.getItem("key")).toBeNull(); + }); + + it("does not serialize or write until the debounce fires", () => { const base = createMockStorage(); - const storage = createDebouncedStorage(base); + const storage = createDebouncedJsonStorage(base); - storage.setItem("key", "v1"); + storage.setItem("key", v1); expect(base.setItem).not.toHaveBeenCalled(); vi.advanceTimersByTime(299); expect(base.setItem).not.toHaveBeenCalled(); vi.advanceTimersByTime(1); - expect(base.setItem).toHaveBeenCalledWith("key", "v1"); + expect(base.setItem).toHaveBeenCalledWith("key", JSON.stringify(v1)); }); it("only writes the last value when setItem is called rapidly", () => { const base = createMockStorage(); - const storage = createDebouncedStorage(base); + const storage = createDebouncedJsonStorage(base); - storage.setItem("key", "v1"); - storage.setItem("key", "v2"); - storage.setItem("key", "v3"); + storage.setItem("key", v1); + storage.setItem("key", v2); + storage.setItem("key", v3); vi.advanceTimersByTime(300); expect(base.setItem).toHaveBeenCalledTimes(1); - expect(base.setItem).toHaveBeenCalledWith("key", "v3"); + expect(base.setItem).toHaveBeenCalledWith("key", JSON.stringify(v3)); }); it("removeItem cancels a pending setItem write", () => { const base = createMockStorage(); - const storage = createDebouncedStorage(base); + const storage = createDebouncedJsonStorage(base); - storage.setItem("key", "v1"); + storage.setItem("key", v1); storage.removeItem("key"); vi.advanceTimersByTime(300); @@ -1723,13 +1734,13 @@ describe("createDebouncedStorage", () => { it("flush writes the pending value immediately", () => { const base = createMockStorage(); - const storage = createDebouncedStorage(base); + const storage = createDebouncedJsonStorage(base); - storage.setItem("key", "v1"); + storage.setItem("key", v1); expect(base.setItem).not.toHaveBeenCalled(); storage.flush(); - expect(base.setItem).toHaveBeenCalledWith("key", "v1"); + expect(base.setItem).toHaveBeenCalledWith("key", JSON.stringify(v1)); // Timer should be cancelled; no duplicate write. vi.advanceTimersByTime(300); @@ -1738,7 +1749,7 @@ describe("createDebouncedStorage", () => { it("flush is a no-op when nothing is pending", () => { const base = createMockStorage(); - const storage = createDebouncedStorage(base); + const storage = createDebouncedJsonStorage(base); storage.flush(); expect(base.setItem).not.toHaveBeenCalled(); @@ -1746,9 +1757,9 @@ describe("createDebouncedStorage", () => { it("flush after removeItem is a no-op", () => { const base = createMockStorage(); - const storage = createDebouncedStorage(base); + const storage = createDebouncedJsonStorage(base); - storage.setItem("key", "v1"); + storage.setItem("key", v1); storage.removeItem("key"); storage.flush(); @@ -1757,14 +1768,14 @@ describe("createDebouncedStorage", () => { it("setItem works normally after removeItem cancels a pending write", () => { const base = createMockStorage(); - const storage = createDebouncedStorage(base); + const storage = createDebouncedJsonStorage(base); - storage.setItem("key", "v1"); + storage.setItem("key", v1); storage.removeItem("key"); - storage.setItem("key", "v2"); + storage.setItem("key", v2); vi.advanceTimersByTime(300); expect(base.setItem).toHaveBeenCalledTimes(1); - expect(base.setItem).toHaveBeenCalledWith("key", "v2"); + expect(base.setItem).toHaveBeenCalledWith("key", JSON.stringify(v2)); }); }); diff --git a/apps/web/src/lib/storage.ts b/apps/web/src/lib/storage.ts index 618b98b1227..ef3e95be87a 100644 --- a/apps/web/src/lib/storage.ts +++ b/apps/web/src/lib/storage.ts @@ -7,10 +7,6 @@ export interface StateStorage { removeItem: (name: string) => R; } -export interface DebouncedStorage extends StateStorage { - flush: () => void; -} - export function createMemoryStorage(): StateStorage { const store = new Map(); return { @@ -40,33 +36,6 @@ export function resolveStorage(storage: Partial | null | undefined return isStateStorage(storage) ? storage : createMemoryStorage(); } -export function createDebouncedStorage( - baseStorage: Partial | null | undefined, - debounceMs: number = 300, -): DebouncedStorage { - const resolvedStorage = resolveStorage(baseStorage); - const debouncedSetItem = new Debouncer( - (name: string, value: string) => { - resolvedStorage.setItem(name, value); - }, - { wait: debounceMs }, - ); - - return { - getItem: (name) => resolvedStorage.getItem(name), - setItem: (name, value) => { - debouncedSetItem.maybeExecute(name, value); - }, - removeItem: (name) => { - debouncedSetItem.cancel(); - resolvedStorage.removeItem(name); - }, - flush: () => { - debouncedSetItem.flush(); - }, - }; -} - export interface DebouncedPersistStorage extends PersistStorage { flush: () => void; } From 6b2f3f8ae88e82701b73375f1b43abe5b5dc0472 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 11 Jul 2026 17:28:19 -0700 Subject: [PATCH 3/4] fix: address review bot findings on hover prewarm, Shiki cache, activity fan-out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Sidebar hover prewarm: match Element instead of HTMLElement so resting on a row's SVG icons (status pill, PR badge) still resolves to the row's prewarm key instead of cancelling the hover. - Shiki negative cache: bound it (100 entries) since fence labels are arbitrary text; evict failed loads from the promise cache so transient failures stay retryable; stop adding render-time codeToHtml throws to the negative cache so one pathological block can't disable highlighting for a supported language. - Activity projection: skip the thread-row upsert entirely for non-summary activity kinds (not just the history rebuild) and skip the corresponding shell re-broadcast in ws.ts — high-volume command/output/progress streams no longer churn updatedAt or fan out per-event shell upserts. Co-Authored-By: Claude Fable 5 --- .../Layers/ProjectionPipeline.ts | 17 ++++++++---- apps/server/src/ws.ts | 10 +++++++ apps/web/src/components/ChatMarkdown.tsx | 27 +++++++++++++------ apps/web/src/components/Sidebar.tsx | 4 ++- 4 files changed, 44 insertions(+), 14 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index aeee73a2557..5e94e5d470b 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -179,8 +179,9 @@ function derivePendingUserInputCountFromActivities( // Activity kinds that can change the shell summary's pending-approval or // pending-user-input counters. All other activity kinds (command output, // file edits, streaming progress, ...) leave the summary untouched, so they -// must not trigger the history-wide summary rebuild. -const SHELL_SUMMARY_ACTIVITY_KINDS: ReadonlySet = new Set([ +// must not trigger the history-wide summary rebuild — and the shell stream +// (ws.ts toShellStreamEvent) skips re-broadcasting the thread row for them. +export const SHELL_SUMMARY_ACTIVITY_KINDS: ReadonlySet = new Set([ "approval.requested", "approval.resolved", "provider.approval.respond.failed", @@ -758,6 +759,14 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti } case "thread.activity-appended": { + // Non-summary activities (command output, file edits, progress, ...) + // change no shell-visible field: the sidebar timestamp prefers + // latestUserMessageAt and status pills derive from session/approval + // state. Skip the row churn entirely so high-volume activity streams + // don't bump updatedAt (and fan out shell upserts) per event. + if (!SHELL_SUMMARY_ACTIVITY_KINDS.has(event.payload.activity.kind)) { + return; + } const existingRow = yield* projectionThreadRepository.getById({ threadId: event.payload.threadId, }); @@ -768,9 +777,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti ...existingRow.value, updatedAt: event.occurredAt, }); - if (SHELL_SUMMARY_ACTIVITY_KINDS.has(event.payload.activity.kind)) { - yield* refreshThreadShellSummary(event.payload.threadId); - } + yield* refreshThreadShellSummary(event.payload.threadId); return; } diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 67bb74225c4..b9385154ecb 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -68,6 +68,7 @@ import * as ServerConfig from "./config.ts"; import * as Keybindings from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts"; +import { SHELL_SUMMARY_ACTIVITY_KINDS } from "./orchestration/Layers/ProjectionPipeline.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import { @@ -677,6 +678,15 @@ const makeWsRpcLayer = ( ) { return Effect.succeed(Option.none()); } + // Same for high-volume activity kinds (command output, file edits, + // progress): the projection skips the thread row for them, so a + // shell broadcast would be a no-op re-send of unchanged data. + if ( + event.type === "thread.activity-appended" && + !SHELL_SUMMARY_ACTIVITY_KINDS.has(event.payload.activity.kind) + ) { + return Effect.succeed(Option.none()); + } return projectionSnapshotQuery .getThreadShellById(ThreadId.make(event.aggregateId)) .pipe( diff --git a/apps/web/src/components/ChatMarkdown.tsx b/apps/web/src/components/ChatMarkdown.tsx index 996b383c66e..0f6a7e1f1eb 100644 --- a/apps/web/src/components/ChatMarkdown.tsx +++ b/apps/web/src/components/ChatMarkdown.tsx @@ -142,7 +142,10 @@ const highlightedCodeCache = new LRUCache( const highlighterPromiseCache = new Map>(); // Languages Shiki has no grammar for (e.g. `sh` fences). Without this negative // cache every render of such a block retries the grammar load and throws again. +// Fence labels are arbitrary text, so cap both caches: past the cap, unknown +// languages just retry (the pre-cache behavior) instead of growing memory. const fallbackHighlightLanguages = new Set(); +const MAX_FALLBACK_HIGHLIGHT_LANGUAGES = 100; function findTaskListMarkerOffset(markdown: string, listItemStart: number): number | null { const firstLineEnd = markdown.indexOf("\n", listItemStart); @@ -276,6 +279,9 @@ function estimateHighlightedSize(html: string, code: string): number { } function getHighlighterPromise(language: string): Promise { + if (fallbackHighlightLanguages.has(language)) { + return getHighlighterPromise("text"); + } const cached = highlighterPromiseCache.get(language); if (cached) return cached; @@ -284,15 +290,19 @@ function getHighlighterPromise(language: string): Promise { langs: [language as SupportedLanguages], preferredHighlighter: "shiki-js", }).catch((err) => { + // Failed loads never stay in the promise cache: unsupported languages are + // remembered in the (bounded) negative cache instead, and transient + // failures for real languages stay retryable. + highlighterPromiseCache.delete(language); if (language === "text") { // "text" itself failed — Shiki cannot initialize at all, surface the error - highlighterPromiseCache.delete(language); throw err; } - // Language not supported by Shiki — remember that and fall back to "text", - // keeping the resolved promise cached so the grammar load isn't retried on - // every render of a block with that language. - fallbackHighlightLanguages.add(language); + // Language not supported by Shiki — remember that and fall back to "text" + // so blocks with that language stop retrying the grammar load per render. + if (fallbackHighlightLanguages.size < MAX_FALLBACK_HIGHLIGHT_LANGUAGES) { + fallbackHighlightLanguages.add(language); + } return getHighlighterPromise("text"); }); highlighterPromiseCache.set(language, promise); @@ -694,13 +704,14 @@ function UncachedShikiCodeBlock({ try { return highlighter.codeToHtml(code, { lang: effectiveLanguage, theme: themeName }); } catch (error) { - // Log highlighting failures for debugging while falling back to plain text + // Log highlighting failures for debugging while falling back to plain + // text. Render-time throws can be content-specific, so they don't join + // the negative cache — one pathological block must not disable + // highlighting for the whole language. Retries are bounded by the memo. console.warn( `Code highlighting failed for language "${language}", falling back to plain text.`, error instanceof Error ? error.message : error, ); - // Remember the failure so subsequent renders skip the throwing path. - fallbackHighlightLanguages.add(language); return highlighter.codeToHtml(code, { lang: "text", theme: themeName }); } }, [code, highlighter, language, themeName]); diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index f0eee3769f3..1c4de4480c6 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -260,7 +260,9 @@ function SidebarHoverThreadPrewarmer() { onPrewarmTargetChange: setPrewarmThreadKey, }); const onPointerOver = (event: globalThis.PointerEvent) => { - const target = event.target instanceof HTMLElement ? event.target : null; + // Element, not HTMLElement: row icons are SVG and must still resolve to + // their row's key instead of cancelling the hover. + const target = event.target instanceof Element ? event.target : null; controller.hover( target?.closest("[data-thread-prewarm-key]")?.getAttribute("data-thread-prewarm-key") ?? null, From e5d5249680d718c7b22d093adb77f2950c29064d Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Mon, 13 Jul 2026 16:21:24 -0700 Subject: [PATCH 4/4] refactor: trim redundant optimizations from the perf PR (#3928) Co-authored-by: Claude Fable 5 --- apps/web/src/components/Sidebar.logic.ts | 10 +-- apps/web/src/components/Sidebar.tsx | 3 +- apps/web/src/composerDraftStore.ts | 17 ----- .../connection/useDesktopLocalBootstraps.ts | 66 ++++++------------- apps/web/src/session-logic.ts | 43 +----------- 5 files changed, 28 insertions(+), 111 deletions(-) diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index abfdadda238..e08bf42c142 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -294,20 +294,16 @@ export interface SidebarHoverPrewarmController { export function createSidebarHoverPrewarmController(input: { delayMs: number; onPrewarmTargetChange: (threadKey: string | null) => void; - setTimeoutFn?: typeof globalThis.setTimeout; - clearTimeoutFn?: typeof globalThis.clearTimeout; }): SidebarHoverPrewarmController { - const setTimeoutFn = input.setTimeoutFn ?? globalThis.setTimeout; - const clearTimeoutFn = input.clearTimeoutFn ?? globalThis.clearTimeout; let target: string | null = null; let pendingKey: string | null = null; - let timeoutId: NodeJS.Timeout | null = null; + let timeoutId: ReturnType | null = null; const clearPending = () => { if (timeoutId === null) { return; } - clearTimeoutFn(timeoutId); + clearTimeout(timeoutId); timeoutId = null; pendingKey = null; }; @@ -334,7 +330,7 @@ export function createSidebarHoverPrewarmController(input: { clearPending(); pendingKey = threadKey; - timeoutId = setTimeoutFn(() => { + timeoutId = setTimeout(() => { timeoutId = null; pendingKey = null; target = threadKey; diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 1c4de4480c6..18497970e9c 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -250,7 +250,8 @@ function SidebarThreadDetailPrewarmer({ threadRef }: { readonly threadRef: Scope } // Self-contained so hover changes re-render only this component, never the -// sidebar tree. Delegated `pointerover` avoids per-row handler props. +// sidebar tree. Delegated `pointerover` avoids threading a hover callback +// through the memoized row tree. function SidebarHoverThreadPrewarmer() { const [prewarmThreadKey, setPrewarmThreadKey] = useState(null); diff --git a/apps/web/src/composerDraftStore.ts b/apps/web/src/composerDraftStore.ts index 81974009f7e..db156beeb48 100644 --- a/apps/web/src/composerDraftStore.ts +++ b/apps/web/src/composerDraftStore.ts @@ -1815,14 +1815,6 @@ function migratePersistedComposerDraftStoreState( }; } -// Runs on every store update (each prompt keystroke); drafts are immutable -// snapshots, so unchanged drafts reuse their previously partialized form -// instead of re-cloning nested contexts/annotations/comments per update. -const partializedDraftCache = new WeakMap< - ComposerThreadDraftState, - PersistedComposerThreadDraftState | null ->(); - function partializeComposerDraftStoreState( state: ComposerDraftStoreState, ): PersistedComposerDraftStoreState { @@ -1833,16 +1825,7 @@ function partializeComposerDraftStoreState( if (typeof threadKey !== "string" || threadKey.length === 0) { continue; } - const cached = partializedDraftCache.get(draft); - if (cached !== undefined) { - if (cached !== null) { - persistedDraftsByThreadKey[threadKey] = - cached as DeepMutable; - } - continue; - } const persisted = partializeComposerThreadDraft(draft); - partializedDraftCache.set(draft, persisted); if (persisted !== null) { persistedDraftsByThreadKey[threadKey] = persisted as DeepMutable; diff --git a/apps/web/src/connection/useDesktopLocalBootstraps.ts b/apps/web/src/connection/useDesktopLocalBootstraps.ts index 3db1708d0f1..fb511bba799 100644 --- a/apps/web/src/connection/useDesktopLocalBootstraps.ts +++ b/apps/web/src/connection/useDesktopLocalBootstraps.ts @@ -1,5 +1,5 @@ import type { DesktopEnvironmentBootstrap } from "@t3tools/contracts"; -import { useSyncExternalStore } from "react"; +import { useEffect, useState } from "react"; import { readDesktopSecondaryBootstraps } from "./desktopLocal"; @@ -26,53 +26,27 @@ function bootstrapsEqual( }); } -// One shared poller for all consumers (sidebar, command palette, ...): a single -// interval runs only while someone is subscribed, and listeners are notified -// only when the topology actually changed — each poll returns a fresh array, -// so publishing it unconditionally would re-render every consumer per tick. -const listeners = new Set<() => void>(); -let currentSnapshot: ReadonlyArray | null = null; -let pollInterval: ReturnType | null = null; - -function poll(): void { - const next = readDesktopSecondaryBootstraps(); - if (currentSnapshot !== null && bootstrapsEqual(currentSnapshot, next)) { - return; - } - currentSnapshot = next; - for (const listener of listeners) { - listener(); - } -} - -function subscribe(listener: () => void): () => void { - listeners.add(listener); - if (pollInterval === null) { - poll(); - pollInterval = setInterval(poll, DESKTOP_LOCAL_BOOTSTRAP_POLL_MS); - } - return () => { - listeners.delete(listener); - if (listeners.size === 0 && pollInterval !== null) { - clearInterval(pollInterval); - pollInterval = null; - } - }; -} - -function getSnapshot(): ReadonlyArray { - // First read happens during render, before subscribe() has polled. - currentSnapshot ??= readDesktopSecondaryBootstraps(); - return currentSnapshot; -} - /** * Reactively track the desktop's secondary local backends (e.g. a parallel WSL - * backend). The bridge exposes no change event, so we re-read on an interval; - * failed reads retain the latest successful snapshot, while a successful empty - * read clears it. Use this instead of polling the bridge ad hoc so every - * renderer consumer reads the same topology. + * backend). The bridge exposes no change event, so each hook instance re-reads + * on its own interval; a poll returns a fresh array, so the previous reference + * is kept when the topology is unchanged to avoid re-rendering the consumer + * every tick. Use this instead of polling the bridge ad hoc. */ export function useDesktopLocalBootstraps(): ReadonlyArray { - return useSyncExternalStore(subscribe, getSnapshot, getSnapshot); + const [bootstraps, setBootstraps] = useState>( + readDesktopSecondaryBootstraps, + ); + + useEffect(() => { + const read = () => { + const next = readDesktopSecondaryBootstraps(); + setBootstraps((previous) => (bootstrapsEqual(previous, next) ? previous : next)); + }; + read(); + const interval = setInterval(read, DESKTOP_LOCAL_BOOTSTRAP_POLL_MS); + return () => clearInterval(interval); + }, []); + + return bootstraps; } diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index bd8fd2f5e47..f2746e23fa6 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -1366,42 +1366,6 @@ function compareActivityLifecycleRank(kind: string): number { return 1; } -function isSortedByCreatedAt(rows: ReadonlyArray): boolean { - for (let index = 1; index < rows.length; index += 1) { - if (rows[index - 1]!.createdAt.localeCompare(rows[index]!.createdAt) > 0) { - return false; - } - } - return true; -} - -// Stable k-way merge of per-kind rows that are each already in createdAt order -// (the reducers maintain that invariant). Equal timestamps keep the same order -// a stable sort of the concatenation would produce. -function mergeTimelineRows(sources: ReadonlyArray>): TimelineEntry[] { - const merged: TimelineEntry[] = []; - const cursors = sources.map(() => 0); - const total = sources.reduce((sum, rows) => sum + rows.length, 0); - while (merged.length < total) { - let nextSource = -1; - for (let index = 0; index < sources.length; index += 1) { - const candidate = sources[index]![cursors[index]!]; - if (candidate === undefined) { - continue; - } - if ( - nextSource === -1 || - candidate.createdAt.localeCompare(sources[nextSource]![cursors[nextSource]!]!.createdAt) < 0 - ) { - nextSource = index; - } - } - merged.push(sources[nextSource]![cursors[nextSource]!]!); - cursors[nextSource] = cursors[nextSource]! + 1; - } - return merged; -} - export function deriveTimelineEntries( messages: ReadonlyArray, proposedPlans: ReadonlyArray, @@ -1425,10 +1389,9 @@ export function deriveTimelineEntries( createdAt: entry.createdAt, entry, })); - const sources = [messageRows, proposedPlanRows, workRows]; - return sources.every(isSortedByCreatedAt) - ? mergeTimelineRows(sources) - : sources.flat().toSorted((a, b) => a.createdAt.localeCompare(b.createdAt)); + return [...messageRows, ...proposedPlanRows, ...workRows].toSorted((a, b) => + a.createdAt.localeCompare(b.createdAt), + ); } export function inferCheckpointTurnCountByTurnId(