diff --git a/apps/mobile/src/connection/environment-cache-store.ts b/apps/mobile/src/connection/environment-cache-store.ts index 3be6c3d302801..5b3191d1c996d 100644 --- a/apps/mobile/src/connection/environment-cache-store.ts +++ b/apps/mobile/src/connection/environment-cache-store.ts @@ -111,9 +111,9 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { select: (stored) => stored.environmentId === environmentId ? Option.some( - stored.origin === undefined + stored.originV2 === undefined ? stored.snapshot - : { ...stored.snapshot, origin: stored.origin }, + : { ...stored.snapshot, origin: stored.originV2 }, ) : Option.none(), }).pipe( @@ -130,7 +130,7 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, environmentId, snapshot, - ...(origin === undefined ? {} : { origin }), + ...(origin === undefined ? {} : { originV2: origin }), }).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause))); yield* database .saveCache( diff --git a/apps/mobile/src/features/layout/AdaptiveWorkspaceLayout.tsx b/apps/mobile/src/features/layout/AdaptiveWorkspaceLayout.tsx index 19ffe072fb58e..4031d014ce246 100644 --- a/apps/mobile/src/features/layout/AdaptiveWorkspaceLayout.tsx +++ b/apps/mobile/src/features/layout/AdaptiveWorkspaceLayout.tsx @@ -106,9 +106,11 @@ import { useAndroidWorkspaceRoot } from "./workspace/AndroidWorkspaceTabsProvide import { dispatchRootStackReset, rootStackState } from "../../lib/root-stack-reset"; import { currentWorkspaceSelections, + markWorkspaceThreadReopened, workspaceSelectionCheck, } from "./workspace/workspace-selection-store"; import { + threadReopenedByTab, workspaceTabColumns, workspaceTabNavigationState, } from "./workspace/workspace-tab-navigation"; @@ -555,6 +557,11 @@ function AdaptiveWorkspaceLayoutContent( seedPullRequest: withKnownPullRequestSummary, }, ); + if (tab === "threads") { + markWorkspaceThreadReopened( + threadReopenedByTab(workspace, next.routes, currentWorkspaceSelections()), + ); + } if (layout.variant === "medium") { const opened = next.routes.at(-1 - overlays.length)?.key ?? null; setOverlaySidebar((current) => ({ ...current, visible: true, keepOpenFor: opened })); diff --git a/apps/mobile/src/features/layout/workspace/workspace-selection-store.ts b/apps/mobile/src/features/layout/workspace/workspace-selection-store.ts index 19d2653ddedfd..789b0acc80347 100644 --- a/apps/mobile/src/features/layout/workspace/workspace-selection-store.ts +++ b/apps/mobile/src/features/layout/workspace/workspace-selection-store.ts @@ -19,13 +19,12 @@ import { workspaceRoutesOf } from "../workspace-overlay-routes"; import type { WorkspaceSelections } from "../../../persistence/mobile-preferences"; import { appAtomRegistry } from "../../../state/atom-registry"; import { environmentPresentations } from "../../../state/presentation"; -import { - environmentThreadShells, - startOpeningEnvironmentThread, - threadEnvironment, -} from "../../../state/threads"; +import { environmentShell } from "../../../state/shell"; +import { environmentThreadShells, startOpeningEnvironmentThread } from "../../../state/threads"; import { rememberedSettingsPage, + rememberedThreadReopens, + threadHasPageOpened, workspaceSelectionsIn, type WorkspaceSelectionCheck, } from "./workspace-tab-navigation"; @@ -38,6 +37,9 @@ const REMEMBERS = Platform.OS === "ios"; let selections: WorkspaceSelections = {}; let saved = selections; let hydration: Promise | null = null; +// The chat a tab switch or a cold start reopened by itself, until the user opens something. +let reopened: WorkspaceSelections["threads"]; +let reopenings = 0; function save() { const writing = selections; @@ -88,7 +90,40 @@ export function currentWorkspaceSelections() { */ export function recordWorkspaceSelections(state: NavigationState | undefined) { if (!REMEMBERS || !state) return; - selections = workspaceSelectionsIn(state.routes.slice(0, state.index + 1), selections); + const routes = state.routes.slice(0, state.index + 1); + selections = workspaceSelectionsIn(routes, selections); + // Opening another chat, or anything from the reopened one, makes what is on screen the user's. + if ( + reopened && + (!sameThread(reopened, selections.threads) || threadHasPageOpened(routes, reopened)) + ) { + reopened = undefined; + } +} + +function sameThread( + left: { readonly environmentId: string; readonly threadId: string }, + right: { readonly environmentId: string; readonly threadId: string } | undefined, +) { + return left.environmentId === right?.environmentId && left.threadId === right.threadId; +} + +/** Marks the chat a tab switch reopened by itself; see `threadReopenedByTab`. */ +export function markWorkspaceThreadReopened(thread: WorkspaceSelections["threads"]) { + reopened = thread; + if (thread) reopenings += 1; +} + +/** + * The number of the reopening that put this chat on screen, or null when somebody opened it. + * Nobody asked for a reopened chat, so its screen closes it when it turns out archived or + * deleted, where a chat somebody opened shows what it has. iOS keeps a screen mounted across tab + * switches, so what the screen learned counts for one reopening only. + */ +export function workspaceThreadReopening(environmentId: EnvironmentId, threadId: ThreadId) { + return reopened !== undefined && sameThread(reopened, { environmentId, threadId }) + ? reopenings + : null; } function environmentRemoved(environmentId: string) { @@ -101,22 +136,31 @@ function environmentRemoved(environmentId: string) { ); } -/** - * A remembered chat is gone once its environment was removed, or once that environment's - * snapshot has loaded without it. Before it loads there is nothing to contradict it. - */ +/** Reads what `rememberedThreadReopens` decides from: the shell rows and the removals it applied. */ export const workspaceSelectionCheck: WorkspaceSelectionCheck = { thread: (selection) => { if (environmentRemoved(selection.environmentId)) return false; const environmentId = selection.environmentId as EnvironmentId; - if (!appAtomRegistry.get(threadEnvironment.snapshotAtom(environmentId))) return true; - return appAtomRegistry - .get(environmentThreadShells.environmentThreadIndexAtom(environmentId)) - .has(selection.threadId as ThreadId); + const threadId = selection.threadId as ThreadId; + return rememberedThreadReopens({ + listed: appAtomRegistry + .get(environmentThreadShells.environmentThreadIndexAtom(environmentId)) + .has(threadId), + removed: appAtomRegistry + .get(environmentShell.stateValueAtom(environmentId)) + .removedAt.has(threadId), + }); }, pullRequest: (selection) => !environmentRemoved(selection.environmentId), }; +/** Stops reopening a chat its own screen found archived or deleted. */ +export function forgetWorkspaceThread(environmentId: EnvironmentId, threadId: ThreadId) { + const { threads, ...rest } = selections; + if (threads && sameThread(threads, { environmentId, threadId })) selections = rest; + if (reopened && sameThread(reopened, { environmentId, threadId })) reopened = undefined; +} + /** Whether the window lays the workspace out in columns, by the rule the layout itself uses. */ export function workspaceHasColumns() { const metrics = currentNativeLayoutMetrics(); @@ -153,9 +197,9 @@ export async function launchWorkspaceURL(): Promise { } await metrics; if (!thread || !workspaceHasColumns() || !workspaceSelectionCheck.thread(thread)) return null; - return Linking.createURL( - `threads/${encodeURIComponent(thread.environmentId)}/${encodeURIComponent(thread.threadId)}`, - ); + markWorkspaceThreadReopened(thread); + // Not `createURL(path)`: it encodes the path again, and a subagent's id holds `%` and `:`. + return `${Linking.createURL("/")}threads/${encodeURIComponent(thread.environmentId)}/${encodeURIComponent(thread.threadId)}`; } /** Whether the workspace shows its first detail page, directly above its list. */ diff --git a/apps/mobile/src/features/layout/workspace/workspace-tab-navigation.test.ts b/apps/mobile/src/features/layout/workspace/workspace-tab-navigation.test.ts index 3f217fd70c8bc..eda3e5853401a 100644 --- a/apps/mobile/src/features/layout/workspace/workspace-tab-navigation.test.ts +++ b/apps/mobile/src/features/layout/workspace/workspace-tab-navigation.test.ts @@ -2,6 +2,9 @@ import { describe, expect, it } from "vite-plus/test"; import { restoreSettingsLinkState } from "../../settings/settings-navigation"; import { + rememberedThreadReopens, + threadHasPageOpened, + threadReopenedByTab, workspaceSelectionsIn, workspaceTabColumns, workspaceTabNavigationState, @@ -48,6 +51,42 @@ describe("workspace tab selections", () => { }); }); +describe("reopening the remembered chat", () => { + it("reopens a chat the shell left out, and not one the shell removed", () => { + // An idle subagent or an older settled chat: no row, and nothing removed it. + expect(rememberedThreadReopens({ listed: false, removed: false })).toBe(true); + expect(rememberedThreadReopens({ listed: true, removed: false })).toBe(true); + // Archived or deleted in this session. + expect(rememberedThreadReopens({ listed: false, removed: true })).toBe(false); + // Archived, then restored: the row is back. + expect(rememberedThreadReopens({ listed: true, removed: true })).toBe(true); + }); + + it("marks the chat a tab switch put on screen, and not one that was already there", () => { + const selections = { threads: thread.params }; + expect(threadReopenedByTab([home, pullRequest], [home, thread], selections)).toBe( + selections.threads, + ); + expect(threadReopenedByTab([home, thread], [home, thread], selections)).toBe(undefined); + expect(threadReopenedByTab([home, pullRequest], [home], selections)).toBe(undefined); + }); +}); + +describe("a reopened chat the user opened something from", () => { + it("counts a page on its route or a subagent beside it, and nothing else", () => { + const sheet = { key: "agents", name: "ThreadAgents" }; + const other = { key: "b", name: "Thread", params: { environmentId: "env", threadId: "b" } }; + const withSubagent = { ...thread, params: { ...thread.params, inspector: { type: "thread" } } }; + expect(threadHasPageOpened([home, thread], thread.params)).toBe(false); + expect(threadHasPageOpened([home, thread, sheet], thread.params)).toBe(true); + expect(threadHasPageOpened([home, thread, other], thread.params)).toBe(true); + expect(threadHasPageOpened([home, withSubagent], thread.params)).toBe(true); + // Not on screen: a tab switch away, or the launch state before the chat's route exists. + expect(threadHasPageOpened([home], thread.params)).toBe(false); + expect(threadHasPageOpened([home, pullRequest], thread.params)).toBe(false); + }); +}); + describe("opening a workspace tab", () => { const columns = { columns: true, check: available }; diff --git a/apps/mobile/src/features/layout/workspace/workspace-tab-navigation.ts b/apps/mobile/src/features/layout/workspace/workspace-tab-navigation.ts index 6a44e617e53ed..76791149d3568 100644 --- a/apps/mobile/src/features/layout/workspace/workspace-tab-navigation.ts +++ b/apps/mobile/src/features/layout/workspace/workspace-tab-navigation.ts @@ -38,6 +38,47 @@ export function rememberedSettingsPage(selections: WorkspaceSelections): Setting : DEFAULT_SETTINGS_PAGE; } +/** + * Whether a remembered chat is reopened. Shells leave idle subagents and older settled chats out, + * so a missing row does not say the chat is gone; a removal the shell applied does. A chat + * reopened without a row closes itself when its detail says it was archived or deleted. + */ +export function rememberedThreadReopens(input: { + readonly listed: boolean; + readonly removed: boolean; +}) { + return input.listed || !input.removed; +} + +function showsThread(route: RouteLike, thread: ThreadSelection) { + return ( + route.name === "Thread" && + stringParam(route.params, "environmentId") === thread.environmentId && + stringParam(route.params, "threadId") === thread.threadId + ); +} + +/** The remembered chat, when opening a tab put it on screen and it was not there before. */ +export function threadReopenedByTab( + previous: readonly RouteLike[], + next: readonly RouteLike[], + selections: WorkspaceSelections, +) { + const thread = selections.threads; + if (!thread || previous.some((route) => showsThread(route, thread))) return undefined; + return next.some((route) => showsThread(route, thread)) ? thread : undefined; +} + +/** + * Whether the user opened something from the chat: a page or a sheet on its route, or a subagent + * beside it, which lives in the route's `inspector` param. A chat that is not on screen says no. + */ +export function threadHasPageOpened(routes: readonly RouteLike[], thread: ThreadSelection) { + const index = routes.findLastIndex((route) => showsThread(route, thread)); + const params = routes[index]?.params as { readonly inspector?: unknown } | undefined; + return index >= 0 && (index < routes.length - 1 || params?.inspector !== undefined); +} + /** Whether a remembered chat or pull request can still be opened. */ export interface WorkspaceSelectionCheck { readonly thread: (selection: ThreadSelection) => boolean; @@ -167,12 +208,7 @@ function detailRoute( const thread = selections.threads; if (!thread || !check.thread(thread)) return undefined; return ( - state.routes.find( - (route) => - route.name === "Thread" && - stringParam(route.params, "environmentId") === thread.environmentId && - stringParam(route.params, "threadId") === thread.threadId, - ) ?? { + state.routes.find((route) => showsThread(route, thread)) ?? { key: thread.key, name: "Thread", params: { environmentId: thread.environmentId, threadId: thread.threadId }, diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index 6205fd1281797..f9cd3fadbd274 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -8,6 +8,7 @@ import { worktreeSetupAgentStarted } from "@t3tools/client-runtime/worktree-setu import { AppBar } from "../../components/ui/AppBar"; import { StackActions, + useIsFocused, useNavigation, useRoute, type StaticScreenProps, @@ -92,7 +93,7 @@ import { useSelectedThreadWorktree } from "../../state/use-selected-thread-workt import { useThreadComposerState } from "../../state/use-thread-composer-state"; import { resolveMergeBackTargetThreadId } from "@t3tools/client-runtime/state/thread-relationships"; import { resolveLatestMergeBackRun } from "@t3tools/client-runtime/state/thread-workflows"; -import { threadEnvironment } from "../../state/threads"; +import { threadEnvironment, useOpenThreadFamily } from "../../state/threads"; import { EMPTY_ENVIRONMENT_THREAD_STATE } from "@t3tools/client-runtime/state/threads"; import { useProjects } from "../../state/entities"; import { projectThreadContentPresentation } from "./threadContentPresentation"; @@ -105,6 +106,7 @@ import { ThreadFileNavigatorPane } from "../files/thread-file-navigator-pane"; import { InspectorSheetContext } from "../layout/workspace-inspector-sheet"; import type { InspectorContent } from "../layout/inspector-panes"; import { + reopenedThreadOutcome, threadDetailErrorIsCurrent, threadRouteIsHydrating, threadRouteOpensDraft, @@ -126,6 +128,10 @@ import { useStartSideChat } from "./useStartSideChat"; import { useThreadStatusAnnouncements } from "./useThreadStatusAnnouncements"; import type { ThreadInspectorMode, ThreadInspectorTarget } from "./thread-inspector-navigation"; import { useSelectWorkspaceDetail } from "../layout/workspace/useSelectWorkspaceDetail"; +import { + forgetWorkspaceThread, + workspaceThreadReopening, +} from "../layout/workspace/workspace-selection-store"; function firstRouteParam(value: string | string[] | undefined): string | null { const first = Array.isArray(value) ? value[0] : value; @@ -225,6 +231,26 @@ function UnavailableThreadDraftRedirect(props: { readonly environmentId: Environ return ; } +/** + * A chat the workspace reopened by itself and that cannot open: back to the list, for good. + * Render it only for the focused route: going back from any other route acts on the page in front. + */ +function ReopenedThreadClose(props: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; +}) { + const goBack = useWorkspaceBackNavigation("threads"); + const close = useEffectEvent(() => { + forgetWorkspaceThread(props.environmentId, props.threadId); + goBack(); + }); + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- leaves this route for the list on mount: the workspace reopened the chat by itself and it cannot open + useEffect(() => { + close(); + }, []); + return ; +} + export function ThreadRouteScreen(props: ThreadRouteScreenProps) { return ( @@ -244,7 +270,8 @@ function LiveThreadRouteController(props: ThreadRouteScreenProps) { useAndroidWorkspaceBack("threads"); const connectionsReady = useConnectionsReady(); const { connectionState } = useRemoteConnectionStatus(); - const { selectedThread } = useThreadSelection(); + const { selectedThread, selectedThreadListed, selectedThreadDetailRef } = useThreadSelection(); + const familyRead = useOpenThreadFamily(selectedThreadDetailRef); const params = props.route.params ?? {}; const environmentIdRaw = firstRouteParam(params.environmentId); const threadIdRaw = firstRouteParam(params.threadId); @@ -265,6 +292,26 @@ function LiveThreadRouteController(props: ThreadRouteScreenProps) { ? null : scopedThreadKey(selectedThread.environmentId, selectedThread.id); const selectedThreadDetailState = useSelectedThreadDetailState(); + const focused = useIsFocused(); + const reopening = + routeName === "Thread" && + !props.inspectorPresentation && + environmentId !== null && + threadIdRaw !== null + ? workspaceThreadReopening(environmentId, ThreadId.make(threadIdRaw)) + : null; + // A row of the synced shell proves a reopened chat exists; a cached row can predate its removal. + // Until that proof arrives, its detail can still close it. + const [provenReopening, setProvenReopening] = useState(null); + if ( + reopening !== null && + selectedThreadListed && + routeEnvironmentShellState.synced && + provenReopening !== reopening + ) { + setProvenReopening(reopening); + } + const reopenedUnproven = reopening !== null && provenReopening !== reopening; const detailError = selectedThreadDetailState.error; const [errorBeforeConnected, setErrorBeforeConnected] = useState(() => routeConnectionState === "connected" ? Option.none() : detailError, @@ -289,6 +336,29 @@ function LiveThreadRouteController(props: ThreadRouteScreenProps) { return withInspectorHeader(); } + const closeReopened = ( + + ); + const detailHasError = threadDetailErrorIsCurrent({ + connectionState: routeConnectionState, + error: detailError, + errorBeforeConnected, + }); + const reopened = reopenedUnproven + ? reopenedThreadOutcome({ + // A subagent opened beside the chat lives in this route's params, not on a route above. + alone: focused && params.inspector === undefined, + isLoadingConnections: !connectionsReady, + environmentRegistered: routeEnvironmentRuntime !== null, + connectionState: routeConnectionState, + detailStatus: selectedThreadDetailState.status, + detailHasError, + thread: selectedThreadKey === routeThreadKey ? selectedThread : null, + }) + : "keep"; + if (reopened === "close") return closeReopened; + if (reopened === "wait") return ; + if (selectedThread !== null && selectedThreadKey === routeThreadKey) { return withInspectorHeader( , @@ -305,11 +375,8 @@ function LiveThreadRouteController(props: ThreadRouteScreenProps) { shellStatus: routeEnvironmentShellState.status, shellHasError: routeEnvironmentShellState.hasError, detailStatus: selectedThreadDetailState.status, - detailHasError: threadDetailErrorIsCurrent({ - connectionState: routeConnectionState, - error: detailError, - errorBeforeConnected, - }), + detailHasError, + familyRead, }); if (stillHydrating) { @@ -324,7 +391,15 @@ function LiveThreadRouteController(props: ThreadRouteScreenProps) { detailStatus: selectedThreadDetailState.status, }) ) { - return ; + // Leaving or replacing a route acts on the page in front, so only the focused route does it. + // iOS keeps a screen mounted after its route is gone, and a route can sit under another page. + if (!focused) return ; + // A draft nobody asked for must not take the place of a chat that only failed to reopen. + return reopenedUnproven ? ( + closeReopened + ) : ( + + ); } return withInspectorHeader( diff --git a/apps/mobile/src/features/threads/thread-route-hydration.test.ts b/apps/mobile/src/features/threads/thread-route-hydration.test.ts index 9a7f1043609df..967cfaa8907fe 100644 --- a/apps/mobile/src/features/threads/thread-route-hydration.test.ts +++ b/apps/mobile/src/features/threads/thread-route-hydration.test.ts @@ -2,6 +2,7 @@ import * as Option from "effect/Option"; import { describe, expect, it } from "vite-plus/test"; import { + reopenedThreadOutcome, threadDetailErrorIsCurrent, threadRouteIsHydrating, threadRouteOpensDraft, @@ -16,9 +17,32 @@ const settled = { shellHasError: false, detailStatus: "live" as const, detailHasError: false, + familyRead: "settled" as const, }; describe("threadRouteIsHydrating", () => { + it("waits for the family read of a thread the shell left out when its detail fails", () => { + const failed = { ...settled, detailStatus: "empty" as const, detailHasError: true }; + expect(threadRouteIsHydrating({ ...failed, familyRead: "pending" })).toBe(true); + expect(threadRouteIsHydrating(failed)).toBe(false); + expect( + threadRouteIsHydrating({ ...failed, familyRead: "pending", connectionState: "reconnecting" }), + ).toBe(false); + expect( + threadRouteIsHydrating({ ...failed, familyRead: "pending", detailStatus: "deleted" }), + ).toBe(false); + }); + + it("keeps waiting when both reads failed: a failed read does not say the thread is gone", () => { + const failed = { ...settled, detailStatus: "empty" as const, detailHasError: true }; + expect(threadRouteIsHydrating({ ...failed, familyRead: "failed" })).toBe(true); + // Not found is an answer: a detail read that missed, or a family read without the row. + expect( + threadRouteIsHydrating({ ...failed, familyRead: "failed", detailStatus: "deleted" }), + ).toBe(false); + expect(threadRouteIsHydrating({ ...failed, familyRead: "settled" })).toBe(false); + }); + it("waits for shell and thread detail hydration", () => { expect(threadRouteIsHydrating({ ...settled, shellStatus: "synchronizing" })).toBe(true); expect(threadRouteIsHydrating({ ...settled, shellStatus: "empty" })).toBe(true); @@ -153,3 +177,74 @@ describe("threadDetailErrorIsCurrent", () => { ).toBe(true); }); }); + +describe("reopenedThreadOutcome", () => { + const open = { archivedAt: null, deletedAt: null }; + const archived = { ...open, archivedAt: "2026-01-01" }; + const synced = { + alone: true, + isLoadingConnections: false, + environmentRegistered: true, + connectionState: "connected" as const, + detailStatus: "live" as const, + detailHasError: false, + thread: open, + }; + + it("closes a reopened chat only when its detail says it was archived or deleted", () => { + expect(reopenedThreadOutcome({ ...synced, thread: archived })).toBe("close"); + expect(reopenedThreadOutcome({ ...synced, thread: { ...open, deletedAt: "2026-01-01" } })).toBe( + "close", + ); + expect(reopenedThreadOutcome({ ...synced, detailStatus: "deleted", thread: null })).toBe( + "close", + ); + expect(reopenedThreadOutcome(synced)).toBe("keep"); + for (const detailStatus of ["empty", "cached", "synchronizing", "live"] as const) { + expect(reopenedThreadOutcome({ ...synced, detailStatus, thread: null })).toBe("keep"); + } + }); + + it("keeps a chat still being created, whose first read can miss", () => { + expect(reopenedThreadOutcome({ ...synced, detailStatus: "deleted" })).toBe("keep"); + }); + + it("leaves the chat alone while another page is on it or in it", () => { + // A route above it, a sheet, or a subagent opened in its inspector: closing would remove those. + expect(reopenedThreadOutcome({ ...synced, alone: false, thread: archived })).toBe("keep"); + expect( + reopenedThreadOutcome({ ...synced, alone: false, detailStatus: "deleted", thread: null }), + ).toBe("keep"); + expect( + reopenedThreadOutcome({ + ...synced, + alone: false, + detailStatus: "cached", + connectionState: "offline", + thread: archived, + }), + ).toBe("keep"); + }); + + it("trusts cached archived detail only when no synced read is coming", () => { + // Another client can unarchive the chat while this app is closed. + const cached = { ...synced, detailStatus: "cached" as const, thread: archived }; + expect(reopenedThreadOutcome(cached)).toBe("wait"); + expect(reopenedThreadOutcome({ ...cached, detailStatus: "synchronizing" })).toBe("wait"); + expect(reopenedThreadOutcome({ ...cached, connectionState: "connecting" })).toBe("wait"); + expect( + reopenedThreadOutcome({ + ...cached, + isLoadingConnections: true, + environmentRegistered: false, + connectionState: "available", + }), + ).toBe("wait"); + // No endless loading screen: offline, a failed attempt or a failed read closes it. + for (const connectionState of ["offline", "reconnecting", "error", "available"] as const) { + expect(reopenedThreadOutcome({ ...cached, connectionState })).toBe("close"); + } + expect(reopenedThreadOutcome({ ...cached, detailHasError: true })).toBe("close"); + expect(reopenedThreadOutcome({ ...cached, environmentRegistered: false })).toBe("close"); + }); +}); diff --git a/apps/mobile/src/features/threads/thread-route-hydration.ts b/apps/mobile/src/features/threads/thread-route-hydration.ts index a6e3454d47659..aac342f3dce3c 100644 --- a/apps/mobile/src/features/threads/thread-route-hydration.ts +++ b/apps/mobile/src/features/threads/thread-route-hydration.ts @@ -1,6 +1,9 @@ import type { EnvironmentConnectionPhase } from "@t3tools/client-runtime/connection"; import type { EnvironmentShellStatus } from "@t3tools/client-runtime/state/shell"; -import type { EnvironmentThreadStatus } from "@t3tools/client-runtime/state/threads"; +import type { + EnvironmentThreadStatus, + ThreadFamilyRead, +} from "@t3tools/client-runtime/state/threads"; import * as Option from "effect/Option"; export function threadRouteIsHydrating(input: { @@ -12,10 +15,17 @@ export function threadRouteIsHydrating(input: { readonly shellHasError: boolean; readonly detailStatus: EnvironmentThreadStatus; readonly detailHasError: boolean; + /** The open thread's family read, which brings the row of a thread the shell left out. */ + readonly familyRead: ThreadFamilyRead; }): boolean { - if (input.detailStatus === "deleted" || input.shellHasError || input.detailHasError) { + if (input.detailStatus === "deleted" || input.shellHasError) { return false; } + if (input.detailHasError) { + // A failed read is no answer. Only a settled family read without the row says the thread is + // gone; until then both reads run again and the route waits. + return input.connectionState === "connected" && input.familyRead !== "settled"; + } if (input.isLoadingConnections) { return true; } @@ -73,3 +83,35 @@ export function threadDetailErrorIsCurrent(input: { !(input.connectionState === "connected" && input.error === input.errorBeforeConnected) ); } + +/** + * What the screen of a chat the workspace reopened by itself does while no shell row proves the + * chat. Only its detail can say it was archived or deleted, and cached detail can predate an + * unarchive on another client, so that closes the chat only when no synced read is coming. A chat + * with a page the user opened on it or in it is left alone: closing would take that page along. + */ +export function reopenedThreadOutcome(input: { + readonly alone: boolean; + readonly isLoadingConnections: boolean; + readonly environmentRegistered: boolean; + readonly connectionState: EnvironmentConnectionPhase; + readonly detailStatus: EnvironmentThreadStatus; + readonly detailHasError: boolean; + readonly thread: { + readonly archivedAt: string | null; + readonly deletedAt: string | null; + } | null; +}): "keep" | "wait" | "close" { + const gone = + input.thread === null + ? input.detailStatus === "deleted" + : input.thread.archivedAt !== null || input.thread.deletedAt !== null; + if (!gone || !input.alone) return "keep"; + const synced = input.thread === null || input.detailStatus === "live"; + const syncComing = + !input.detailHasError && + (input.isLoadingConnections || + (input.environmentRegistered && + (input.connectionState === "connecting" || input.connectionState === "connected"))); + return !synced && syncComing ? "wait" : "close"; +} diff --git a/apps/mobile/src/state/shell.ts b/apps/mobile/src/state/shell.ts index 8f92c7c1a42c7..71ccf3c45f00b 100644 --- a/apps/mobile/src/state/shell.ts +++ b/apps/mobile/src/state/shell.ts @@ -27,6 +27,8 @@ const shellStatus = (state: EnvironmentShellState) => state.status; const shellHasError = (state: EnvironmentShellState) => Option.isSome(state.error); // The empty placeholder stands until the environment has read its cached shell. const shellCacheRead = (state: EnvironmentShellState) => state !== EMPTY_SHELL_STATE; +// Generation 0 is the cached shell: its rows can predate what happened while the app was closed. +const shellSynced = (state: EnvironmentShellState) => state.generation !== 0; const shellCacheReadAtom = Atom.family((environmentId: EnvironmentId) => Atom.make((get) => shellCacheRead(get(environmentShell.stateValueAtom(environmentId)))), @@ -56,5 +58,6 @@ export function useEnvironmentShellReadiness(environmentId: EnvironmentId | null status: useAtomValue(atom, shellStatus), hasError: useAtomValue(atom, shellHasError), cacheRead: useAtomValue(atom, shellCacheRead), + synced: useAtomValue(atom, shellSynced), }; } diff --git a/apps/mobile/src/state/thread-detail-shell.test.ts b/apps/mobile/src/state/thread-detail-shell.test.ts new file mode 100644 index 0000000000000..d9bb2efff028f --- /dev/null +++ b/apps/mobile/src/state/thread-detail-shell.test.ts @@ -0,0 +1,23 @@ +import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { v2Projection } from "../../../../packages/client-runtime/src/state/orchestrationV2TestFixtures"; +import { threadDetailToShell } from "./thread-detail-shell"; + +describe("threadDetailToShell", () => { + it("keeps where a delegated task came from, which the thread header names", () => { + const delegatedFrom = { + environmentId: EnvironmentId.make("environment-linked"), + threadId: ThreadId.make("thread-parent"), + title: "Parent thread", + }; + const linkOrigin = { sessionId: "session-1", label: "T3 Code", access: "full-access" as const }; + const shell = threadDetailToShell(EnvironmentId.make("environment-1"), { + ...v2Projection, + thread: { ...v2Projection.thread, delegatedFrom, linkOrigin }, + }); + + expect(shell.delegatedFrom).toEqual(delegatedFrom); + expect(shell.linkOrigin).toEqual(linkOrigin); + }); +}); diff --git a/apps/mobile/src/state/thread-detail-shell.ts b/apps/mobile/src/state/thread-detail-shell.ts new file mode 100644 index 0000000000000..99e02b13a4765 --- /dev/null +++ b/apps/mobile/src/state/thread-detail-shell.ts @@ -0,0 +1,84 @@ +import type { + EnvironmentId, + OrchestrationV2ThreadProjection, + OrchestrationV2ThreadShell, +} from "@t3tools/contracts"; +import { + presentThreadShell, + type EnvironmentThreadShell, +} from "@t3tools/client-runtime/state/shell"; +import { + deriveLatestThreadRun, + deriveThreadRuntime, +} from "@t3tools/client-runtime/state/thread-execution"; + +function latestUserMessageAt( + projection: OrchestrationV2ThreadProjection, +): OrchestrationV2ThreadShell["latestUserMessageAt"] { + for (let index = projection.messages.length - 1; index >= 0; index -= 1) { + const message = projection.messages[index]; + if (message?.role === "user") { + return message.createdAt; + } + } + + return null; +} + +/** The shell row of a thread the shell left out, from its loaded detail. */ +export function threadDetailToShell( + environmentId: EnvironmentId, + projection: OrchestrationV2ThreadProjection, +): EnvironmentThreadShell { + const thread = projection.thread; + const latestRun = deriveLatestThreadRun(projection); + const runtime = deriveThreadRuntime(projection); + const pendingRequest = + projection.runtimeRequests.find((request) => request.status === "pending") ?? null; + return presentThreadShell(environmentId, { + id: thread.id, + projectId: thread.projectId, + title: thread.title, + providerInstanceId: thread.providerInstanceId, + modelSelection: thread.modelSelection, + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + branch: thread.branch, + worktreePath: thread.worktreePath, + linkedPullRequest: thread.linkedPullRequest ?? null, + pullRequests: thread.pullRequests, + branchPullRequest: thread.branchPullRequest ?? null, + activeProviderThreadId: thread.activeProviderThreadId, + lineage: thread.lineage, + linkOrigin: thread.linkOrigin, + delegatedFrom: thread.delegatedFrom, + forkedFrom: thread.forkedFrom, + createdBy: thread.createdBy, + creationSource: thread.creationSource, + latestRunId: latestRun?.runId ?? null, + activeRunId: runtime?.activeRunId ?? null, + status: runtime?.status ?? "idle", + pendingRuntimeRequest: + pendingRequest === null + ? null + : { id: pendingRequest.id, kind: pendingRequest.kind, createdAt: pendingRequest.createdAt }, + latestVisibleMessage: null, + latestUserMessageAt: latestUserMessageAt(projection), + hasActionableProposedPlan: false, + itemCount: projection.turnItems.length, + visibleItemCount: projection.visibleTurnItems.length, + createdAt: thread.createdAt, + updatedAt: thread.updatedAt, + archivedAt: thread.archivedAt, + settledOverride: thread.settledOverride, + settledAt: thread.settledAt, + unsettledAt: thread.unsettledAt, + activeOrderKey: thread.activeOrderKey, + autoSettleDisabledAt: thread.autoSettleDisabledAt, + pinnedAt: thread.pinnedAt, + pinOrderKey: thread.pinOrderKey, + snoozedUntil: thread.snoozedUntil ?? null, + snoozedAt: thread.snoozedAt ?? null, + deletedAt: thread.deletedAt, + }); +} diff --git a/apps/mobile/src/state/threads.ts b/apps/mobile/src/state/threads.ts index bf2013aef5398..ae27db6b98bed 100644 --- a/apps/mobile/src/state/threads.ts +++ b/apps/mobile/src/state/threads.ts @@ -9,10 +9,12 @@ import { createOlderSettledThreadAtoms, createThreadEnvironmentAtoms, createThreadFamilyAtoms, + type ThreadFamilyRead, } from "@t3tools/client-runtime/state/threads"; import { createSteadyShellAtoms } from "@t3tools/client-runtime/state/shell"; -import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import type { EnvironmentId, ScopedThreadRef, ThreadId } from "@t3tools/contracts"; import { AsyncResult, Atom } from "effect/reactivity"; +import { useEffect } from "react"; import { environmentCatalog } from "../connection/catalog"; import { connectionAtomRuntime } from "../connection/runtime"; @@ -111,3 +113,31 @@ export function useEnvironmentThread( ) as EnvironmentThreadState; return state; } + +const NO_OPEN_FAMILY = Atom.make("settled"); + +/** + * Keeps the open thread's family in the shell rows until its screen leaves. Shells leave idle + * subagents out, so the rows of the thread, of its parents and of its subagents may exist only + * there. Pass the ref the detail is read by: a thread still being created has no family to read. + * Returns that family's read, and reads it again while it fails. + */ +export function useOpenThreadFamily(ref: ScopedThreadRef | null): ThreadFamilyRead { + const environmentId = ref?.environmentId ?? null; + const threadId = ref?.threadId ?? null; + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- holds the open thread's family in the thread family registry until the screen leaves it + useEffect(() => { + if (environmentId === null || threadId === null) return; + const release = threadFamilies.mount(appAtomRegistry, environmentId, threadId); + const stopRetry = threadFamilies.retryWhileFailed(appAtomRegistry, environmentId, threadId); + return () => { + stopRetry(); + release(); + }; + }, [environmentId, threadId]); + return useAtomValue( + environmentId === null || threadId === null + ? NO_OPEN_FAMILY + : threadFamilies.mountedReadAtom(environmentId, threadId), + ); +} diff --git a/apps/mobile/src/state/use-thread-selection.ts b/apps/mobile/src/state/use-thread-selection.ts index 35598d74bc36f..8ac98a57cb579 100644 --- a/apps/mobile/src/state/use-thread-selection.ts +++ b/apps/mobile/src/state/use-thread-selection.ts @@ -4,23 +4,14 @@ import { useState } from "react"; import { EnvironmentId, ThreadId, - type OrchestrationV2ThreadProjection, - type OrchestrationV2ThreadShell, type ScopedProjectRef, type ScopedThreadRef, } from "@t3tools/contracts"; -import { - presentThreadShell, - type EnvironmentThreadShell, -} from "@t3tools/client-runtime/state/shell"; -import { - deriveLatestThreadRun, - deriveThreadRuntime, -} from "@t3tools/client-runtime/state/thread-execution"; import * as Option from "effect/Option"; import { scopedThreadKey } from "../lib/scopedEntities"; import { useProject, useQuietThreadShell } from "./entities"; +import { threadDetailToShell } from "./thread-detail-shell"; import { useEnvironmentThread } from "./threads"; import { resolvePendingThreadCreation, @@ -44,74 +35,6 @@ function firstRouteParam(value: string | string[] | undefined): string | null { return first === undefined || first.trim().length === 0 ? null : first; } -function latestUserMessageAt( - projection: OrchestrationV2ThreadProjection, -): OrchestrationV2ThreadShell["latestUserMessageAt"] { - for (let index = projection.messages.length - 1; index >= 0; index -= 1) { - const message = projection.messages[index]; - if (message?.role === "user") { - return message.createdAt; - } - } - - return null; -} - -function threadDetailToShell( - environmentId: EnvironmentId, - projection: OrchestrationV2ThreadProjection, -): EnvironmentThreadShell { - const thread = projection.thread; - const latestRun = deriveLatestThreadRun(projection); - const runtime = deriveThreadRuntime(projection); - const pendingRequest = - projection.runtimeRequests.find((request) => request.status === "pending") ?? null; - return presentThreadShell(environmentId, { - id: thread.id, - projectId: thread.projectId, - title: thread.title, - providerInstanceId: thread.providerInstanceId, - modelSelection: thread.modelSelection, - runtimeMode: thread.runtimeMode, - interactionMode: thread.interactionMode, - branch: thread.branch, - worktreePath: thread.worktreePath, - linkedPullRequest: thread.linkedPullRequest ?? null, - pullRequests: thread.pullRequests, - branchPullRequest: thread.branchPullRequest ?? null, - activeProviderThreadId: thread.activeProviderThreadId, - lineage: thread.lineage, - forkedFrom: thread.forkedFrom, - createdBy: thread.createdBy, - creationSource: thread.creationSource, - latestRunId: latestRun?.runId ?? null, - activeRunId: runtime?.activeRunId ?? null, - status: runtime?.status ?? "idle", - pendingRuntimeRequest: - pendingRequest === null - ? null - : { id: pendingRequest.id, kind: pendingRequest.kind, createdAt: pendingRequest.createdAt }, - latestVisibleMessage: null, - latestUserMessageAt: latestUserMessageAt(projection), - hasActionableProposedPlan: false, - itemCount: projection.turnItems.length, - visibleItemCount: projection.visibleTurnItems.length, - createdAt: thread.createdAt, - updatedAt: thread.updatedAt, - archivedAt: thread.archivedAt, - settledOverride: thread.settledOverride, - settledAt: thread.settledAt, - unsettledAt: thread.unsettledAt, - activeOrderKey: thread.activeOrderKey, - autoSettleDisabledAt: thread.autoSettleDisabledAt, - pinnedAt: thread.pinnedAt, - pinOrderKey: thread.pinOrderKey, - snoozedUntil: thread.snoozedUntil ?? null, - snoozedAt: thread.snoozedAt ?? null, - deletedAt: thread.deletedAt, - }); -} - function useResolvedThreadSelection(params: ThreadSelectionRouteParams | undefined) { const routeParams = params ?? {}; const routeEnvironmentId = firstRouteParam(routeParams.environmentId); @@ -199,6 +122,8 @@ function useResolvedThreadSelection(params: ThreadSelectionRouteParams | undefin return { selectedThreadRef, selectedThread, + /** Whether the shell rows hold the thread; `selectedThread` can also come from its detail. */ + selectedThreadListed: selectedThreadShell !== null, selectedThreadCreation, selectedThreadDetailRef, selectedThreadProject, diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index bcfa2f3896f20..b0e0e7bc2e105 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -150,7 +150,8 @@ const SOCKET_BUDGETS = { staleShellResume: { decodedBytes: 400, messages: 2 }, }; -// About 30% above the observed 42 KB. Without the shell opt-ins it carries all 142 rows (217 KB). +// About 30% above the observed 42 KB. Without the shell opt-ins it carries all 162 rows (250 KB), +// and without `omitIdleSubagentParents` the 20 idle subagents that have a child (47 rows, 73 KB). const FLEET_SHELL_BUDGET = { decodedBytes: 55_000, rows: 27, diff --git a/apps/server/integration/TransferBudgetV2Fixture.integration.ts b/apps/server/integration/TransferBudgetV2Fixture.integration.ts index 68cb2c4e841da..fcdedc49f8b83 100644 --- a/apps/server/integration/TransferBudgetV2Fixture.integration.ts +++ b/apps/server/integration/TransferBudgetV2Fixture.integration.ts @@ -878,11 +878,13 @@ export const fleetProjectCreated: ApplicationProjectEvent = { }; export const SHELL_SETTLED_THREADS = 100; -export const SHELL_IDLE_SUBAGENTS = 40; +export const SHELL_IDLE_SUBAGENTS = 60; +const SHELL_NESTED_IDLE_SUBAGENTS = 20; /** * A busy sidebar: settled top-level threads, settled a minute apart, with idle subagent children - * under the newest ones. Opted-in cold connects skip the children and all but 25 settled threads. + * under the newest ones, the newest of which have an idle child of their own. Opted-in cold + * connects skip the children, the nested chains and all but 25 settled threads. */ export function shellFleetEvents( provider: ProviderDriverKind, @@ -897,19 +899,21 @@ export function shellFleetEvents( settledAt, updatedAt: settledAt, }); - if (index < SHELL_SETTLED_THREADS - SHELL_IDLE_SUBAGENTS) return [settled]; + const fromNewest = SHELL_SETTLED_THREADS - index; + if (fromNewest > SHELL_IDLE_SUBAGENTS - SHELL_NESTED_IDLE_SUBAGENTS) return [settled]; + const subagent = (id: string, title: string, parentThreadId: ThreadId) => + threadCreated(provider, ThreadId.make(id), { + title, + createdBy: "agent", + lineage: { parentThreadId, relationshipToParent: "subagent", rootThreadId: threadId }, + }); const childId = ThreadId.make(`transfer-budget-settled-${index}-subagent`); + const child = subagent(childId, `Subagent of settled thread ${index}`, threadId); + if (fromNewest > SHELL_NESTED_IDLE_SUBAGENTS) return [settled, child]; return [ settled, - threadCreated(provider, childId, { - title: `Subagent of settled thread ${index}`, - createdBy: "agent", - lineage: { - parentThreadId: threadId, - relationshipToParent: "subagent", - rootThreadId: threadId, - }, - }), + child, + subagent(`${childId}-nested`, `Nested subagent of settled thread ${index}`, childId), ]; }).flat(); } diff --git a/apps/server/integration/transferBudgetV2.integration.test.ts b/apps/server/integration/transferBudgetV2.integration.test.ts index 4aaac3b9e30e9..fcf0d97a4b51f 100644 --- a/apps/server/integration/transferBudgetV2.integration.test.ts +++ b/apps/server/integration/transferBudgetV2.integration.test.ts @@ -108,6 +108,7 @@ const CLIENT_HISTORY_OPT_INS = { compactIds: "1", deferWork: "1" } satisfies Req >; const CLIENT_SHELL_OPT_INS = { omitIdleSubagents: "1", + omitIdleSubagentParents: "1", omitOlderSettled: "1", } satisfies Required>; const CLIENT_THREAD_SUBSCRIPTION = { @@ -122,6 +123,7 @@ type ShellResumeField = "afterSequence" | "httpSnapshotFallback" | "knownProject const CLIENT_SHELL_SUBSCRIPTION = { requestCompletionMarker: true, omitIdleSubagents: true, + omitIdleSubagentParents: true, omitOlderSettled: true, } satisfies Required>; const query = (optIns: Readonly>) => new URLSearchParams(optIns).toString(); diff --git a/apps/server/src/orchestration-v2/ShellStream.test.ts b/apps/server/src/orchestration-v2/ShellStream.test.ts index 336f9b12f96bf..6203d0c6e795a 100644 --- a/apps/server/src/orchestration-v2/ShellStream.test.ts +++ b/apps/server/src/orchestration-v2/ShellStream.test.ts @@ -181,15 +181,35 @@ describe("omitIdleSubagentShells", () => { // Idle, but a woken grandchild walks through it to reach the root's badge. row("middle", "root"), row("grandchild", "middle"), + row("upper", "root"), + row("lower", "upper"), + row("deep", "lower", { activityRunStatus: "running" }), + // Idle, but a side chat under it needs it as its parent. + row("chat-parent", "root"), + row("chat-sibling", "chat-parent"), + row("chat", "chat-parent", { + lineage: { + parentThreadId: ThreadId.make("chat-parent"), + relationshipToParent: null, + rootThreadId: ThreadId.make("root"), + sideChat: true, + }, + }), ]; + const live = ["root", "running", "queued", "asking", "background"]; expect(omitIdleSubagentShells(threads).map((thread) => thread.id)).toEqual([ - "root", - "running", - "queued", - "asking", - "background", + ...live, "middle", + "upper", + "lower", + "deep", + "chat-parent", + "chat", ]); + // Clients that walk past a missing parent keep only the chains above live or other children. + expect( + omitIdleSubagentShells(threads, { omitIdleSubagentParents: true }).map((thread) => thread.id), + ).toEqual([...live, "upper", "lower", "deep", "chat-parent", "chat"]); }); it("still filters the transcript events of omitted children for a subscriber that opts in", () => { diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index 9a2b3c9a949f5..2ba619e800c1b 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -247,16 +247,56 @@ function hiddenChildAncestorIds( return ancestorIds; } +/** + * The subagents `isLive` accepts, the parents of other children (side chats, forks), and the + * subagents above them, so each of those keeps its chain to the root. + */ +function liveSubagentChainIds>( + rows: ReadonlyArray, + isLive: (row: T) => boolean, +): ReadonlySet { + const subagents = new Map(); + for (const row of rows) { + if (row.lineage.relationshipToParent === "subagent") subagents.set(row.id, row); + } + const kept = new Set(); + for (const row of rows) { + const isSubagent = row.lineage.relationshipToParent === "subagent"; + if (isSubagent && !isLive(row)) continue; + const rowParentId = row.lineage.parentThreadId; + let current: T | undefined = isSubagent + ? row + : rowParentId === null + ? undefined + : subagents.get(rowParentId); + while (current !== undefined && !kept.has(current.id)) { + kept.add(current.id); + const parentId: string | null = current.lineage.parentThreadId; + current = parentId === null ? undefined : subagents.get(parentId); + } + } + return kept; +} + function omitIdleSubagentShellsWith( threads: ReadonlyArray, facts: Pick, + omitIdleSubagentParents: boolean, ): ReadonlyArray { + const chains = omitIdleSubagentParents + ? liveSubagentChainIds( + threads, + (thread) => isLiveSubagentShell(thread) || facts.remoteTaskParentIds.has(thread.id), + ) + : null; return threads.filter( (thread) => thread.lineage.relationshipToParent !== "subagent" || - isLiveSubagentShell(thread) || - facts.subagentParentIds.has(thread.id) || - facts.remoteTaskParentIds.has(thread.id), + (chains === null + ? isLiveSubagentShell(thread) || + facts.subagentParentIds.has(thread.id) || + facts.remoteTaskParentIds.has(thread.id) + : chains.has(thread.id)), ); } @@ -297,15 +337,20 @@ function omitOlderSettledShellsWith( } /** - * Idle subagent children have no sidebar row, so clients that opt in read them from their thread - * family instead. A child stays while it is live or the parent of another child, so the request - * and working walks over shell rows never meet a gap in a nested chain. It also stays as the - * parent of a task in a linked environment, whose clients nest that task under it. + * Idle subagent children have no sidebar row; opted-in clients read them from their thread family. + * A child stays while live, the parent of another child, or the parent of a task in a linked + * environment (its clients nest the task there). With `omitIdleSubagentParents`, a parent of + * another child stays only above a live subagent, a side chat, a fork, or such a task parent. */ export function omitIdleSubagentShells( threads: ReadonlyArray, + options: { readonly omitIdleSubagentParents: boolean } = { omitIdleSubagentParents: false }, ): ReadonlyArray { - return omitIdleSubagentShellsWith(threads, shellTrimFacts(threads)); + return omitIdleSubagentShellsWith( + threads, + shellTrimFacts(threads), + options.omitIdleSubagentParents, + ); } /** @@ -321,7 +366,7 @@ export function activeShellRowsForSubscriber( } { const facts = shellTrimFacts(threads); return { - threads: omitIdleSubagents ? omitIdleSubagentShellsWith(threads, facts) : threads, + threads: omitIdleSubagents ? omitIdleSubagentShellsWith(threads, facts, false) : threads, subagentThreadIds: facts.subagentThreadIds, }; } @@ -345,6 +390,8 @@ export function omitOlderSettledShells(input: { interface ShellSubscriberOptions { readonly omitIdleSubagents: boolean; + /** Only with `omitIdleSubagents`. */ + readonly omitIdleSubagentParents: boolean; readonly omitOlderSettled: boolean; } @@ -354,7 +401,7 @@ function trimShellSnapshot( options: ShellSubscriberOptions & { readonly now: DateTime.Utc }, ): OrchestrationV2ShellSnapshot { const threads = options.omitIdleSubagents - ? omitIdleSubagentShellsWith(base.threads, facts) + ? omitIdleSubagentShellsWith(base.threads, facts, options.omitIdleSubagentParents) : base.threads; if (!options.omitOlderSettled) return { ...base, threads }; const older = omitOlderSettledShellsWith( @@ -373,13 +420,17 @@ function shellSubscriberThreadIds( options: ShellSubscriberOptions, ): ReadonlyArray { const facts = shellTrimFacts(index); + const chains = + options.omitIdleSubagents && options.omitIdleSubagentParents + ? liveSubagentChainIds(index, (entry) => entry.mayBeLive || entry.hasRemoteTasks) + : null; const subagents = index.filter( (entry) => entry.lineage.relationshipToParent === "subagent" && (!options.omitIdleSubagents || - entry.mayBeLive || - entry.hasRemoteTasks || - facts.subagentParentIds.has(entry.id)), + (chains === null + ? entry.mayBeLive || entry.hasRemoteTasks || facts.subagentParentIds.has(entry.id) + : chains.has(entry.id))), ); const ancestorIds = hiddenChildAncestorIds([ ...subagents, diff --git a/apps/server/src/orchestration-v2/http.test.ts b/apps/server/src/orchestration-v2/http.test.ts index 07385bf134d08..949918a057889 100644 --- a/apps/server/src/orchestration-v2/http.test.ts +++ b/apps/server/src/orchestration-v2/http.test.ts @@ -10,6 +10,7 @@ import { ORCHESTRATION_PROTOCOL_VERSION_TEXT, OrchestrationV2AppThread, OrchestrationV2Run, + OrchestrationV2Subagent, OrchestrationV2TurnItem, OrchestrationV2ThreadTranscript, PROVIDER_SEND_TURN_MAX_FILE_BYTES, @@ -117,6 +118,30 @@ const rolledBackRun = Schema.decodeSync(OrchestrationV2Run)({ contextHandoffId: null, }); +// A finished task that `task-parent` delegated to a linked environment. +const remoteTask = Schema.decodeSync(OrchestrationV2Subagent)({ + id: "node:remote-task", + threadId: "task-parent", + runId: null, + parentNodeId: "node:task-parent-root", + origin: "app_owned", + createdBy: "agent", + driver: "remote", + providerInstanceId: "codex", + providerThreadId: null, + childThreadId: null, + remoteChild: { environmentId: "environment:box", threadId: "thread:on-the-box", label: "Box" }, + nativeTaskRef: null, + prompt: "Run the suite there.", + title: null, + model: null, + status: "completed", + result: "Done", + startedAt: now, + completedAt: now, + updatedAt: now, +}); + // Serves the orchestration HTTP routes over real stores; `allowed` gates the read scope. const serveOrchestrationHttp = (allowed: () => boolean) => Effect.gen(function* () { @@ -340,7 +365,7 @@ it.effect("exports full history with size, scope, and missing-thread checks", () }), ); -it.effect("leaves older settled threads out of the HTTP shell only when asked", () => +it.effect("trims the HTTP shell only when asked", () => Effect.gen(function* () { const { eventSink, handler } = yield* serveOrchestrationHttp(() => true); yield* eventSink.write({ @@ -392,5 +417,60 @@ it.effect("leaves older settled threads out of the HTTP shell only when asked", for (const cursor of [full.snapshotSequence + 1, "x"]) { expect((yield* shell(`?omitOlderSettled=1&resumeFrom=${cursor}`)).threads).toHaveLength(26); } + + // Two idle chains under the source thread. Only `task-parent` has a task in a linked + // environment, whose clients nest that task under it. + const subagent = (id: string, parent: string) => { + const threadId = ThreadId.make(id); + return { + id: EventId.make(`create-${id}`), + type: "thread.created" as const, + threadId, + occurredAt: now, + payload: { + ...thread, + id: threadId, + lineage: { + rootThreadId: thread.id, + parentThreadId: ThreadId.make(parent), + relationshipToParent: "subagent" as const, + }, + }, + }; + }; + yield* eventSink.write({ + events: [ + subagent("above-task-parent", thread.id), + subagent("task-parent", "above-task-parent"), + subagent("idle-parent", thread.id), + subagent("idle-leaf", "idle-parent"), + { + id: EventId.make("create-remote-task"), + type: "subagent.updated", + threadId: remoteTask.threadId, + occurredAt: now, + payload: remoteTask, + }, + ], + }); + const subagentIds = (query: string) => + shell(query).pipe( + Effect.map((snapshot) => + snapshot.threads + .flatMap((row) => (row.id.startsWith("settled-") ? [] : [row.id])) + .toSorted(), + ), + ); + expect(yield* subagentIds("?omitIdleSubagents=1")).toEqual([ + "above-task-parent", + "idle-parent", + thread.id, + "task-parent", + ]); + expect(yield* subagentIds("?omitIdleSubagents=1&omitIdleSubagentParents=1")).toEqual([ + "above-task-parent", + thread.id, + "task-parent", + ]); }), ); diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index b58ce1b143119..1072dab210dff 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -83,6 +83,7 @@ export const layer = HttpApiBuilder.group( const loadShellSnapshot = Effect.fn("http.orchestration.loadShellSnapshot")( function* (options: { readonly omitIdleSubagents: boolean; + readonly omitIdleSubagentParents: boolean; readonly omitOlderSettled: boolean; }) { const { snapshot } = yield* loadSubscriberShellSnapshot({ @@ -105,6 +106,7 @@ export const layer = HttpApiBuilder.group( // A client that holds rows names their cursor; a small answer tells it to keep them. const answerShellSnapshot = Effect.fn("http.orchestration.answerShellSnapshot")((query: { readonly omitIdleSubagents?: string; + readonly omitIdleSubagentParents?: string; readonly omitOlderSettled?: string; readonly resumeFrom?: string; }) => { @@ -114,6 +116,7 @@ export const layer = HttpApiBuilder.group( resumeFrom: Number.isSafeInteger(resumeFrom) && resumeFrom >= 0 ? resumeFrom : undefined, loadSnapshot: loadShellSnapshot({ omitIdleSubagents: query.omitIdleSubagents === "1", + omitIdleSubagentParents: query.omitIdleSubagentParents === "1", omitOlderSettled: query.omitOlderSettled === "1", }), }); diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 7f52d5963a07d..e2eb211a20066 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -710,11 +710,21 @@ it.effect("shell snapshots omit idle children and older settled threads only on shell("live", "parent", "running"), ...Array.from({ length: 25 }, (_, index) => shell(`recent-${index}`, null, null, 1)), shell("old", null, null, 30), + shell("middle", "parent", null), + shell("leaf", "middle", null), + // Idle, and kept either way: a live subagent is below it. + shell("upper", "parent", null), + shell("deep", "upper", "running"), shell("remote-parent", null, null, 40), shell("remote-child-parent", "parent", null), + // Idle, and kept either way: an idle subagent with a remote task is below it. + shell("above-remote", "parent", null), + shell("remote-nested", "above-remote", null), ]; + let readThreadIds: ReadonlyArray | undefined; const firstSnapshot = (input: { readonly omitIdleSubagents?: boolean; + readonly omitIdleSubagentParents?: boolean; readonly omitOlderSettled?: boolean; }) => subscribeOrchestrationV2Shell(input).pipe( @@ -732,15 +742,19 @@ it.effect("shell snapshots omit idle children and older settled threads only on })), ), readShellSnapshot: (options?: { readonly threadIds?: ReadonlyArray }) => - Effect.succeed( - Effect.succeed({ - schemaVersion: 1, - snapshotSequence: 0, - threads: threads.filter( - (thread) => options?.threadIds?.includes(thread.id) ?? true, - ), - archivedThreads: [], - }), + Effect.sync(() => { + readThreadIds = options?.threadIds; + }).pipe( + Effect.as( + Effect.succeed({ + schemaVersion: 1, + snapshotSequence: 0, + threads: threads.filter( + (thread) => options?.threadIds?.includes(thread.id) ?? true, + ), + archivedThreads: [], + }), + ), ), } as unknown as ThreadManagement.ThreadManagementService["Service"]), Effect.provideService(ProjectStore.ProjectStoreV2, { @@ -763,6 +777,7 @@ it.effect("shell snapshots omit idle children and older settled threads only on ); const recent = threads.slice(3, 28).map((thread) => thread.id); + const remote = ["remote-parent", "remote-child-parent", "above-remote", "remote-nested"]; const full = yield* firstSnapshot({}); assert.deepEqual( full?.threads.map((thread) => thread.id), @@ -772,14 +787,24 @@ it.effect("shell snapshots omit idle children and older settled threads only on const withoutChildren = yield* firstSnapshot({ omitIdleSubagents: true }); assert.deepEqual( withoutChildren?.threads.map((thread) => thread.id), - ["parent", "live", ...recent, "old", "remote-parent", "remote-child-parent"], + ["parent", "live", ...recent, "old", "middle", "upper", "deep", ...remote], ); // A linked environment's clients nest its task under its parent, so that parent stays: an - // older settled thread or an idle subagent. + // older settled thread, or an idle subagent with the idle subagents above it. + const withoutIdleParents = yield* firstSnapshot({ + omitIdleSubagents: true, + omitIdleSubagentParents: true, + }); + assert.deepEqual( + withoutIdleParents?.threads.map((thread) => thread.id), + ["parent", "live", ...recent, "old", "upper", "deep", ...remote], + ); + // The store builds only these shells, so the ids it is asked for keep them too. + assert.includeMembers([...(readThreadIds ?? [])], remote); const withoutOld = yield* firstSnapshot({ omitIdleSubagents: true, omitOlderSettled: true }); assert.deepEqual( withoutOld?.threads.map((thread) => thread.id), - ["parent", "live", ...recent, "remote-parent", "remote-child-parent"], + ["parent", "live", ...recent, "middle", "upper", "deep", ...remote], ); assert.deepEqual(withoutOld?.olderSettled?.counts, [ { projectId: ProjectId.make("project"), count: 2 }, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 5e81541ff3530..08b95d5ebaef0 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1128,6 +1128,7 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc readonly knownProjectIdentities?: string; readonly requestCompletionMarker?: boolean; readonly omitIdleSubagents?: boolean; + readonly omitIdleSubagentParents?: boolean; readonly omitOlderSettled?: boolean; }) { const sql = yield* SqlClient.SqlClient; @@ -1169,6 +1170,7 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc latestSequence: applicationEvents.latestApplicationSequence, options: { omitIdleSubagents: input.omitIdleSubagents === true, + omitIdleSubagentParents: input.omitIdleSubagentParents === true, omitOlderSettled: input.omitOlderSettled === true, }, }); diff --git a/apps/web/src/components/ThreadRouteView.tsx b/apps/web/src/components/ThreadRouteView.tsx index a7b7da0a6b53b..2a810f8abb0df 100644 --- a/apps/web/src/components/ThreadRouteView.tsx +++ b/apps/web/src/components/ThreadRouteView.tsx @@ -1,9 +1,10 @@ import { useAtomValue } from "@effect/atom-react"; import { parseScopedThreadKey, scopedThreadKey } from "@t3tools/client-runtime/environment"; import { parseThreadKey, threadKey } from "@t3tools/client-runtime/state/entities"; -import { visibilityParentKey } from "@t3tools/client-runtime/state/threads"; -import type { ScopedThreadRef } from "@t3tools/contracts"; +import { type ThreadFamilyRead, visibilityParentKey } from "@t3tools/client-runtime/state/threads"; +import { isListedThreadLineage, type ScopedThreadRef } from "@t3tools/contracts"; import * as Option from "effect/Option"; +import { Atom } from "effect/reactivity"; import { Navigate, useNavigate } from "@tanstack/react-router"; import { useEffect, useState } from "react"; @@ -27,11 +28,14 @@ import { type ThreadRouteTarget, } from "../threadRoutes"; +const NO_OPEN_FAMILY = Atom.make("settled"); + /** - * Keeps the family of an open thread the shell leaves out (an older settled thread) in the shell - * rows until the route leaves it, so every reader of its shell row finds it. + * Keeps the family of an open thread the shell leaves out (an older settled thread, an idle + * subagent), or whose parent it leaves out, in the shell rows until the route leaves it, so every + * reader of its shell row finds it. Returns whether those rows can still arrive. */ -function useOpenThreadFamily(threadRef: ScopedThreadRef | null, missing: boolean) { +function useOpenThreadFamily(threadRef: ScopedThreadRef | null, missing: boolean): boolean { const key = threadRef === null ? null : scopedThreadKey(threadRef); // Sticks to the route thread, so the row the family brings in does not release it. const [mountedKey, setMountedKey] = useState(null); @@ -43,6 +47,12 @@ function useOpenThreadFamily(threadRef: ScopedThreadRef | null, missing: boolean if (ref === null) return; return threadFamilies.mount(appAtomRegistry, ref.environmentId, ref.threadId); }, [mountedKey]); + const read = useAtomValue( + threadRef === null + ? NO_OPEN_FAMILY + : threadFamilies.mountedReadAtom(threadRef.environmentId, threadRef.threadId), + ); + return missing && read === "pending"; } /** @@ -108,7 +118,20 @@ export function ThreadRouteView({ unlistedThreadRef?.environmentId ?? null, unlistedThreadRef?.threadId ?? null, ); - useOpenThreadFamily(serverThreadRef, bootstrapComplete && unlistedThreadRef !== null); + // A woken child can arrive before the idle parent the shell left out; its family brings it. + const listedParentRef = + serverThreadRef === null || + serverThreadShell === null || + serverThreadShell.lineage.parentThreadId === null || + isListedThreadLineage(serverThreadShell.lineage) + ? null + : { ...serverThreadRef, threadId: serverThreadShell.lineage.parentThreadId }; + const listedParentShell = useQuietThreadShell(listedParentRef); + const parentMissing = listedParentRef !== null && listedParentShell === null; + const familyPending = useOpenThreadFamily( + serverThreadRef, + bootstrapComplete && (unlistedThreadRef !== null || parentMissing), + ); const hiddenKeys = useAtomValue(environmentThreadShells.hiddenThreadKeysAtom); const unlistedThread = Option.getOrNull(Option.map(unlistedThreadDetail.data, (p) => p.thread)); const unlistedParentKey = @@ -209,10 +232,14 @@ export function ThreadRouteView({ redirect = ; } } else if (hiddenByTagFilter) { - redirect = ; + // A thread whose parent row has not arrived yet reads as hidden until its family read settles. + if (!familyPending) redirect = ; } else if (offShellBlocked) { - // Before the shell bootstraps, a listed visible thread can still look off-shell. - if (bootstrapComplete && unlistedThread !== null) redirect = ; + // Before the shell bootstraps, a listed visible thread can still look off-shell. So can an + // idle subagent until its family read brings in the parent the shell left out. + if (bootstrapComplete && unlistedThread !== null && !familyPending) { + redirect = ; + } } else if (renderState === "ready" || (renderState === "loading" && serverThreadShell !== null)) { view = ( stored.environmentId === environmentId ? Option.some( - stored.origin === undefined + stored.originV2 === undefined ? stored.snapshot - : { ...stored.snapshot, origin: stored.origin }, + : { ...stored.snapshot, origin: stored.originV2 }, ) : Option.none(), ), @@ -871,7 +871,7 @@ export const layer = Layer.effectContext( schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, environmentId, snapshot, - ...(origin === undefined ? {} : { origin }), + ...(origin === undefined ? {} : { originV2: origin }), }).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause))); yield* writeDatabaseValue(database, SHELL_STORE_NAME, environmentId, encoded); }, diff --git a/docs/internals/connection-runtime.md b/docs/internals/connection-runtime.md index 4ebad4c7056ef..3225e88cdc0eb 100644 --- a/docs/internals/connection-runtime.md +++ b/docs/internals/connection-runtime.md @@ -108,11 +108,17 @@ read its cached shell, so the first frame holds all cached rows. Shell snapshots can leave rows out when the client asks. Servers that advertise `shellThreadFamilies` leave out idle subagent children: a child stays while it has a live or queued run, a pending request or background work, or a subagent -child of its own, so badge and working walks never meet a gap. Servers that also -advertise `shellOlderSettled` keep only the 25 newest settled top-level threads -(and any tied with the 25th, pinned, snoozed, live, or above a kept child); the -snapshot's `olderSettled` names that cut and counts the rest per project. Left-out -rows load on demand and merge behind shell rows +child of its own, so badge and working walks never meet a gap. It also stays as +the parent of a task in a linked environment, whose clients nest that task under +it. Clients that also send `omitIdleSubagentParents` get an idle child with a +subagent child only while a live subagent, a side chat, a fork or such a task +parent is below it. A chain that wakes later reaches them as a delta without the +idle rows above it, so they read the family of any shell row whose parent is +missing. +Servers that also advertise `shellOlderSettled` keep only the 25 newest settled +top-level threads (and any tied with the 25th, pinned, snoozed, live, or above a +kept child); the snapshot's `olderSettled` names that cut and counts the rest per +project. Left-out rows load on demand and merge behind shell rows ([thread families](../../packages/client-runtime/src/state/threadFamilies.ts)): the family a thread view or a queued message mounts holds every active thread sharing the thread's root, so it also brings back an old settled thread opened by diff --git a/packages/client-runtime/src/platform/orchestrationCache.test.ts b/packages/client-runtime/src/platform/orchestrationCache.test.ts index 8b9b16829a4be..82635ffaa2ac1 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.test.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.test.ts @@ -96,6 +96,36 @@ describe("orchestration cache envelopes", () => { expect(decodeStoredShellSnapshotSync(shell).schemaVersion).toBe(3); }); + it("gives a client from before `originV2` no origin to resume from", () => { + // What that client decodes. It drops keys and origin flags it does not know. + const decodeAsOlderClient = Schema.decodeUnknownSync( + Schema.Struct({ + schemaVersion: Schema.Literal(3), + origin: Schema.optionalKey(Schema.Struct({ omitIdleSubagents: Schema.Boolean })), + }), + ); + const origin = { + serverVersion: "1.0.0", + databaseMigration: 70, + omitIdleSubagents: true, + omitOlderSettled: true, + snapshotAt: 1_800_000_000_000, + }; + const stored = encodeStoredShellSnapshot({ + schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, + environmentId, + snapshot: v2ShellSnapshot, + originV2: origin, + }); + + expect(decodeStoredShellSnapshotSync(stored).originV2).toEqual(origin); + expect(decodeAsOlderClient(stored)).toEqual({ schemaVersion: 3 }); + // A new flag under `origin` would not stop it: it decodes, and the flags it knows match. + expect( + decodeAsOlderClient({ ...stored, origin: { ...origin, omitIdleSubagentParents: true } }), + ).toEqual({ schemaVersion: 3, origin: { omitIdleSubagents: true } }); + }); + it.effect("discards V1-versioned cache envelopes after a decode failure", () => Effect.gen(function* () { let discardCount = 0; diff --git a/packages/client-runtime/src/platform/orchestrationCache.ts b/packages/client-runtime/src/platform/orchestrationCache.ts index 3d24e03f113f9..2b7afdf28aeb0 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.ts @@ -31,7 +31,12 @@ export const StoredOrchestrationShellSnapshot = Schema.Struct({ schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION), environmentId: EnvironmentId, snapshot: OrchestrationV2ShellSnapshotJson, - origin: Schema.optionalKey(ShellCacheOrigin), + /** + * Not `origin`: clients from before `omitIdleSubagentParents` read that key, ignore flags they + * do not know, and would resume rows that lack the idle parents they expect. They find no + * origin here, so they load a fresh snapshot. Rename again when an omission changes the rows. + */ + originV2: Schema.optionalKey(ShellCacheOrigin), }); export const StoredOrchestrationThreadSnapshot = Schema.Struct({ diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 2699c8846f7bc..170a999a19899 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -880,6 +880,7 @@ describe("environment shell synchronization", () => { expect(yield* Queue.take(subscribeInputs)).toMatchObject({ afterSequence: 1, omitIdleSubagents: true, + omitIdleSubagentParents: true, omitOlderSettled: true, }); expect(yield* Ref.get(loadOptions)).toEqual([ diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 86584c4a03faa..adb3e5560a50e 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -682,7 +682,9 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const current = yield* SubscriptionRef.get(state); const omission = { - ...(omitIdleSubagents ? { omitIdleSubagents: true as const } : {}), + ...(omitIdleSubagents + ? { omitIdleSubagents: true as const, omitIdleSubagentParents: true as const } + : {}), ...(omitOlderSettled ? { omitOlderSettled: true as const } : {}), }; if (Option.isNone(current.snapshot) || !(resumingHeldRows || loaded?.kind === "snapshot")) { diff --git a/packages/client-runtime/src/state/shellSnapshotHttp.ts b/packages/client-runtime/src/state/shellSnapshotHttp.ts index e98846a0e3c73..508bd4bf0befb 100644 --- a/packages/client-runtime/src/state/shellSnapshotHttp.ts +++ b/packages/client-runtime/src/state/shellSnapshotHttp.ts @@ -53,7 +53,9 @@ export const fetchEnvironmentShellSnapshot = Effect.fn( ) { const endpoint = { query: { - ...(input.omitIdleSubagents === true ? { omitIdleSubagents: "1" } : {}), + ...(input.omitIdleSubagents === true + ? { omitIdleSubagents: "1", omitIdleSubagentParents: "1" } + : {}), ...(input.omitOlderSettled === true ? { omitOlderSettled: "1" } : {}), ...(input.resumeFrom === undefined ? {} : { resumeFrom: String(input.resumeFrom) }), }, diff --git a/packages/client-runtime/src/state/threadFamilies.test.ts b/packages/client-runtime/src/state/threadFamilies.test.ts index 2b2a4a6a12780..92f1d3475705f 100644 --- a/packages/client-runtime/src/state/threadFamilies.test.ts +++ b/packages/client-runtime/src/state/threadFamilies.test.ts @@ -33,8 +33,12 @@ import { threadFamilyRowsMissingFromShell, } from "./threadFamilies.ts"; -const row = (id: string, title = id) => - ({ id: ThreadId.make(id), title }) as unknown as OrchestrationV2ThreadShell; +const row = (id: string, title = id, parentThreadId: string | null = null, rootThreadId = id) => + ({ + id: ThreadId.make(id), + title, + lineage: { parentThreadId, rootThreadId, relationshipToParent: "subagent" }, + }) as unknown as OrchestrationV2ThreadShell; describe("threadFamilyRowsMissingFromShell", () => { it("adds only the rows the shell left out, once, with shell rows winning", () => { @@ -177,10 +181,10 @@ describe("createThreadFamilyAtoms", () => { threads: [parent], archivedThreads: [], } as OrchestrationV2ShellSnapshot; - const shellAt = (generation: number): EnvironmentShellState => ({ + const shellAt = (generation: number, threads = shellSnapshot.threads) => ({ ...EMPTY_SHELL_STATE, - status: "live", - snapshot: Option.some(shellSnapshot), + status: "live" as const, + snapshot: Option.some({ ...shellSnapshot, threads }), generation, }); const shellState = Atom.make>( @@ -199,7 +203,7 @@ describe("createThreadFamilyAtoms", () => { Effect.tap(() => Effect.yieldNow), ); registry.mount(families.snapshotAtom(TARGET.environmentId)); - families.mount(registry, TARGET.environmentId, parent.id); + const release = families.mount(registry, TARGET.environmentId, parent.id); // A cached shell from before this session's first snapshot reads nothing. expect(threadFamilyRead(registry.get(family))).toBe("pending"); @@ -224,12 +228,71 @@ describe("createThreadFamilyAtoms", () => { expect(threadFamilyRead(registry.get(family))).toBe("failed"); expect(mergedIds()).toEqual(["parent"]); - families.refresh(registry, TARGET.environmentId, parent.id); + // An open thread reads a failed family again by itself, with a growing pause. + const pauses: Array<{ readonly run: () => void; readonly ms: number }> = []; + const stopRetry = families.retryWhileFailed( + registry, + TARGET.environmentId, + parent.id, + (run, ms) => { + pauses.push({ run, ms }); + return () => undefined; + }, + ); + expect(pauses.map(({ ms }) => ms)).toEqual([1_000]); + pauses[0]?.run(); + expect(threadFamilyRead(registry.get(family))).toBe("pending"); + yield* Queue.offer( + replies, + Effect.fail(new OrchestrationV2GetShellSnapshotError({ message: "read failed" })), + ); + yield* settle; + expect(threadFamilyRead(registry.get(family))).toBe("failed"); + expect(pauses.map(({ ms }) => ms)).toEqual([1_000, 2_000]); + pauses[1]?.run(); expect(threadFamilyRead(registry.get(family))).toBe("pending"); yield* Queue.offer(replies, Effect.succeed({ snapshotSequence: 5, threads: [child] })); yield* settle; expect(threadFamilyRead(registry.get(family))).toBe("settled"); expect(mergedIds()).toEqual(["parent", "idle-child"]); + expect(pauses).toHaveLength(2); + stopRetry(); + + // Children that woke after the shell left out their idle parent arrive as deltas, and + // read their family unmounted, once for the root. + release(); + const woken = row("woken", "woken", child.id, parent.id); + const sibling = row("woken-sibling", "woken-sibling", child.id, parent.id); + registry.set(shellState, AsyncResult.success(shellAt(2, [parent, woken, sibling]))); + const wokenFamily = families.familyAtom(TARGET.environmentId, woken.id); + const settleWoken = AtomRegistry.getResult(registry, wokenFamily, { + suspendOnWaiting: true, + }).pipe( + Effect.ignore, + Effect.tap(() => Effect.yieldNow), + ); + yield* Queue.offer( + replies, + Effect.fail(new OrchestrationV2GetShellSnapshotError({ message: "read failed" })), + ); + yield* settleWoken; + expect(threadFamilyRead(registry.get(wokenFamily))).toBe("failed"); + expect(mergedIds()).toEqual(["parent", "woken", "woken-sibling"]); + + // The next delta retries the failed read instead of keeping the gap until reconnect. + registry.set(shellState, AsyncResult.success(shellAt(2, [parent, { ...woken }, sibling]))); + yield* Queue.offer( + replies, + Effect.succeed({ snapshotSequence: 6, threads: [parent, child, woken, sibling] }), + ); + yield* settleWoken; + expect(mergedIds()).toEqual(["parent", "woken", "woken-sibling", "idle-child"]); + + // Opening the woken child mounts that same read. It answers only once it is mounted. + const mountedRead = families.mountedReadAtom(TARGET.environmentId, woken.id); + expect(registry.get(mountedRead)).toBe("pending"); + families.mount(registry, TARGET.environmentId, woken.id); + expect(registry.get(mountedRead)).toBe("settled"); }), ), ); diff --git a/packages/client-runtime/src/state/threadFamilies.ts b/packages/client-runtime/src/state/threadFamilies.ts index 0f1bb8d9cd274..74aeab0f75672 100644 --- a/packages/client-runtime/src/state/threadFamilies.ts +++ b/packages/client-runtime/src/state/threadFamilies.ts @@ -6,6 +6,7 @@ import { type OrchestrationV2ThreadShell, type ThreadId, } from "@t3tools/contracts"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as SubscriptionRef from "effect/SubscriptionRef"; @@ -13,7 +14,7 @@ import { AsyncResult, Atom, type AtomRegistry } from "effect/reactivity"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import type * as EnvironmentRegistry from "../connection/registry.ts"; -import { EnvironmentRpcUnavailableError, request } from "../rpc/client.ts"; +import { EnvironmentRpcUnavailableError, request, subscriptionRetryDelay } from "../rpc/client.ts"; import { arrayElementsEqual } from "./entities.ts"; import { createEnvironmentQueryAtomFamily, selectEnvironmentQueryData } from "./runtime.ts"; import type { EnvironmentShellState } from "./shell.ts"; @@ -65,11 +66,33 @@ export function threadFamilyRowsMissingFromShell(input: { return missing; } +/** + * Subagents and side chats whose parent row the shell left out, one per root. A shell sent with + * `omitIdleSubagentParents` drops idle chains, so a child that wakes or is opened later arrives + * without them. The family is read by the child, the read an open thread's mount already made. + */ +function threadsWithMissingParents( + threads: ReadonlyArray, +): ReadonlyArray { + const present = new Set(threads.map((thread) => thread.id)); + const byRoot = new Map(); + for (const { id, lineage } of threads) { + if (lineage.parentThreadId === null || present.has(lineage.parentThreadId)) continue; + if ( + (lineage.relationshipToParent === "subagent" || lineage.sideChat === true) && + !byRoot.has(lineage.rootThreadId) + ) { + byRoot.set(lineage.rootThreadId, id); + } + } + return [...byRoot.values()]; +} + /** * Shells with `omitIdleSubagents` or `omitOlderSettled` leave rows to on-demand reads: a family is * every active thread sharing a `lineage.rootThreadId`, read once per shell generation; shelf pages * and search rows mount through `mountRows`. All merge behind shell rows, which win (deltas carry - * every change), and are never persisted. + * every change), and are never persisted. A child whose parent row is missing reads its family too. */ export function createThreadFamilyAtoms( runtime: Atom.AtomRuntime, @@ -143,6 +166,14 @@ export function createThreadFamilyAtoms( Atom.withLabel(`environment-thread-family-mounts:${environmentId}`), ), ); + const mountedReadAtom = Atom.family((key: string) => { + const [environmentId] = JSON.parse(key) as [EnvironmentId, ThreadId]; + return Atom.make((get): ThreadFamilyRead => + get(mountsAtom(environmentId)).has(`family:${key}`) + ? threadFamilyRead(get(familyAtom(key))) + : "pending", + ); + }); const familyRowsAtom = Atom.family((key: string) => Atom.make((get): ReadonlyArray => { @@ -196,6 +227,44 @@ export function createThreadFamilyAtoms( return release; }; + /** + * Reads a failed family again after a growing pause, until the returned release runs. A failed + * read says nothing about the thread, and nothing else reads again on an idle connection. + */ + const retryWhileFailed = ( + registry: AtomRegistry.AtomRegistry, + environmentId: EnvironmentId, + threadId: ThreadId, + later: (run: () => void, ms: number) => () => void = (run, ms) => { + // @effect-diagnostics-next-line globalTimers:off - The pause runs in an atom subscription, outside an Effect runtime. + const timer = setTimeout(run, ms); + return () => clearTimeout(timer); + }, + ): (() => void) => { + let failures = 0; + let cancel: (() => void) | undefined; + const stop = registry.subscribe( + familyAtom(familyKey(environmentId, threadId)), + (result) => { + const read = threadFamilyRead(result); + if (read === "settled") failures = 0; + if (read !== "failed" || cancel !== undefined) return; + cancel = later( + () => { + cancel = undefined; + refresh(registry, environmentId, threadId); + }, + Duration.toMillis(subscriptionRetryDelay("1 second", failures++)), + ); + }, + { immediate: true }, + ); + return () => { + stop(); + cancel?.(); + }; + }; + const snapshotAtom = Atom.family((environmentId: EnvironmentId) => { let previous: { readonly snapshot: OrchestrationV2ShellSnapshot; @@ -203,16 +272,36 @@ export function createThreadFamilyAtoms( readonly families: ReadonlyArray; readonly merged: OrchestrationV2ShellSnapshot; } | null = null; + let orphans: { + readonly threads: ReadonlyArray; + readonly rows: ReadonlyArray>>; + } | null = null; return Atom.make((get): OrchestrationV2ShellSnapshot | null => { const state = shellStateValue(get(shellStateAtom(environmentId))); const snapshot = state === null ? null : Option.getOrNull(state.snapshot); const mounts = get(mountsAtom(environmentId)); - if (snapshot === null || state === null || mounts.size === 0) return snapshot; + if (snapshot === null || state === null) return snapshot; + if (orphans?.threads !== snapshot.threads) { + const orphanIds = threadsWithMissingParents(snapshot.threads); + // Like a remount, a shell change retries a failed read instead of keeping the gap. + for (const threadId of orphanIds) { + if ( + threadFamilyRead(get.once(familyAtom(familyKey(environmentId, threadId)))) === "failed" + ) { + get.refresh(queryFor(environmentId, threadId, state.generation)); + } + } + orphans = { + threads: snapshot.threads, + rows: orphanIds.map((threadId) => familyRowsAtom(familyKey(environmentId, threadId))), + }; + } + if (mounts.size === 0 && orphans.rows.length === 0) return snapshot; // A read from before the latest authoritative snapshot can hold threads removed before it, // and that snapshot cleared those removals; such reads (a cached search) wait for new ones. const cut = snapshot.olderSettled?.asOfSequence ?? 0; - const families = [...mounts.values()].flatMap(({ rows }) => - get(rows).filter((read) => read.snapshotSequence >= cut), + const families = [...[...mounts.values()].map(({ rows }) => rows), ...orphans.rows].flatMap( + (rows) => get(rows).filter((read) => read.snapshotSequence >= cut), ); // Most shell updates move only the cursor. With the same rows, removals and families, keep // the merged rows so thread lists skip their rebuild. @@ -250,8 +339,15 @@ export function createThreadFamilyAtoms( /** The current generation's family read, for decisions that need a settled answer. */ familyAtom: (environmentId: EnvironmentId, threadId: ThreadId) => familyAtom(familyKey(environmentId, threadId)), + /** + * `familyAtom`'s read once the family is mounted, and pending until then: a read cached from + * an earlier mount settles before `mount` merges its rows. Does not start a read. + */ + mountedReadAtom: (environmentId: EnvironmentId, threadId: ThreadId) => + mountedReadAtom(familyKey(environmentId, threadId)), mount, mountRows, refresh, + retryWhileFailed, }; } diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index a9c635486db7d..5496dbdb5917c 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -685,6 +685,7 @@ const EnvironmentOrchestrationThreadSnapshotParams = Schema.Struct({ export const EnvironmentOrchestrationShellSnapshotQuery = Schema.Struct({ omitIdleSubagents: Schema.optionalKey(Schema.String), + omitIdleSubagentParents: Schema.optionalKey(Schema.String), omitOlderSettled: Schema.optionalKey(Schema.String), /** * The cursor of rows the client holds, loaded with the same omissions. The server answers diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 38046a17e8b4a..950f5c47946dc 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3845,6 +3845,13 @@ export const OrchestrationV2SubscribeShellInput = Schema.Struct({ * servers that advertise `shellThreadFamilies`. */ omitIdleSubagents: Schema.optionalKey(Schema.Boolean), + /** + * With `omitIdleSubagents`, an idle subagent no longer stays as the parent of an idle subagent: + * only above a live subagent, a side chat, a fork, or a subagent with a task in a linked + * environment. Send only from clients that read the family of a child whose parent row is + * missing; older servers ignore it. + */ + omitIdleSubagentParents: Schema.optionalKey(Schema.Boolean), /** * Snapshots leave out older settled threads and count them in `olderSettled`; read them with * `getOlderSettledThreads`. Send only to servers that advertise `shellOlderSettled`.