- , "timelineEntries"> & { + /** Null until the thread's detail may load; see resolveThreadDetailRef. */ + threadDetailRef: ScopedThreadRef | null; + entriesInput: Omit; + /** A local draft's optimistic rows; null for a server thread. */ + draftEntries: ReadonlyArray | null; + activeThreadKey: string | null; + syncing: boolean; + detailLoading: boolean; + readyMarkdownCwd: string | null; + readyWorkspaceRoot: string | null; +}) { + const visibleTurnItems = useThreadVisibleTurnItems(threadDetailRef); + const serverEntries = useIncrementalTimelineEntries(activeThreadKey, { + ...entriesInput, + visibleTurnItems, + }); + const entries = draftEntries ?? serverEntries; + const displayed = resolveThreadSwitchTimeline({ + loading: entries.length === 0 && syncing, + activeThreadKey, + nextEntries: entries, + rememberedForActive: peekRememberedThreadTimeline(activeThreadKey), + }); + useLayoutEffect(() => { + if (detailLoading || entries.length === 0 || timelineHasEphemeralPreviewUrls(entries)) { + return; + } + rememberReadyThreadTimeline({ + threadKey: activeThreadKey, + entries, + markdownCwd: readyMarkdownCwd, + workspaceRoot: readyWorkspaceRoot, + }); + }, [activeThreadKey, detailLoading, entries, readyMarkdownCwd, readyWorkspaceRoot]); + return ; +} ``` ## apps/web/src/components/chat/useAttachmentPreviewHandoff.ts Status: modified; +2; −2 ```diff @@ -3,12 +3,12 @@ import { useEffect, useLayoutEffect, useRef, useState } from "react"; import { selectHandoffImageResources } from "../../session-logic"; import { type ChatMessage } from "../../types"; -import type { useThreadVisibleTurnItems } from "../../state/entities"; +import type { useQuietThreadVisibleTurnItems } from "../../state/entities"; import { revokeBlobPreviewUrl, revokeUserMessagePreviewUrls } from "../ChatView.logic"; import { useAssetUrls } from "../../assets/assetUrls"; interface AttachmentPreviewHandoffInput { readonly environmentId: EnvironmentId; - readonly serverVisibleTurnItems: ReturnType; + readonly serverVisibleTurnItems: ReturnType; readonly optimisticUserMessages: ReadonlyArray; } ``` ## apps/web/src/components/chat/useComposerSelection.ts Status: modified; +2; −2 ```diff @@ -12,7 +12,7 @@ ProviderInteractionMode, RuntimeMode, } from "@t3tools/contracts"; -import { deriveThreadRuntime } from "@t3tools/client-runtime/state/thread-execution"; +import type { deriveThreadStructureRuntime } from "@t3tools/client-runtime/state/thread-execution"; import { scopeThreadRef } from "@t3tools/client-runtime/environment"; import { type SetStateAction } from "react"; import { useNavigate } from "@tanstack/react-router"; @@ -30,7 +30,7 @@ readonly composerDraftTarget: ComposerThreadTarget; readonly draftThread: DraftSessionState | null; readonly activeThread: EnvironmentThreadShell | undefined; - readonly activeRuntime: ReturnType; + readonly activeRuntime: ReturnType; readonly activeProjectSettings: { readonly settings: { readonly newWorktreesStartFromOrigin: boolean }; }; ``` ## apps/web/src/components/chat/usePendingThreadRequests.ts Status: modified; +2; −2 ```diff @@ -1,7 +1,7 @@ +import type { ThreadStructure } from "@t3tools/client-runtime/state/shell"; import { AuthOrchestrationOperateScope, type EnvironmentId, - type OrchestrationV2ThreadProjection, type ProviderApprovalDecision, type RuntimeRequestId, type ScopedThreadRef, @@ -110,7 +110,7 @@ readonly environmentId: EnvironmentId; readonly activeThreadId: ThreadId | null; readonly activeThread: { readonly id: ThreadId } | undefined; - readonly serverProjection: OrchestrationV2ThreadProjection | null; + readonly serverProjection: ThreadStructure | null; readonly supportsQuestionAttachments: boolean; readonly composerDraftTarget: ScopedThreadRef | DraftId; readonly setComposerDraftPrompt: (target: ScopedThreadRef | DraftId, prompt: string) => void; ``` ## apps/web/src/components/chat/useQueuedRunEdit.ts Status: modified; +2; −2 ```diff @@ -1,10 +1,10 @@ +import type { ThreadStructure } from "@t3tools/client-runtime/state/shell"; import { PROVIDER_SEND_TURN_MAX_ATTACHMENTS, type ChatAttachment, type EnvironmentId, type MessageId, type OrchestrationMessageContext, - type OrchestrationV2ThreadProjection, type PreviewAnnotationPayload, type RunId, type ScopedThreadRef, @@ -170,7 +170,7 @@ readonly threadDraftTarget: ComposerThreadTarget; // Null while the route has no thread yet; an edit needs one to target. readonly activeThreadId: ThreadId | null; - readonly serverProjection: OrchestrationV2ThreadProjection | null; + readonly serverProjection: ThreadStructure | null; readonly promptRef: { current: string }; readonly composerRef: RefObject; }) { ``` ## apps/web/src/components/chat/useThreadFind.ts Status: modified; +5; −7 ```diff @@ -1,13 +1,10 @@ import type { InlineSkill } from "@t3tools/shared/inlineSkills"; -import type { - OrchestrationV2ThreadProjection, - OrchestrationV2SearchThreadResult, - ScopedThreadRef, +import type { OrchestrationV2SearchThreadResult, ScopedThreadRef } from "@t3tools/contracts"; -} from "@t3tools/contracts"; import { scopedThreadKey } from "@t3tools/client-runtime/environment"; import { useEffect, useLayoutEffect, useRef, useState, type RefObject } from "react"; import { useStableCallback } from "~/hooks/useStableCallback"; import { orchestrationEnvironment } from "~/state/orchestration"; +import { useThreadProjection } from "~/state/entities"; import { useEnvironmentQuery } from "~/state/query"; import { useDebouncedValue } from "~/state/queries"; import { @@ -42,14 +39,15 @@ enabled, progressive = false, skills = EMPTY_SKILLS, - content, }: { thread: ScopedThreadRef | null; enabled: boolean; progressive?: boolean; skills?: readonly InlineSkill[]; - content: Pick | undefined; }) { + // Read here rather than passed in: a streamed delta re-renders only the find + // provider, which refreshes open results from it. + const content = useThreadProjection(enabled ? thread : null)?.projection; // Held in state so the timeline props that carry it can be stored in state too. const [positionReader] = useState>(() => ({ current: null, ``` ## apps/web/src/components/chat/useTimelineFollow.ts Status: modified; +11; −2 ```diff @@ -31,7 +31,7 @@ import { shouldReleaseTimelineAnchorForToolActivity } from "../ChatView.logic"; import { releaseChatTimelineAnchor } from "../ChatView.logic"; import type { Dispatch, RefObject } from "react"; -import type { EnvironmentThread } from "@t3tools/client-runtime/state/shell"; +import type { ThreadStructure } from "@t3tools/client-runtime/state/shell"; import type { deriveThreadActivityRun } from "@t3tools/client-runtime/state/thread-execution"; import type { EnvironmentThreadStatus } from "@t3tools/client-runtime/state/threads"; import type { TimelineEntry } from "../../session-logic"; @@ -52,7 +52,7 @@ readonly activeThreadKey: string | null; readonly activeThreadId: ThreadId | undefined; readonly isServerThread: boolean; - readonly serverProjection: EnvironmentThread["projection"] | null; + readonly serverProjection: ThreadStructure | null; readonly threadStatus: EnvironmentThreadStatus; readonly activeActivityRun: ReturnType; readonly activeRunningTurnId: RunId | null; @@ -586,9 +586,17 @@ }; const onIsAtEndChangeStable = useStableCallback(onIsAtEndChange); + // timelineEntries leave out reasoning rows (their text is not in the + // structure), and reasoning counts as tool activity, like any work row. + const runningTurnHasReasoning = + activeRunningTurnId !== null && + serverProjection?.visibleTurnItems.some( + (row) => row.item.type === "reasoning" && row.item.runId === activeRunningTurnId, + ) === true; useLayoutEffect(() => { if (timelineScrollModeRef.current !== "anchoring-new-turn") return; if ( + (runningTurnHasReasoning && timelineAnchorMessageId !== null && timelineLiveFollowEnabled) || shouldReleaseTimelineAnchorForToolActivity({ anchorMessageId: timelineAnchorMessageId, liveFollowEnabled: timelineLiveFollowEnabled, @@ -600,6 +608,7 @@ } }, [ activeRunningTurnId, + runningTurnHasReasoning, scrollToEndStable, timelineAnchorMessageId, timelineEntries, ``` ## apps/web/src/components/chat/useTimelineRewind.ts Status: modified; +2; −3 ```diff @@ -1,5 +1,4 @@ -import type { EnvironmentThread } from "@t3tools/client-runtime/state/shell"; +import type { EnvironmentThreadShell, ThreadStructure } from "@t3tools/client-runtime/state/shell"; -import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell"; import type { ComposerThreadTarget } from "../../composerDraftStore"; import type { ComposerHandleRef } from "../../composerHandleContext"; import type { SessionPhase } from "../../types"; @@ -39,7 +38,7 @@ readonly routeThreadRef: ScopedThreadRef; readonly routeThreadKey: string; readonly currentRouteThreadKeyRef: { readonly current: string | null }; - readonly serverProjection: EnvironmentThread["projection"] | null; + readonly serverProjection: ThreadStructure | null; readonly activeThread: EnvironmentThreadShell | undefined; readonly timelineMessages: ReadonlyArray; readonly paintOnlyDisplayedTimeline: boolean; ``` ## apps/web/src/components/chat/useWorktreeSetup.ts Status: modified; +2; −2 ```diff @@ -1,6 +1,6 @@ +import type { ThreadStructure } from "@t3tools/client-runtime/state/shell"; import type { EnvironmentId, - OrchestrationV2ThreadProjection, RunId, ScopedThreadRef, ThreadId, @@ -34,7 +34,7 @@ | { readonly id: ThreadId; readonly worktreePath: string | null } | undefined; readonly threadRef: ScopedThreadRef | null; - readonly serverProjection: OrchestrationV2ThreadProjection | null; + readonly serverProjection: ThreadStructure | null; readonly activityRun: Thread["latestRun"] | null; readonly localPreparing: boolean; // A send that the server queues behind the setup run is not held here. ``` ## apps/web/src/lib/contextWindow.ts Status: modified; +2; −2 ```diff @@ -1,9 +1,9 @@ +import type { ThreadStructureTurnItem } from "@t3tools/client-runtime/state/shell"; import type { OrchestrationV2ProviderTurn, OrchestrationV2ProviderTurnTokenUsage, OrchestrationV2ProviderThread, OrchestrationV2ThreadProjection, - OrchestrationV2TurnItem, ThreadTokenUsageSnapshot, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -58,7 +58,7 @@ export function deriveLatestContextWindowSnapshot( entries: ReadonlyArray<{ - readonly item: OrchestrationV2TurnItem; + readonly item: ThreadStructureTurnItem; }>, liveUsage?: OrchestrationV2ProviderTurnTokenUsage | null, providerThread?: Pick | null, ``` ## apps/web/src/session-logic.ts Status: modified; +23; −12 ```diff @@ -1,3 +1,4 @@ +import type { ThreadStructure, ThreadStructureRuntime } from "@t3tools/client-runtime/state/shell"; import { resolveThreadWorkingStartedAt } from "@t3tools/client-runtime/state/models"; import { type AssetResource, @@ -251,7 +252,7 @@ } export function deriveActivePlanState( + projection: ThreadStructure | null, - projection: OrchestrationV2ThreadProjection | null, latestRunId: RunId | undefined, ): ActivePlanState | null { if (projection === null) return null; @@ -273,7 +274,7 @@ ); if (plan === null || plan.steps.length === 0) return null; return { - createdAt: planItemTime(projection, plan.id), + createdAt: planItemTime(projection, plan), runId: plan.runId, explanation: plan.explanation ?? null, steps: plan.steps.map(({ text, status, durationMs }) => ({ @@ -286,7 +287,7 @@ // Claude's unfinished tasks belong to the provider session, not just the turn. export function selectComposerTasksPlan( - projection: OrchestrationV2ThreadProjection | null, + projection: ThreadStructure | null, plan: ActivePlanState | null, activeRunId: RunId | null, activeRunSettled: boolean, @@ -315,20 +316,30 @@ return latestStartedRun?.providerThreadId === planProviderThreadId ? plan : null; } -function planItemTime(projection: OrchestrationV2ThreadProjection, planId: PlanId): string { +function planItemTime( + projection: ThreadStructure, + plan: Pick, +): string { const item = projection.turnItems.findLast( (candidate) => (candidate.type === "proposed_plan" || candidate.type === "todo_list") && + candidate.planId === plan.id, + ); + if (item?.type === "proposed_plan" || item?.type === "todo_list") { + return DateTime.formatIso(item.updatedAt); + } + // Without a plan item, the plan's run dates it; the thread's own timestamps move while a reply streams. + const run = projection.runs.find((candidate) => candidate.id === plan.runId); + return DateTime.formatIso( + run?.completedAt ?? run?.startedAt ?? run?.requestedAt ?? projection.thread.createdAt, - candidate.planId === planId, ); - return DateTime.formatIso(item?.updatedAt ?? projection.updatedAt); } function toLatestProposedPlanState( - projection: OrchestrationV2ThreadProjection, + projection: ThreadStructure, plan: Extract, ): LatestProposedPlanState { - const updatedAt = planItemTime(projection, plan.id); + const updatedAt = planItemTime(projection, plan); return { id: plan.id, createdAt: updatedAt, @@ -340,7 +351,7 @@ } export function findLatestProposedPlan( - projection: OrchestrationV2ThreadProjection | null, + projection: ThreadStructure | null, latestRunId: RunId | string | null | undefined, ): LatestProposedPlanState | null { if (projection === null) return null; @@ -349,7 +360,7 @@ const plan = [...(candidates.length > 0 ? candidates : plans)] .toSorted( (left, right) => - planItemTime(projection, left.id).localeCompare(planItemTime(projection, right.id)) || + planItemTime(projection, left).localeCompare(planItemTime(projection, right)) || left.id.localeCompare(right.id), ) .at(-1); @@ -1122,7 +1133,7 @@ ); } -export function derivePhase(runtime: ThreadRuntimeSummary | null): SessionPhase { +export function derivePhase(runtime: ThreadStructureRuntime | null): SessionPhase { if (runtime === null) return "disconnected"; if ( runtime.status === "preparing" || @@ -1140,7 +1151,7 @@ */ export function deriveCanInterruptRunningThread( hasActiveThread: boolean, - runtime: ThreadRuntimeSummary | null, + runtime: ThreadStructureRuntime | null, ): boolean { return ( hasActiveThread && ``` ## apps/web/src/state/entities.ts Status: modified; +33; −0 ```diff @@ -1,8 +1,11 @@ +import type * as DateTime from "effect/DateTime"; import { useAtomValue } from "@effect/atom-react"; import type { EnvironmentProject, EnvironmentThread, EnvironmentThreadShell, + EnvironmentThreadStructure, + ThreadStructureVisibleTurnItem, } from "@t3tools/client-runtime/state/shell"; import { EMPTY_THREAD_HISTORY_META, @@ -62,6 +65,9 @@ const EMPTY_CHECKPOINT_SUMMARIES_ATOM = Atom.make>( Object.freeze([]), ).pipe(Atom.withLabel("web-thread-checkpoint-summaries:empty")); +const EMPTY_UPDATED_AT_ATOM = Atom.make(null).pipe( + Atom.withLabel("web-thread-updated-at:empty"), +); const EMPTY_THREAD_HISTORY_ATOM = Atom.make(EMPTY_THREAD_HISTORY_META).pipe( Atom.withLabel("web-thread-history:empty"), ); @@ -156,6 +162,22 @@ ); } +/** The thread's structure, held through streamed assistant and reasoning text; see ThreadStructure. */ +export function useQuietThreadProjection( + ref: ScopedThreadRef | null, +): EnvironmentThreadStructure | null { + return useAtomValue( + ref === null ? EMPTY_THREAD_PROJECTION_ATOM : environmentThreadDetails.quietThreadAtom(ref), + ); +} + +/** The detail's live `updatedAt`, which the thread structure leaves out. */ +export function useThreadUpdatedAt(ref: ScopedThreadRef | null): DateTime.Utc | null { + return useAtomValue( + ref === null ? EMPTY_UPDATED_AT_ATOM : environmentThreadDetails.updatedAtAtom(ref), + ); +} + export function useThreadStatus(ref: ScopedThreadRef | null): EnvironmentThreadStatus { return useAtomValue( ref === null ? EMPTY_THREAD_STATUS_ATOM : environmentThreadDetails.statusAtom(ref), @@ -188,6 +210,17 @@ ); } +/** Visible turn items without their streamed text; see ThreadStructure. */ +export function useQuietThreadVisibleTurnItems( + ref: ScopedThreadRef | null, +): ReadonlyArray { + return useAtomValue( + ref === null + ? EMPTY_VISIBLE_TURN_ITEMS_ATOM + : environmentThreadDetails.quietVisibleTurnItemsAtom(ref), + ); +} + export function useThreadCheckpointSummaries( ref: ScopedThreadRef | null, ): ReadonlyArray { ``` ## packages/client-runtime/src/handoff.ts Status: modified; +3; −2 ```diff @@ -1,12 +1,13 @@ import type { ModelSelection, OrchestrationV2Run, - OrchestrationV2ThreadProjection, OrchestrationV2TurnItem, ProviderInstanceId, ServerProvider, } from "@t3tools/contracts"; +import type { ThreadStructure } from "./state/models.ts"; + export type HandoffTimelineRun = Pick< OrchestrationV2Run, "id" | "ordinal" | "providerInstanceId" | "modelSelection" @@ -15,7 +16,7 @@ export type HandoffEndpoints = ReturnType; export function resolvePendingHandoff( - projection: OrchestrationV2ThreadProjection | null | undefined, + projection: ThreadStructure | null | undefined, selection: ModelSelection | null | undefined, providers: ReadonlyArray, ): HandoffEndpoints | null { ``` ## packages/client-runtime/src/state/models.ts Status: modified; +41; −1 ```diff @@ -34,6 +34,43 @@ readonly projection: OrchestrationV2ThreadProjection; } +type ProjectionTurnItem = OrchestrationV2ThreadProjection["turnItems"][number]; +type StreamedProjectionTurnItem = Extract< + ProjectionTurnItem, + { readonly type: "assistant_message" | "reasoning" } +>; + +/** A turn item without the fields that move while it streams. */ +export type ThreadStructureTurnItem = + | Exclude + | Omit, "text" | "updatedAt"> + | Omit, "text" | "updatedAt">; + +export type ThreadStructureVisibleTurnItem = Omit< + OrchestrationV2ThreadProjection["visibleTurnItems"][number], + "item" +> & { + readonly item: ThreadStructureTurnItem; +}; + +/** + * A thread projection without the fields that move while a reply streams, so a + * reader of the quiet thread cannot depend on them and go stale. A full + * projection is assignable to it. + */ +export type ThreadStructure = Omit< + OrchestrationV2ThreadProjection, + "turnItems" | "visibleTurnItems" | "updatedAt" +> & { + readonly turnItems: ReadonlyArray; + readonly visibleTurnItems: ReadonlyArray; +}; + +export interface EnvironmentThreadStructure { + readonly environmentId: EnvironmentId; + readonly projection: ThreadStructure; +} + export interface ThreadRunSummary { readonly runId: RunId; readonly status: OrchestrationV2RunStatus; @@ -65,7 +102,10 @@ readonly updatedAt: string; } +/** The runtime summary a thread's structure determines; see ThreadStructure. */ +export type ThreadStructureRuntime = Omit; + -export function threadRuntimeIsActive(runtime: ThreadRuntimeSummary | null | undefined): boolean { +export function threadRuntimeIsActive(runtime: ThreadStructureRuntime | null | undefined): boolean { return runtime !== null && runtime !== undefined && threadRunStatusIsActive(runtime.status); } ``` ## packages/client-runtime/src/state/threadDetail.test.ts Status: added; +136; −0 ```diff @@ -1,0 +1,136 @@ +import { + EnvironmentId, + MessageId, + RunId, + TurnItemId, + type OrchestrationV2ThreadProjection, + type OrchestrationV2TurnItem, +} from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as DateTime from "effect/DateTime"; +import * as Option from "effect/Option"; +import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity"; + +import { v2Now, v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; +import { createEnvironmentThreadDetailAtoms, sameThreadStructure } from "./threadDetail.ts"; +import { EMPTY_ENVIRONMENT_THREAD_STATE } from "./threadState.ts"; + +const later = DateTime.makeUnsafe("2026-06-20T00:00:01.000Z"); + +const assistant: OrchestrationV2TurnItem = { + id: TurnItemId.make("item-assistant"), + threadId: v2ThreadId, + runId: RunId.make("run-1"), + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: null, + startedAt: v2Now, + completedAt: null, + updatedAt: v2Now, + type: "assistant_message", + messageId: MessageId.make("message-assistant"), + text: "Hel", + streaming: true, +}; + +function withItem(item: OrchestrationV2TurnItem, updatedAt = v2Now) { + return { + ...v2Projection, + turnItems: [item], + visibleTurnItems: [ + { + position: 0, + visibility: "local" as const, + sourceThreadId: v2ThreadId, + sourceItemId: item.id, + item, + }, + ], + updatedAt, + }; +} + +describe("sameThreadStructure", () => { + const before = withItem(assistant); + + it("holds through streamed text", () => { + const next = withItem({ ...assistant, text: "Hello", updatedAt: later }, later); + expect(sameThreadStructure(before, next)).toBe(true); + }); + + it("sees the item finish streaming", () => { + const next = withItem({ ...assistant, text: "Hello", streaming: false, status: "completed" }); + expect(sameThreadStructure(before, next)).toBe(false); + }); + + it("sees a new item", () => { + const next = { ...before, turnItems: [assistant, { ...assistant, id: TurnItemId.make("b") }] }; + expect(sameThreadStructure(before, next)).toBe(false); + }); + + it("sees other fields change on a streaming item", () => { + const next = withItem({ ...assistant, text: "Hello", title: "Answer" }); + expect(sameThreadStructure(before, next)).toBe(false); + }); + + it("sees changes outside the turn items", () => { + expect(sameThreadStructure(before, { ...before, runs: [] })).toBe(false); + }); +}); + +describe("quietThreadAtom", () => { + const ref = { environmentId: EnvironmentId.make("environment-1"), threadId: v2ThreadId }; + + function setup(initial: OrchestrationV2ThreadProjection) { + const stateAtom = Atom.make( + AsyncResult.success({ ...EMPTY_ENVIRONMENT_THREAD_STATE, data: Option.some(initial) }), + ); + const quietAtom = createEnvironmentThreadDetailAtoms(() => stateAtom).quietThreadAtom(ref); + const registry = AtomRegistry.make(); + registry.mount(quietAtom); + return { + read: () => registry.get(quietAtom), + push: (projection: OrchestrationV2ThreadProjection) => + registry.set( + stateAtom, + AsyncResult.success({ ...EMPTY_ENVIRONMENT_THREAD_STATE, data: Option.some(projection) }), + ), + }; + } + + it("keeps the same value through streamed text", () => { + const thread = setup(withItem(assistant)); + const first = thread.read(); + thread.push(withItem({ ...assistant, text: "Hello", updatedAt: later }, later)); + thread.push(withItem({ ...assistant, text: "Hello there", updatedAt: later }, later)); + expect(thread.read()).toBe(first); + }); + + it("passes a structure change on", () => { + const thread = setup(withItem(assistant)); + const first = thread.read(); + thread.push(withItem({ ...assistant, text: "Hello", updatedAt: later }, later)); + thread.push(withItem({ ...assistant, text: "Hello", status: "completed", streaming: false })); + const next = thread.read(); + expect(next).not.toBe(first); + expect(next?.projection.turnItems[0]).toMatchObject({ status: "completed", streaming: false }); + }); + + it("leaves out the fields that streaming moves", () => { + const thread = setup(withItem(assistant)); + const projection = thread.read()?.projection; + expect(projection).not.toHaveProperty("updatedAt"); + expect(projection?.turnItems[0]).not.toHaveProperty("text"); + expect(projection?.turnItems[0]).not.toHaveProperty("updatedAt"); + expect(projection?.visibleTurnItems[0]?.item).not.toHaveProperty("text"); + expect(projection?.visibleTurnItems[0]?.item).toMatchObject({ + messageId: assistant.messageId, + streaming: true, + }); + }); +}); ``` ## packages/client-runtime/src/state/threadDetail.ts Status: modified; +167; −2 ```diff @@ -1,5 +1,10 @@ +import { + OrchestrationV2TurnItem, -import type { OrchestrationV2ThreadProjection, ScopedThreadRef } from "@t3tools/contracts"; + type OrchestrationV2ThreadProjection, + type ScopedThreadRef, +} from "@t3tools/contracts"; import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; import { AsyncResult, Atom } from "effect/reactivity"; import { @@ -12,7 +17,13 @@ deriveThreadCheckpointSummaries, type ThreadCheckpointSummary, } from "./threadCheckpoints.ts"; +import type { + EnvironmentThread, -import type { EnvironmentThread } from "./models.ts"; + EnvironmentThreadStructure, + ThreadStructure, + ThreadStructureTurnItem, + ThreadStructureVisibleTurnItem, +} from "./models.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threadState.ts"; import { createQuestionHistoryProjector, @@ -24,8 +35,114 @@ const EMPTY_VISIBLE_TURN_ITEMS: OrchestrationV2ThreadProjection["visibleTurnItems"] = Object.freeze( [], ); +const EMPTY_THREAD_STRUCTURE_TURN_ITEMS: ReadonlyArray = + Object.freeze([]); const EMPTY_CHECKPOINT_SUMMARIES: ReadonlyArray = Object.freeze([]); +type TurnItem = OrchestrationV2ThreadProjection["turnItems"][number]; +type VisibleTurnItem = OrchestrationV2ThreadProjection["visibleTurnItems"][number]; + +// Items are stripped once each: unchanged items keep their identity across +// structure changes, which the incremental timeline relies on. +const structureTurnItems = new WeakMap(); +const structureVisibleTurnItems = new WeakMap(); + +function toThreadStructureTurnItem(item: TurnItem): ThreadStructureTurnItem { + if (item.type !== "assistant_message" && item.type !== "reasoning") return item; + let structure = structureTurnItems.get(item); + if (structure === undefined) { + const { text: _text, updatedAt: _updatedAt, ...rest } = item; + structure = rest; + structureTurnItems.set(item, structure); + } + return structure; +} + +function toThreadStructureVisibleTurnItem( + visibleItem: VisibleTurnItem, +): ThreadStructureVisibleTurnItem { + const item = toThreadStructureTurnItem(visibleItem.item); + if (item === visibleItem.item) return visibleItem; + let structure = structureVisibleTurnItems.get(visibleItem); + if (structure === undefined) { + structure = { ...visibleItem, item }; + structureVisibleTurnItems.set(visibleItem, structure); + } + return structure; +} + +/** Whether the structure kept every field of this row: true for all but streamed items. */ +export function isCompleteTurnItemRow(row: ThreadStructureVisibleTurnItem): row is VisibleTurnItem { + return row.item.type !== "assistant_message" && row.item.type !== "reasoning"; +} + +function toThreadStructure(projection: OrchestrationV2ThreadProjection): ThreadStructure { + const { updatedAt: _updatedAt, ...structure } = projection; + return { + ...structure, + turnItems: projection.turnItems.map(toThreadStructureTurnItem), + visibleTurnItems: projection.visibleTurnItems.map(toThreadStructureVisibleTurnItem), + }; +} + +const sameTurnItem = Schema.toEquivalence(OrchestrationV2TurnItem); + +// Compares through toThreadStructureTurnItem, so the fields it strips are +// exactly the ones ignored here. +function sameTurnItemStructure(previous: TurnItem, next: TurnItem): boolean { + if (previous === next) return true; + const structure = toThreadStructureTurnItem(previous); + if (previous.id !== next.id || previous.type !== next.type || structure === previous) { + return false; + } + return ( + Object.keys(previous).length === Object.keys(next).length && + sameTurnItem(Object.assign({}, next, structure), next) + ); +} + +/** + * Whether `previous` and `next` have the same ThreadStructure: they differ + * only by text streamed into existing assistant and reasoning items. The same + * items exist, with the same status, and every other field is unchanged. + */ +export function sameThreadStructure( + previous: OrchestrationV2ThreadProjection, + next: OrchestrationV2ThreadProjection, +): boolean { + if (previous === next) return true; + for (const key of Object.keys(next) as Array) { + if (key === "turnItems" || key === "visibleTurnItems" || key === "updatedAt") continue; + if (previous[key] !== next[key]) return false; + } + if ( + previous.turnItems.length !== next.turnItems.length || + previous.visibleTurnItems.length !== next.visibleTurnItems.length + ) { + return false; + } + for (let index = 0; index < next.turnItems.length; index += 1) { + if (!sameTurnItemStructure(previous.turnItems[index]!, next.turnItems[index]!)) { + return false; + } + } + for (let index = 0; index < next.visibleTurnItems.length; index += 1) { + const before = previous.visibleTurnItems[index]!; + const after = next.visibleTurnItems[index]!; + if (before === after) continue; + if ( + before.position !== after.position || + before.visibility !== after.visibility || + before.sourceThreadId !== after.sourceThreadId || + before.sourceItemId !== after.sourceItemId || + !sameTurnItemStructure(before.item, after.item) + ) { + return false; + } + } + return true; +} + export function createEnvironmentThreadDetailAtoms( threadStateAtom: ( environmentId: ScopedThreadRef["environmentId"], @@ -55,6 +172,32 @@ }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread:${key}`)); }); + // Streamed assistant and reasoning text replaces an item several times a + // second. Readers of the thread's structure (runs, requests, which items + // exist and their status) keep the previous value through those updates. + const quietThreadAtomFamily = Atom.family((key: string) => { + let previousProjection: OrchestrationV2ThreadProjection | null = null; + let previous: EnvironmentThreadStructure | null = null; + return Atom.make((get) => { + const next = get(threadAtomFamily(key)); + if (next === null) { + previousProjection = null; + previous = null; + } else if ( + previousProjection === null || + !sameThreadStructure(previousProjection, next.projection) + ) { + previous = { + environmentId: next.environmentId, + projection: toThreadStructure(next.projection), + }; + } + // Compared against the latest projection, a delta only compares the item it replaced. + previousProjection = next?.projection ?? null; + return previous; + }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-quiet-thread:${key}`)); + }); + const visibleTurnItemsAtomFamily = Atom.family((key: string) => { const projectQuestionHistory = createQuestionHistoryProjector(); return Atom.make((get): OrchestrationV2ThreadProjection["visibleTurnItems"] => { @@ -63,6 +206,19 @@ }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-visible-turn-items:${key}`)); }); + const quietVisibleTurnItemsAtomFamily = Atom.family((key: string) => { + const projectQuestionHistory = createQuestionHistoryProjector(); + return Atom.make((get) => { + const thread = get(quietThreadAtomFamily(key)); + return thread === null + ? EMPTY_THREAD_STRUCTURE_TURN_ITEMS + : projectQuestionHistory(thread.projection); + }).pipe( + Atom.setIdleTTL(0), + Atom.withLabel(`environment-quiet-thread-visible-turn-items:${key}`), + ); + }); + const queueWorkflowAtomFamily = Atom.family((key: string) => { let previous: Pick< OrchestrationV2ThreadProjection, @@ -148,6 +304,11 @@ }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-checkpoint-summaries:${key}`)); }); + const updatedAtAtomFamily = Atom.family((key: string) => + Atom.make( + (get) => Option.getOrNull(get(threadStateValueAtomFamily(key)).data)?.updatedAt ?? null, + ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-updated-at:${key}`)), + ); const worktreePathAtomFamily = Atom.family((key: string) => Atom.make( (get) => @@ -210,6 +371,7 @@ return { worktreePathAtom: (ref: ScopedThreadRef) => worktreePathAtomFamily(threadKey(ref)), + updatedAtAtom: (ref: ScopedThreadRef) => updatedAtAtomFamily(threadKey(ref)), pendingRequestsAtom: (ref: ScopedThreadRef) => pendingRequestsAtomFamily(threadKey(ref)), queueWorkflowAtom: (ref: ScopedThreadRef) => queueWorkflowAtomFamily(threadKey(ref)), queuedCountAtom: (ref: ScopedThreadRef) => queuedCountAtomFamily(threadKey(ref)), @@ -218,7 +380,10 @@ checkpointSummariesAtomFamily(threadKey(ref)), stateAtom: (ref: ScopedThreadRef) => threadStateValueAtomFamily(threadKey(ref)), threadAtom: (ref: ScopedThreadRef) => threadAtomFamily(threadKey(ref)), + quietThreadAtom: (ref: ScopedThreadRef) => quietThreadAtomFamily(threadKey(ref)), visibleTurnItemsAtom: (ref: ScopedThreadRef) => visibleTurnItemsAtomFamily(threadKey(ref)), + quietVisibleTurnItemsAtom: (ref: ScopedThreadRef) => + quietVisibleTurnItemsAtomFamily(threadKey(ref)), statusAtom: (ref: ScopedThreadRef) => statusAtomFamily(threadKey(ref)), errorAtom: (ref: ScopedThreadRef) => errorAtomFamily(threadKey(ref)), historyAtom: (ref: ScopedThreadRef) => historyAtomFamily(threadKey(ref)), ``` ## packages/client-runtime/src/state/threadExecution.ts Status: modified; +23; −20 ```diff @@ -30,6 +30,8 @@ threadRuntimeIsActive, type ThreadRunSummary, type ThreadRuntimeSummary, + type ThreadStructure, + type ThreadStructureRuntime, } from "./models.ts"; import { formatSubagentDisplayTitle } from "./subagentDisplay.ts"; @@ -37,7 +39,7 @@ const INTERRUPTIBLE_RUN_STATUSES = new Set(["preparing", "starting", "running"]); function latestMatchingRun( - projection: OrchestrationV2ThreadProjection, + projection: ThreadStructure, predicate: (run: OrchestrationV2ThreadProjection["runs"][number]) => boolean, ): OrchestrationV2ThreadProjection["runs"][number] | null { return projection.runs.reduce( @@ -50,7 +52,7 @@ } function summarizeThreadRun( + projection: ThreadStructure, - projection: OrchestrationV2ThreadProjection, run: OrchestrationV2ThreadProjection["runs"][number], ): ThreadRunSummary { const turnStatus = ACTIVITY_RUN_STATUSES.has(run.status) @@ -74,7 +76,7 @@ } function presentedUsageLimitRun( - projection: OrchestrationV2ThreadProjection, + projection: ThreadStructure, ): OrchestrationV2ThreadProjection["runs"][number] | null { const providerSession = projection.providerSessions.findLast( (session) => session.providerInstanceId === projection.thread.providerInstanceId, @@ -87,21 +89,17 @@ } function presentedLatestRun( + projection: ThreadStructure, - projection: OrchestrationV2ThreadProjection, ): OrchestrationV2ThreadProjection["runs"][number] | null { return presentedUsageLimitRun(projection) ?? latestUnheldRun(projection.runs); } -export function deriveLatestThreadRun( - projection: OrchestrationV2ThreadProjection, +export function deriveLatestThreadRun(projection: ThreadStructure): ThreadRunSummary | null { -): ThreadRunSummary | null { const run = presentedLatestRun(projection); return run === null ? null : summarizeThreadRun(projection, run); } -export function deriveThreadActivityRun( +export function deriveThreadActivityRun(projection: ThreadStructure): ThreadRunSummary | null { - projection: OrchestrationV2ThreadProjection, -): ThreadRunSummary | null { const run = latestMatchingRun(projection, (candidate) => ACTIVITY_RUN_STATUSES.has(candidate.status)) ?? presentedLatestRun(projection); @@ -112,9 +110,7 @@ * Provider-native subagent threads never get app runs: their work is a runless * root turn whose status follows the subagent. */ +export function deriveRunlessWorkStartedAt(projection: ThreadStructure): string | null { -export function deriveRunlessWorkStartedAt( - projection: OrchestrationV2ThreadProjection, -): string | null { const status = deriveProviderSubagentStatus(projection); return status !== null && isOrchestrationV2WorkActive(status.status) ? status.startedAt : null; } @@ -126,7 +122,7 @@ } export function deriveProviderSubagentStatus( - projection: OrchestrationV2ThreadProjection, + projection: ThreadStructure, ): ProviderSubagentStatus | null { if (!isProviderNativeSubagentThread(projection.thread)) return null; const node = projection.nodes.findLast( @@ -140,9 +136,7 @@ }; } -export function deriveReportedModelSelection( - projection: OrchestrationV2ThreadProjection, +export function deriveReportedModelSelection(projection: ThreadStructure): ModelSelection | null { -): ModelSelection | null { const providerThread = projection.providerThreads.find( (candidate) => candidate.id === projection.thread.activeProviderThreadId && @@ -215,8 +209,18 @@ } export function deriveThreadRuntime( - projection: OrchestrationV2ThreadProjection, + projection: ThreadStructure & Pick, ): ThreadRuntimeSummary | null { + const runtime = deriveThreadStructureRuntime(projection); + return runtime === null + ? null + : { ...runtime, updatedAt: DateTime.formatIso(projection.updatedAt) }; +} + +/** The runtime summary a thread's structure determines: all of it but `updatedAt`. */ +export function deriveThreadStructureRuntime( + projection: ThreadStructure, +): ThreadStructureRuntime | null { const latestRun = deriveLatestThreadRun(projection); const providerSession = projection.providerSessions.findLast( (session) => session.providerInstanceId === projection.thread.providerInstanceId, @@ -264,12 +268,11 @@ latestRootProviderFailure(latestRunProjection, projection.turnItems), providerSession?.lastError ?? null, ), - updatedAt: DateTime.formatIso(projection.updatedAt), }; } export function threadRuntimeHasInterruptibleRun( - runtime: ThreadRuntimeSummary | null | undefined, + runtime: ThreadStructureRuntime | null | undefined, ): boolean { return ( threadRuntimeIsActive(runtime) && ``` ## packages/client-runtime/src/state/threadRequests.ts Status: modified; +13; −8 ```diff @@ -10,6 +10,7 @@ } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import type { AtomCommandResult } from "./runtime.ts"; +import type { ThreadStructure, ThreadStructureVisibleTurnItem } from "./models.ts"; export type ApprovalResponder = ( requestId: RuntimeRequestId, @@ -93,7 +94,7 @@ } export function derivePendingThreadRequests( + projection: Pick, - projection: Pick, ): PendingThreadRequests { const approvals: ThreadPendingApproval[] = []; const userInputs: ThreadPendingUserInput[] = []; @@ -147,21 +148,25 @@ return { approvals, userInputs }; } +export function createQuestionHistoryProjector< + Row extends ThreadStructureVisibleTurnItem = + OrchestrationV2ThreadProjection["visibleTurnItems"][number], +>() { + type Rows = ReadonlyArray; -export function createQuestionHistoryProjector() { - type Rows = OrchestrationV2ThreadProjection["visibleTurnItems"]; type Requests = OrchestrationV2ThreadProjection["runtimeRequests"]; let previousRows: Rows | undefined; let previousRequests: Requests | undefined; let previousResult: Rows | undefined; + const cache = new WeakMap(); - const cache = new WeakMap(); + return (projection: { - return ( - projection: Pick, - ): Rows => { + readonly visibleTurnItems: Rows; + readonly runtimeRequests: Requests; + }): Rows => { const { visibleTurnItems: rows, runtimeRequests: requests } = projection; if (rows === previousRows && requests === previousRequests && previousResult) return previousResult; const byId = new Map(requests.map((request) => [request.id, request])); - let result: Rows[number][] | undefined; + let result: Row[] | undefined; for (const [index, row] of rows.entries()) { const item = row.item; const request = ``` [Next page](https://git.cafe/pingdotgg/t3code/pull/23117/diffs?after=InBhY2thZ2VzL2NsaWVudC1ydW50aW1lL3NyYy9zdGF0ZS90aHJlYWRSZXF1ZXN0cy50cyI&expectedVersion=10&format=markdown)