From 6e9701d1c657ed53cd5020bd95e83119e9fb154a Mon Sep 17 00:00:00 2001 From: Mike Olson Date: Sun, 12 Jul 2026 17:09:17 -0400 Subject: [PATCH 1/2] fix(mobile): Stabilize CTM iOS draft attachments, Hermes sort, and Home Persist dual-tagged draft image attachments, replace Hermes-missing toSorted with copySorted on mobile-reachable paths, hide subagent threads from Home, and stabilize Home native-stack options to avoid PreventRemove max-update-depth. --- apps/mobile/src/features/home/HomeHeader.tsx | 122 ++++++++++-------- .../src/features/home/HomeRouteScreen.tsx | 29 +++-- .../src/features/home/homeThreadList.test.ts | 48 +++++++ .../src/features/home/homeThreadList.ts | 3 + .../threads/ThreadRelationshipsBanner.tsx | 4 +- apps/mobile/src/state/use-thread-selection.ts | 3 +- .../src/operations/commands.test.ts | 109 ++++++++++++++++ .../client-runtime/src/operations/commands.ts | 48 ++++--- .../src/state/threadWorkflows.ts | 25 ++-- packages/shared/package.json | 4 + packages/shared/src/Array.test.ts | 12 ++ packages/shared/src/Array.ts | 6 + packages/shared/src/model.ts | 18 ++- 13 files changed, 330 insertions(+), 101 deletions(-) create mode 100644 packages/shared/src/Array.test.ts create mode 100644 packages/shared/src/Array.ts diff --git a/apps/mobile/src/features/home/HomeHeader.tsx b/apps/mobile/src/features/home/HomeHeader.tsx index 652dc02fccb..a131eb8447a 100644 --- a/apps/mobile/src/features/home/HomeHeader.tsx +++ b/apps/mobile/src/features/home/HomeHeader.tsx @@ -3,8 +3,12 @@ import type { SidebarProjectGroupingMode, SidebarThreadSortOrder, } from "@t3tools/contracts"; -import { NativeHeaderToolbar, NativeStackScreenOptions } from "../../native/StackHeader"; -import { useCallback, useRef } from "react"; +import { + NativeHeaderToolbar, + NativeStackScreenOptions, + type AppNativeStackNavigationOptions, +} from "../../native/StackHeader"; +import { useCallback, useMemo, useRef } from "react"; import { Platform } from "react-native"; import type { SearchBarCommands } from "react-native-screens"; @@ -43,6 +47,8 @@ export function HomeHeader(props: { readonly onStartNewTask: () => void; }) { const searchBarRef = useRef(null); + const propsRef = useRef(props); + propsRef.current = props; const iconColor = useThemeColor("--color-icon"); const hasCustomListOptions = hasCustomHomeListOptions(props); const focusSearch = useCallback(() => { @@ -50,63 +56,67 @@ export function HomeHeader(props: { return searchBarRef.current !== null; }, []); useHardwareKeyboardCommand("focusSearch", focusSearch); - const filterMenu = buildHomeListFilterMenu(props); + const screenOptions = useMemo((): AppNativeStackNavigationOptions => { + return { + headerTintColor: iconColor, + unstable_headerRightItems: + Platform.OS === "ios" + ? () => [ + withNativeGlassHeaderItem({ + accessibilityLabel: "Open settings", + icon: { name: "ellipsis", type: "sfSymbol" } as const, + identifier: "home-settings", + label: "", + onPress: () => propsRef.current.onOpenSettings(), + type: "button", + }), + ] + : undefined, + unstable_headerToolbarItems: + Platform.OS === "ios" + ? () => [ + createNativeMailSearchToolbarItem({ + composeButtonId: "home-new-task", + composeSystemImageName: "square.and.pencil", + // Build at invocation time so filter onPress handlers always + // read current callbacks via propsRef, matching compose/settings. + filterMenu: buildHomeListFilterMenu(propsRef.current), + filterButtonId: "home-filter", + filterSystemImageName: hasCustomListOptions + ? "line.3.horizontal.decrease.circle.fill" + : "line.3.horizontal.decrease", + onComposePress: () => propsRef.current.onStartNewTask(), + onSearchTextChange: (query) => propsRef.current.onSearchQueryChange(query), + placeholder: "Search", + searchTextChangeId: "home-search-text", + }), + ] + : undefined, + headerSearchBarOptions: + Platform.OS === "ios" + ? undefined + : { + ref: searchBarRef, + allowToolbarIntegration: true, + hideNavigationBar: false, + placeholder: "Search", + onCancelButtonPress: () => propsRef.current.onSearchQueryChange(""), + onChangeText: (event) => propsRef.current.onSearchQueryChange(event.nativeEvent.text), + }, + }; + }, [ + hasCustomListOptions, + iconColor, + props.environments, + props.projectGroupingMode, + props.projectSortOrder, + props.selectedEnvironmentId, + props.threadSortOrder, + ]); return ( <> - [ - withNativeGlassHeaderItem({ - accessibilityLabel: "Open settings", - icon: { name: "ellipsis", type: "sfSymbol" } as const, - identifier: "home-settings", - label: "", - onPress: props.onOpenSettings, - type: "button", - }), - ] - : undefined, - unstable_headerToolbarItems: - Platform.OS === "ios" - ? () => [ - createNativeMailSearchToolbarItem({ - composeButtonId: "home-new-task", - composeSystemImageName: "square.and.pencil", - filterMenu, - filterButtonId: "home-filter", - filterSystemImageName: hasCustomListOptions - ? "line.3.horizontal.decrease.circle.fill" - : "line.3.horizontal.decrease", - onComposePress: props.onStartNewTask, - onSearchTextChange: props.onSearchQueryChange, - placeholder: "Search", - searchTextChangeId: "home-search-text", - }), - ] - : undefined, - headerSearchBarOptions: - Platform.OS === "ios" - ? undefined - : { - ref: searchBarRef, - allowToolbarIntegration: true, - hideNavigationBar: false, - placeholder: "Search", - onCancelButtonPress: () => { - props.onSearchQueryChange(""); - }, - onChangeText: (event) => { - props.onSearchQueryChange(event.nativeEvent.text); - }, - }, - }} - /> + {Platform.OS === "ios" ? null : ( diff --git a/apps/mobile/src/features/home/HomeRouteScreen.tsx b/apps/mobile/src/features/home/HomeRouteScreen.tsx index eaea597211f..b4a14ffb1a3 100644 --- a/apps/mobile/src/features/home/HomeRouteScreen.tsx +++ b/apps/mobile/src/features/home/HomeRouteScreen.tsx @@ -1,7 +1,7 @@ import * as Arr from "effect/Array"; import * as Order from "effect/Order"; import { useNavigation } from "@react-navigation/native"; -import { useMemo, useState } from "react"; +import { useCallback, useMemo, useState } from "react"; import { NativeHeaderToolbar, NativeStackScreenOptions } from "../../native/StackHeader"; import { useProjects, useThreadShells } from "../../state/entities"; @@ -17,6 +17,9 @@ import { useHomeListOptions } from "./home-list-options"; import { usePendingTaskListActions } from "./usePendingTaskListActions"; import { useThreadListActions } from "./useThreadListActions"; +const EMPTY_HOME_TITLE_OPTIONS = { title: "", headerTitle: "" } as const; +const THREADS_HOME_TITLE_OPTIONS = { title: "Threads", headerTitle: "Threads" } as const; + /* ─── Route screen ───────────────────────────────────────────────────── */ export function HomeRouteScreen() { @@ -56,25 +59,29 @@ export function HomeRouteScreen() { setThreadSortOrder, } = useHomeListOptions(availableEnvironmentIds); const selectedEnvironmentId = listOptions.selectedEnvironmentId; + const openSettings = useCallback(() => { + navigation.navigate("SettingsSheet", { screen: "Settings" }); + }, [navigation]); + const openNewTask = useCallback(() => { + navigation.navigate("NewTaskSheet", { screen: "NewTask" }); + }, [navigation]); // In split layouts the persistent sidebar IS the thread list — Home becomes // an empty detail pane so selecting a thread never transitions layouts. if (layout.usesSplitView) { return ( <> - + navigation.navigate("NewTaskSheet", { screen: "NewTask" })} + onPress={openNewTask} /> } /> - navigation.navigate("NewTaskSheet", { screen: "NewTask" })} - /> + ); } @@ -82,7 +89,7 @@ export function HomeRouteScreen() { return ( <> {/* Restore the compact title in case the split branch blanked it. */} - + navigation.navigate("SettingsSheet", { screen: "Settings" })} + onOpenSettings={openSettings} onProjectGroupingModeChange={setProjectGroupingMode} onProjectSortOrderChange={setProjectSortOrder} onSearchQueryChange={setSearchQuery} - onStartNewTask={() => navigation.navigate("NewTaskSheet", { screen: "NewTask" })} + onStartNewTask={openNewTask} onThreadSortOrderChange={setThreadSortOrder} /> @@ -110,7 +117,7 @@ export function HomeRouteScreen() { onOpenEnvironments={() => navigation.navigate("SettingsSheet", { screen: "SettingsEnvironments" }) } - onOpenSettings={() => navigation.navigate("SettingsSheet", { screen: "Settings" })} + onOpenSettings={openSettings} onProjectGroupingModeChange={setProjectGroupingMode} onProjectSortOrderChange={setProjectSortOrder} onSearchQueryChange={setSearchQuery} @@ -132,7 +139,7 @@ export function HomeRouteScreen() { }, }); }} - onStartNewTask={() => navigation.navigate("NewTaskSheet", { screen: "NewTask" })} + onStartNewTask={openNewTask} onThreadSortOrderChange={setThreadSortOrder} pendingTasks={pendingTasks} projectGroupingMode={listOptions.projectGroupingMode} diff --git a/apps/mobile/src/features/home/homeThreadList.test.ts b/apps/mobile/src/features/home/homeThreadList.test.ts index 88686d15778..bd913f8170d 100644 --- a/apps/mobile/src/features/home/homeThreadList.test.ts +++ b/apps/mobile/src/features/home/homeThreadList.test.ts @@ -176,6 +176,54 @@ describe("buildHomeThreadGroups", () => { expect(groups[0]?.threads.map((thread) => thread.environmentId)).toEqual([remoteEnvironmentId]); }); + it("excludes subagent threads", () => { + const environmentId = EnvironmentId.make("environment-1"); + const project = makeProject({ + environmentId, + id: ProjectId.make("project-1"), + title: "T3 Code", + }); + const parentThreadId = ThreadId.make("thread-parent"); + const forkThreadId = ThreadId.make("thread-fork"); + const threads = [ + makeThread({ + environmentId, + id: parentThreadId, + projectId: project.id, + title: "Parent thread", + }), + makeThread({ + environmentId, + id: forkThreadId, + projectId: project.id, + title: "Forked thread", + lineage: { + rootThreadId: parentThreadId, + parentThreadId, + relationshipToParent: "fork", + }, + }), + makeThread({ + environmentId, + id: ThreadId.make("thread-subagent"), + projectId: project.id, + title: "Continue the delegated task", + lineage: { + rootThreadId: parentThreadId, + parentThreadId, + relationshipToParent: "subagent", + }, + }), + ]; + + const groups = buildGroups([project], threads); + + expect(groups).toHaveLength(1); + expect(groups[0]?.threads.map((thread) => thread.id).toSorted()).toEqual( + [parentThreadId, forkThreadId].toSorted(), + ); + }); + it("matches web repository, repository-path, and separate grouping modes", () => { const environmentId = EnvironmentId.make("environment-1"); const repositoryIdentity = { diff --git a/apps/mobile/src/features/home/homeThreadList.ts b/apps/mobile/src/features/home/homeThreadList.ts index cada956d8bb..46656500fe3 100644 --- a/apps/mobile/src/features/home/homeThreadList.ts +++ b/apps/mobile/src/features/home/homeThreadList.ts @@ -165,6 +165,9 @@ export function buildHomeThreadGroups(input: { if (thread.archivedAt !== null) { continue; } + if (thread.lineage.relationshipToParent === "subagent") { + continue; + } if (input.environmentId !== null && thread.environmentId !== input.environmentId) { continue; } diff --git a/apps/mobile/src/features/threads/ThreadRelationshipsBanner.tsx b/apps/mobile/src/features/threads/ThreadRelationshipsBanner.tsx index 91091a1132a..153df6fd4fc 100644 --- a/apps/mobile/src/features/threads/ThreadRelationshipsBanner.tsx +++ b/apps/mobile/src/features/threads/ThreadRelationshipsBanner.tsx @@ -8,6 +8,7 @@ import { } from "@t3tools/client-runtime/state/thread-relationships"; import { canDetachThreadProviderSession } from "@t3tools/client-runtime/state/thread-workflows"; import type { EnvironmentId, OrchestrationV2ThreadShell, ThreadId } from "@t3tools/contracts"; +import { copySorted } from "@t3tools/shared/Array"; import { useNavigation } from "@react-navigation/native"; import { useMemo, useState } from "react"; import { ActivityIndicator, Modal, Pressable, ScrollView, View } from "react-native"; @@ -78,7 +79,8 @@ export function ThreadRelationshipsBanner(props: { const mergeTargetThreadId = resolveMergeBackTargetThreadId(projection); const rows = useMemo( () => - immediateThreadRelationships(graph, props.threadId).toSorted( + copySorted( + immediateThreadRelationships(graph, props.threadId), (left, right) => Number(right.threadId === mergeTargetThreadId) - Number(left.threadId === mergeTargetThreadId), diff --git a/apps/mobile/src/state/use-thread-selection.ts b/apps/mobile/src/state/use-thread-selection.ts index 524540ad83d..85e874b339c 100644 --- a/apps/mobile/src/state/use-thread-selection.ts +++ b/apps/mobile/src/state/use-thread-selection.ts @@ -13,6 +13,7 @@ import { type EnvironmentThreadShell, } from "@t3tools/client-runtime/state/shell"; import * as Option from "effect/Option"; +import { copySorted } from "@t3tools/shared/Array"; import { useProject, useThreadShell } from "../state/entities"; import { useEnvironmentThread } from "../state/threads"; @@ -56,7 +57,7 @@ function threadDetailToShell( projection: OrchestrationV2ThreadProjection, ): EnvironmentThreadShell { const thread = projection.thread; - const runsByOrdinal = projection.runs.toSorted((left, right) => right.ordinal - left.ordinal); + const runsByOrdinal = copySorted(projection.runs, (left, right) => right.ordinal - left.ordinal); const latestRun = runsByOrdinal[0] ?? null; const activeRun = runsByOrdinal.find( diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index be5950e427c..a4eb594e6d1 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -66,6 +66,11 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct readonly commands: OrchestrationV2Command[]; readonly projects: ProjectMutation[]; readonly launches?: OrchestrationV2ThreadLaunchInput[]; + readonly persistedUploads?: Array<{ + readonly threadId: string; + readonly messageId: string; + readonly attachments: ReadonlyArray<{ readonly name: string; readonly dataUrl: string }>; + }>; readonly projection?: OrchestrationV2ThreadProjection; }) { const client = { @@ -85,6 +90,33 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct resumed: false, }; }), + [WS_METHODS.assetsPersistChatAttachments]: (persistInput: { + readonly threadId: string; + readonly messageId: string; + readonly attachments: ReadonlyArray<{ + readonly type: "image"; + readonly name: string; + readonly mimeType: string; + readonly sizeBytes: number; + readonly dataUrl: string; + }>; + }) => + Effect.sync(() => { + input.persistedUploads?.push({ + threadId: persistInput.threadId, + messageId: persistInput.messageId, + attachments: persistInput.attachments.map(({ name, dataUrl }) => ({ name, dataUrl })), + }); + return { + attachments: persistInput.attachments.map((attachment, index) => ({ + type: "image" as const, + id: `server-attachment-${index}`, + name: attachment.name, + mimeType: attachment.mimeType, + sizeBytes: attachment.sizeBytes, + })), + }; + }), [WS_METHODS.projectsMutate]: (mutation: ProjectMutation) => Effect.sync(() => { input.projects.push(mutation); @@ -122,6 +154,83 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct }); describe("V2 environment commands", () => { + it.effect("remaps dual-tagged draft image uploads to server attachment ids", () => + Effect.gen(function* () { + const commands: OrchestrationV2Command[] = []; + const persistedUploads: Array<{ + readonly threadId: string; + readonly messageId: string; + readonly attachments: ReadonlyArray<{ readonly name: string; readonly dataUrl: string }>; + }> = []; + const supervisor = yield* makeSupervisor({ commands, projects: [], persistedUploads }); + + yield* startThreadTurn({ + commandId: CommandId.make("mobile-image-turn"), + threadId: v2ThreadId, + message: { + messageId: MessageId.make("message-mobile-image"), + role: "user", + text: "see this", + attachments: [ + { + type: "image", + id: "stored-before", + name: "stored-before.png", + mimeType: "image/png", + sizeBytes: 8, + }, + { + type: "image", + id: "local-preview-id", + name: "pasted-image.png", + mimeType: "image/png", + sizeBytes: 12, + dataUrl: "data:image/png;base64,AA==", + previewUri: "file:///tmp/pasted.png", + } as never, + { + type: "image", + name: "camera-image.png", + mimeType: "image/png", + sizeBytes: 16, + dataUrl: "data:image/png;base64,BB==", + }, + { + type: "image", + id: "stored-after", + name: "stored-after.png", + mimeType: "image/png", + sizeBytes: 20, + }, + ], + }, + runtimeMode: "full-access", + interactionMode: "default", + }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor)); + + expect(persistedUploads).toEqual([ + { + threadId: v2ThreadId, + messageId: "message-mobile-image", + attachments: [ + { name: "pasted-image.png", dataUrl: "data:image/png;base64,AA==" }, + { name: "camera-image.png", dataUrl: "data:image/png;base64,BB==" }, + ], + }, + ]); + expect(commands[0]).toMatchObject({ + type: "message.dispatch", + attachments: [ + { id: "stored-before" }, + { id: "server-attachment-0" }, + { id: "server-attachment-1" }, + { id: "stored-after" }, + ], + }); + expect(commands[0]).not.toMatchObject({ attachments: [{ id: "local-preview-id" }] }); + }).pipe(Effect.provide(TEST_CRYPTO_LAYER)), + ); + it.effect("routes projects through the event-sourced project transport", () => Effect.gen(function* () { const projects: ProjectMutation[] = []; diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index fda9980ac55..f6c69bcc4f3 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -184,31 +184,49 @@ const dispatch = (command: OrchestrationV2Command) => const getProjection = (threadId: ThreadId) => request(ORCHESTRATION_V2_WS_METHODS.getThreadProjection, { threadId }); +function isUploadChatAttachment( + attachment: ChatAttachment | UploadChatAttachment, +): attachment is UploadChatAttachment { + return "dataUrl" in attachment; +} + +function isStoredChatAttachment( + attachment: ChatAttachment | UploadChatAttachment, +): attachment is ChatAttachment { + return "id" in attachment && !("dataUrl" in attachment); +} + +function toUploadChatAttachmentPayload(attachment: UploadChatAttachment): UploadChatAttachment { + return { + type: "image", + name: attachment.name, + mimeType: attachment.mimeType, + sizeBytes: attachment.sizeBytes, + dataUrl: attachment.dataUrl, + }; +} + const persistAttachments = Effect.fn("EnvironmentCommands.persistAttachments")(function* ( threadId: ThreadId, messageId: MessageId, attachments: ReadonlyArray, ) { - const stored = attachments.filter( - (attachment): attachment is ChatAttachment => "id" in attachment, - ); - const uploads = attachments.filter( - (attachment): attachment is UploadChatAttachment => "dataUrl" in attachment, - ); - if (uploads.length === 0) return stored; + const uploads = attachments.filter(isUploadChatAttachment); + const storedOnly = attachments.filter(isStoredChatAttachment); + if (uploads.length === 0) return storedOnly; const result = yield* request(WS_METHODS.assetsPersistChatAttachments, { threadId, messageId, - attachments: uploads, + attachments: uploads.map(toUploadChatAttachmentPayload), }); - if (stored.length === 0) return result.attachments; - const byUpload = new Map( - uploads.map((attachment, index) => [attachment, result.attachments[index]]), - ); + let uploadIndex = 0; return attachments.flatMap((attachment) => { - if ("id" in attachment) return [attachment]; - const persisted = byUpload.get(attachment); - return persisted === undefined ? [] : [persisted]; + if (isUploadChatAttachment(attachment)) { + const persisted = result.attachments[uploadIndex]; + uploadIndex += 1; + return persisted === undefined ? [] : [persisted]; + } + return isStoredChatAttachment(attachment) ? [attachment] : []; }); }); diff --git a/packages/client-runtime/src/state/threadWorkflows.ts b/packages/client-runtime/src/state/threadWorkflows.ts index 4fbceeb8c2c..ae2830209f9 100644 --- a/packages/client-runtime/src/state/threadWorkflows.ts +++ b/packages/client-runtime/src/state/threadWorkflows.ts @@ -3,6 +3,7 @@ import type { OrchestrationV2ProviderCapabilities, OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; +import { copySorted } from "@t3tools/shared/Array"; type Projection = OrchestrationV2ThreadProjection; type Run = Projection["runs"][number]; @@ -54,19 +55,17 @@ export function deriveThreadQueueWorkflowState(projection: Projection): ThreadQu const activeRun = resolveActiveThreadRun(projection); const session = resolveThreadProviderSession(projection); const capabilities = session?.capabilities.turns; - const queuedRuns = projection.runs - .filter((run) => run.status === "queued") - .toSorted( - (left, right) => - (left.queuePosition ?? left.ordinal) - (right.queuePosition ?? right.ordinal) || - left.ordinal - right.ordinal, - ) - .map((run) => ({ - run, - text: - projection.messages.find((message) => message.id === run.userMessageId)?.text ?? - "Queued message", - })); + const queuedRuns = copySorted( + projection.runs.filter((run) => run.status === "queued"), + (left, right) => + (left.queuePosition ?? left.ordinal) - (right.queuePosition ?? right.ordinal) || + left.ordinal - right.ordinal, + ).map((run) => ({ + run, + text: + projection.messages.find((message) => message.id === run.userMessageId)?.text ?? + "Queued message", + })); return { activeRun, diff --git a/packages/shared/package.json b/packages/shared/package.json index 9107902566a..477caa8ab24 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -3,6 +3,10 @@ "private": true, "type": "module", "exports": { + "./Array": { + "types": "./src/Array.ts", + "import": "./src/Array.ts" + }, "./model": { "types": "./src/model.ts", "import": "./src/model.ts" diff --git a/packages/shared/src/Array.test.ts b/packages/shared/src/Array.test.ts new file mode 100644 index 00000000000..ab4440ca6e3 --- /dev/null +++ b/packages/shared/src/Array.test.ts @@ -0,0 +1,12 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { copySorted } from "./Array.ts"; + +describe("array copies", () => { + it("sorts a copy without mutating the source", () => { + const source = [3, 1, 2]; + + expect(copySorted(source, (left: number, right: number) => left - right)).toEqual([1, 2, 3]); + expect(source).toEqual([3, 1, 2]); + }); +}); diff --git a/packages/shared/src/Array.ts b/packages/shared/src/Array.ts new file mode 100644 index 00000000000..dea902cfc47 --- /dev/null +++ b/packages/shared/src/Array.ts @@ -0,0 +1,6 @@ +export function copySorted( + values: ReadonlyArray, + compareFn?: (left: T, right: T) => number, +): T[] { + return [...values].sort(compareFn); +} diff --git a/packages/shared/src/model.ts b/packages/shared/src/model.ts index 2345a98f269..8c224176bec 100644 --- a/packages/shared/src/model.ts +++ b/packages/shared/src/model.ts @@ -9,6 +9,7 @@ import { type ProviderOptionDescriptor, type ProviderOptionSelection, } from "@t3tools/contracts"; +import { copySorted } from "./Array.ts"; const DEFAULT_PROVIDER_DRIVER_KIND = ProviderDriverKind.make("codex"); @@ -80,12 +81,21 @@ export function getModelSelectionBooleanOptionValue( function canonicalModelSelectionOptions( modelSelection: ModelSelection, ): ReadonlyArray { - return (modelSelection.options ?? []) - .map((selection) => [selection.id, selection.value] as const) - .toSorted(([leftId, leftValue], [rightId, rightValue]) => { + return copySorted( + (modelSelection.options ?? []).map( + (selection): readonly [id: string, value: string | boolean] => [ + selection.id, + selection.value, + ], + ), + ( + [leftId, leftValue]: readonly [id: string, value: string | boolean], + [rightId, rightValue]: readonly [id: string, value: string | boolean], + ) => { const idOrder = leftId.localeCompare(rightId); return idOrder !== 0 ? idOrder : String(leftValue).localeCompare(String(rightValue)); - }); + }, + ); } /** From a5467d3af11e71da411e281a73749dc87db0f813 Mon Sep 17 00:00:00 2001 From: Mike Olson Date: Sun, 12 Jul 2026 21:27:41 -0400 Subject: [PATCH 2/2] fix(client-runtime): Heal V2 shell archive membership on reconnect Port the ios-fixes-main shell reconnect heal to orchestration V2: always HTTP-heal (or force a full socket snapshot) before afterSequence resume, reject stale full snapshots, and normalize active/archive membership from archivedAt so dropped archive deltas cannot stick threads on the home list. --- .../src/state/shell-sync.test.ts | 209 +++++++++++++++++- packages/client-runtime/src/state/shell.ts | 116 +++++++--- .../src/state/shellReducer.test.ts | 31 ++- .../client-runtime/src/state/shellReducer.ts | 63 +++++- 4 files changed, 373 insertions(+), 46 deletions(-) diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index fadf4736fb7..be77b7d13c8 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -76,7 +76,7 @@ describe("environment shell synchronization", () => { } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); const cache = Persistence.EnvironmentCacheStore.of({ loadShell: () => Effect.succeed(Option.none()), - saveShell: () => Effect.never, + saveShell: () => Effect.void, loadThread: () => Effect.succeed(Option.none()), saveThread: () => Effect.void, removeThread: () => Effect.void, @@ -132,12 +132,25 @@ describe("environment shell synchronization", () => { }), ); - it.effect("resumes a warm shell cache via afterSequence without an HTTP fetch", () => + it.effect("heals warm cache membership from HTTP before resuming afterSequence", () => Effect.gen(function* () { const cachedSnapshot: OrchestrationV2ShellSnapshot = { ...v2ShellSnapshot, snapshotSequence: 5, }; + // Server already archived the thread, but a high-sequence warm cache still + // lists it as active because the archive delta was dropped earlier. + const httpSnapshot: OrchestrationV2ShellSnapshot = { + ...v2ShellSnapshot, + snapshotSequence: 4, + threads: [], + archivedThreads: [ + { + ...v2ShellSnapshot.threads[0]!, + archivedAt: v2ShellSnapshot.threads[0]!.updatedAt, + }, + ], + }; const events = yield* Queue.unbounded(); const capturedAfterSequence = yield* SubscriptionRef.make(undefined); const loaderCalls = yield* SubscriptionRef.make(0); @@ -174,22 +187,204 @@ describe("environment shell synchronization", () => { }); const snapshotLoader = ShellSnapshotLoader.of({ load: () => - SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe(Effect.as(Option.none())), + SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe( + Effect.as(Option.some(httpSnapshot)), + ), }); - yield* makeEnvironmentShellState().pipe( + const shellState = yield* makeEnvironmentShellState().pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ShellSnapshotLoader, snapshotLoader), ); - // Wait until the subscription is established from the warm cache. yield* SubscriptionRef.changes(capturedAfterSequence).pipe( Stream.filter((value) => value !== undefined), Stream.runHead, ); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter( + (state) => Option.isSome(state.snapshot) && state.snapshot.value.threads.length === 0, + ), + Stream.runHead, + ); - expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(5); - expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0); + // HTTP heal is authoritative even when its sequence is behind the cache. + expect(yield* SubscriptionRef.get(loaderCalls)).toBe(1); + expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(4); + const snapshot = Option.getOrThrow((yield* SubscriptionRef.get(shellState)).snapshot); + expect(snapshot.threads).toEqual([]); + expect(snapshot.archivedThreads).toHaveLength(1); + }), + ); + + it.effect("forces a full socket snapshot when HTTP heal fails with a warm cache", () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationV2ShellSnapshot = { + ...v2ShellSnapshot, + snapshotSequence: 50, + }; + const events = yield* Queue.unbounded(); + const capturedAfterSequence = yield* SubscriptionRef.make( + "missing", + ); + const client = { + [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: { + readonly afterSequence?: number; + }) => + Stream.unwrap( + SubscriptionRef.set( + capturedAfterSequence, + input.afterSequence === undefined ? undefined : input.afterSequence, + ).pipe(Effect.as(Stream.fromQueue(events))), + ), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.some(cachedSnapshot)), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.succeed(Option.none()), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + yield* SubscriptionRef.changes(capturedAfterSequence).pipe( + Stream.filter((value) => value !== "missing"), + Stream.runHead, + ); + + // Cache alone must not drive afterSequence; socket should send a full snapshot. + expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBeUndefined(); + + // Server membership is correct but sequence is behind the warm cache. + // The socket heal must still apply (authoritative) or ghost threads remain. + const archivedThread = { + ...cachedSnapshot.threads[0]!, + archivedAt: cachedSnapshot.threads[0]!.updatedAt, + }; + yield* Queue.offer(events, { + kind: "snapshot", + snapshot: { + ...cachedSnapshot, + snapshotSequence: 40, + threads: [], + archivedThreads: [archivedThread], + }, + }); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter( + (state) => Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === 40, + ), + Stream.runHead, + ); + const snapshot = Option.getOrThrow((yield* SubscriptionRef.get(shellState)).snapshot); + expect(snapshot.threads).toEqual([]); + expect(snapshot.archivedThreads).toHaveLength(1); + }), + ); + + it.effect("rejects a stale full snapshot after a newer archive delta", () => + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const client = { + [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.none()), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.succeed(Option.none()), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + const liveSnapshot: OrchestrationV2ShellSnapshot = { + ...LIVE_SHELL_SNAPSHOT, + snapshotSequence: 2, + }; + yield* Queue.offer(events, { kind: "snapshot", snapshot: liveSnapshot }); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((state) => state.status === "live"), + Stream.runHead, + ); + + const archivedThread = { + ...liveSnapshot.threads[0]!, + archivedAt: liveSnapshot.threads[0]!.updatedAt, + }; + yield* Queue.offer(events, { + kind: "thread.updated", + sequence: 3, + location: "archive", + thread: archivedThread, + }); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter( + (state) => Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === 3, + ), + Stream.runHead, + ); + + // Older full snapshot that still lists the thread as active must not win. + yield* Queue.offer(events, { + kind: "snapshot", + snapshot: { + ...liveSnapshot, + snapshotSequence: 2, + threads: liveSnapshot.threads, + archivedThreads: [], + }, + }); + for (let index = 0; index < 10; index += 1) { + yield* Effect.yieldNow; + } + + const state = yield* SubscriptionRef.get(shellState); + const snapshot = Option.getOrThrow(state.snapshot); + expect(snapshot.snapshotSequence).toBe(3); + expect(snapshot.threads).toEqual([]); + expect(snapshot.archivedThreads.map((thread) => thread.id)).toEqual([archivedThread.id]); }), ); }); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index aed33f9e500..d185506df5e 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -10,6 +10,7 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { AsyncResult, Atom } from "effect/unstable/reactivity"; @@ -21,7 +22,7 @@ import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; import { subscribe } from "../rpc/client.ts"; import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts"; -import { applyShellStreamEvent } from "./shellReducer.ts"; +import { applyShellStreamEvent, normalizeShellThreadMembership } from "./shellReducer.ts"; import type { EnvironmentCatalogState } from "./connections.ts"; import { followStreamInEnvironment } from "./runtime.ts"; @@ -68,6 +69,10 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") status: shellStatusForSnapshot(cachedSnapshot), error: Option.none(), }); + // When HTTP heal fails we subscribe without afterSequence so the server + // embeds a full snapshot. That first snapshot must apply even if its sequence + // is behind the warm disk cache (authoritative server membership). + const acceptNextSocketSnapshotAuthoritatively = yield* Ref.make(false); const persistence = yield* Queue.sliding(1); const persist = Effect.fn("EnvironmentShellState.persist")(function* ( @@ -126,11 +131,21 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* ( item: OrchestrationV2ShellStreamItem, + options?: { readonly authoritative?: boolean }, ) { const current = yield* SubscriptionRef.get(state); const nextSnapshot = item.kind === "snapshot" - ? item.snapshot + ? Option.match(current.snapshot, { + // Reject older full snapshots from the live stream so a slow + // enrichment refresh cannot reintroduce archived threads. HTTP + // reconnect heals use authoritative:true and always apply. + onSome: (snapshot) => + !options?.authoritative && item.snapshot.snapshotSequence < snapshot.snapshotSequence + ? null + : normalizeShellThreadMembership(item.snapshot), + onNone: () => normalizeShellThreadMembership(item.snapshot), + }) : Option.match(current.snapshot, { onNone: () => null, onSome: (snapshot) => @@ -152,41 +167,74 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* Effect.forkScoped( Effect.gen(function* () { - // Establish the base shell snapshot to resume from, minimizing bytes over - // the wire: - // - Warm cache: reuse the cached snapshot (zero network) and resume via - // `afterSequence` so we only receive shell events since the cached - // sequence. - // - Cold cache: load the full shell snapshot over HTTP (gzip-compressible, - // and off the socket), then resume via `afterSequence`. - // If no base can be established we fall back to the socket-embedded - // snapshot so the shell still synchronizes. Overlapping/replayed events are - // deduped by sequence in applyItem. - const base = Option.isSome(cachedSnapshot) - ? cachedSnapshot - : yield* Effect.gen(function* () { - const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe( - Stream.filter(Option.isSome), - Stream.map((current) => current.value), - Stream.runHead, - ); - return Option.isSome(prepared) - ? yield* snapshotLoader.load(prepared.value) - : Option.none(); - }); - - if (Option.isSome(base)) { - yield* applyItem({ kind: "snapshot", snapshot: base.value }); + // Paint immediately from disk cache when available, then always heal from + // the HTTP shell snapshot before (and on every) live session. + // + // Why HTTP heal is required even with a warm cache: + // If an archive delta is dropped, later unrelated shell events still + // advance snapshotSequence. Resuming only via afterSequence then skips + // the archive forever and keeps the thread on the home list. + if (Option.isSome(cachedSnapshot)) { + yield* applyItem({ kind: "snapshot", snapshot: cachedSnapshot.value }); } - const subscribeInput = Option.match(base, { - onNone: () => ({}), - onSome: (snapshot) => ({ afterSequence: snapshot.snapshotSequence }), - }); + yield* SubscriptionRef.changes(supervisor.session).pipe( + Stream.switchMap( + Option.match({ + onNone: () => Stream.empty, + onSome: () => + Stream.unwrap( + Effect.gen(function* () { + // Never resume afterSequence from disk cache alone. A dropped + // archive delta plus later events advances snapshotSequence + // past the archive, so delta-only resume keeps the thread on + // the home list forever. Require a server heal first: + // HTTP full snapshot, or a socket-embedded full snapshot. + let healedFromServer = false; + const prepared = yield* SubscriptionRef.get(supervisor.prepared); + if (Option.isSome(prepared)) { + const httpSnapshot = yield* snapshotLoader.load(prepared.value); + if (Option.isSome(httpSnapshot)) { + yield* applyItem( + { kind: "snapshot", snapshot: httpSnapshot.value }, + { authoritative: true }, + ); + healedFromServer = true; + // Clear any leftover flag from a prior session that + // failed HTTP heal and disconnected before its socket + // snapshot arrived. + yield* Ref.set(acceptNextSocketSnapshotAuthoritatively, false); + } + } - yield* subscribe(ORCHESTRATION_V2_WS_METHODS.subscribeShell, subscribeInput, { - onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), - }).pipe(Stream.runForEach(applyItem)); + const live = yield* SubscriptionRef.get(state); + const subscribeInput = + healedFromServer && Option.isSome(live.snapshot) + ? { afterSequence: live.snapshot.value.snapshotSequence } + : {}; + if (!healedFromServer) { + yield* Ref.set(acceptNextSocketSnapshotAuthoritatively, true); + } + return subscribe(ORCHESTRATION_V2_WS_METHODS.subscribeShell, subscribeInput, { + onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), + }); + }), + ), + }), + ), + Stream.runForEach((item) => + Effect.gen(function* () { + if (item.kind === "snapshot") { + const acceptAuthoritative = yield* Ref.get(acceptNextSocketSnapshotAuthoritatively); + if (acceptAuthoritative) { + yield* Ref.set(acceptNextSocketSnapshotAuthoritatively, false); + return yield* applyItem(item, { authoritative: true }); + } + } + return yield* applyItem(item); + }), + ), + ); }), ); yield* SubscriptionRef.changes(supervisor.state).pipe( diff --git a/packages/client-runtime/src/state/shellReducer.test.ts b/packages/client-runtime/src/state/shellReducer.test.ts index f588f17347d..7406cf9e8e0 100644 --- a/packages/client-runtime/src/state/shellReducer.test.ts +++ b/packages/client-runtime/src/state/shellReducer.test.ts @@ -2,7 +2,7 @@ import { ProjectId, ThreadId } from "@t3tools/contracts"; import { describe, expect, it } from "vite-plus/test"; import { v2Project, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts"; -import { applyShellStreamEvent } from "./shellReducer.ts"; +import { applyShellStreamEvent, normalizeShellThreadMembership } from "./shellReducer.ts"; describe("applyShellStreamEvent", () => { it("ignores stale project updates without mutating the snapshot", () => { @@ -62,6 +62,18 @@ describe("applyShellStreamEvent", () => { expect(active.archivedThreads).toEqual([]); }); + it("forces archive membership when archivedAt is set even if location is active", () => { + const next = applyShellStreamEvent(v2ShellSnapshot, { + kind: "thread.updated", + sequence: 3, + location: "active", + thread: { ...v2ThreadShell, archivedAt: v2ThreadShell.updatedAt }, + }); + expect(next.threads).toEqual([]); + expect(next.archivedThreads).toHaveLength(1); + expect(next.archivedThreads[0]?.id).toBe(v2ThreadShell.id); + }); + it("removes a thread from either collection", () => { const next = applyShellStreamEvent(v2ShellSnapshot, { kind: "thread.removed", @@ -82,3 +94,20 @@ describe("applyShellStreamEvent", () => { expect(next).toBe(v2ShellSnapshot); }); }); + +describe("normalizeShellThreadMembership", () => { + it("moves archived threads out of the active list", () => { + const archivedThread = { ...v2ThreadShell, archivedAt: v2ThreadShell.updatedAt }; + const next = normalizeShellThreadMembership({ + ...v2ShellSnapshot, + threads: [archivedThread], + archivedThreads: [], + }); + expect(next.threads).toEqual([]); + expect(next.archivedThreads).toEqual([archivedThread]); + }); + + it("returns the same reference when membership is already consistent", () => { + expect(normalizeShellThreadMembership(v2ShellSnapshot)).toBe(v2ShellSnapshot); + }); +}); diff --git a/packages/client-runtime/src/state/shellReducer.ts b/packages/client-runtime/src/state/shellReducer.ts index 3e18a9ed895..59a505b9293 100644 --- a/packages/client-runtime/src/state/shellReducer.ts +++ b/packages/client-runtime/src/state/shellReducer.ts @@ -1,6 +1,7 @@ import type { OrchestrationV2ShellSnapshot, OrchestrationV2ShellStreamItem, + OrchestrationV2ThreadShell, } from "@t3tools/contracts"; function upsertById( @@ -12,6 +13,57 @@ function upsertById( return items.map((candidate, candidateIndex) => (candidateIndex === index ? item : candidate)); } +/** + * Re-partitions active/archive membership from each thread's `archivedAt`. + * Defends against stale full snapshots or mis-tagged deltas that would otherwise + * keep an archived thread on the home list. + */ +export function normalizeShellThreadMembership( + snapshot: OrchestrationV2ShellSnapshot, +): OrchestrationV2ShellSnapshot { + const byId = new Map(); + + for (const thread of snapshot.threads) { + byId.set(String(thread.id), thread); + } + for (const thread of snapshot.archivedThreads) { + const existing = byId.get(String(thread.id)); + if (existing === undefined) { + byId.set(String(thread.id), thread); + continue; + } + // Prefer a non-null archivedAt when the same id appears in both lists. + if (existing.archivedAt === null && thread.archivedAt !== null) { + byId.set(String(thread.id), thread); + } + } + + const threads: OrchestrationV2ThreadShell[] = []; + const archivedThreads: OrchestrationV2ThreadShell[] = []; + for (const thread of byId.values()) { + if (thread.archivedAt !== null) { + archivedThreads.push(thread); + } else { + threads.push(thread); + } + } + + if ( + threads.length === snapshot.threads.length && + archivedThreads.length === snapshot.archivedThreads.length && + threads.every((thread, index) => thread === snapshot.threads[index]) && + archivedThreads.every((thread, index) => thread === snapshot.archivedThreads[index]) + ) { + return snapshot; + } + + return { + ...snapshot, + threads, + archivedThreads, + }; +} + /** Applies one committed V2 shell delta while preserving active/archive exclusivity. */ export function applyShellStreamEvent( snapshot: OrchestrationV2ShellSnapshot, @@ -33,20 +85,23 @@ export function applyShellStreamEvent( snapshotSequence: event.sequence, }; case "thread.updated": { + // Trust archivedAt over location so a mis-tagged active delta cannot keep + // an archived thread on the home list after another client archives it. + const location = event.thread.archivedAt !== null ? "archive" : event.location; const withoutThread = (threads: OrchestrationV2ShellSnapshot["threads"]) => threads.filter((thread) => thread.id !== event.thread.id); - return { + return normalizeShellThreadMembership({ ...snapshot, threads: - event.location === "active" + location === "active" ? upsertById(withoutThread(snapshot.threads), event.thread) : withoutThread(snapshot.threads), archivedThreads: - event.location === "archive" + location === "archive" ? upsertById(withoutThread(snapshot.archivedThreads), event.thread) : withoutThread(snapshot.archivedThreads), snapshotSequence: event.sequence, - }; + }); } case "thread.removed": return {