diff --git a/apps/mobile/src/state/entities.test.tsx b/apps/mobile/src/state/entities.test.tsx index 9b7453d3f9d91..48dc17289e2b8 100644 --- a/apps/mobile/src/state/entities.test.tsx +++ b/apps/mobile/src/state/entities.test.tsx @@ -3,7 +3,7 @@ import { act } from "react"; import { createRoot } from "react-dom/client"; import { expect, it, vi } from "vite-plus/test"; import { Atom, AtomRegistry } from "effect/reactivity"; -import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { EnvironmentId, ThreadId, type ScopedThreadRef } from "@t3tools/contracts"; vi.mock("./atom-registry", () => ({ appAtomRegistry: AtomRegistry.make() })); vi.mock("./projects", () => ({ environmentProjects: {} })); @@ -18,11 +18,15 @@ vi.mock("../native/ProjectChipIconImages", () => ({ })); vi.mock("./threads", () => { const shell = Atom.make<{ title: string } | null>(null); + const absent = Atom.make(null); return { environmentThreadDetails: {}, - environmentThreadShells: { threadShellAtom: () => shell }, + environmentThreadShells: { + threadShellAtom: (ref: ScopedThreadRef) => (ref.threadId === childId ? shell : absent), + }, threadFamilies: { - mount: (registry: AtomRegistry.AtomRegistry) => { + mount: (registry: AtomRegistry.AtomRegistry, _env: EnvironmentId, id: ThreadId) => { + if (id !== childId) return () => {}; registry.set(shell, { title: "Child result" }); return () => registry.set(shell, null); }, @@ -31,9 +35,11 @@ vi.mock("./threads", () => { }); import { appAtomRegistry } from "./atom-registry"; -import { useLiveThreadLinkLabels } from "./entities"; +import { resolveThreadLink, useLiveThreadLinkLabels } from "./entities"; import { environmentThreadShells } from "./threads"; +const childId = ThreadId.make("thread:delegated-task:command%3Amcp%3Achild"); + it("keeps a linked child title mounted until the message stops citing it", async () => { (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; const container = document.createElement("div"); @@ -47,11 +53,15 @@ it("keeps a linked child title mounted until the message stops citing it", async await act(async () => root.render( - + `} /> , ), ); - expect(container.textContent).toBe("[Child result](t3-thread://v2/local/child)"); + expect(container.textContent).toBe(`[Child result](t3-thread://v2/local/${childId})`); + const target = { environmentId: EnvironmentId.make("local"), threadId: childId }; + expect( + resolveThreadLink((atom) => appAtomRegistry.get(atom), target, target.environmentId), + ).toEqual(target); expect(renders).toBeLessThan(20); await act(async () => root.render( @@ -65,7 +75,7 @@ it("keeps a linked child title mounted until the message stops citing it", async appAtomRegistry.get( environmentThreadShells.threadShellAtom({ environmentId: EnvironmentId.make("local"), - threadId: ThreadId.make("child"), + threadId: childId, }), ), ).toBeNull(); diff --git a/apps/mobile/src/state/entities.ts b/apps/mobile/src/state/entities.ts index 799d92cdf70cc..e01611b5c596f 100644 --- a/apps/mobile/src/state/entities.ts +++ b/apps/mobile/src/state/entities.ts @@ -210,19 +210,17 @@ export function useSideChats(ref: ScopedThreadRef) { export function useLiveThreadLinkLabels(markdown: string, environmentId: EnvironmentId): string { const refs = hasThreadLinks(markdown) ? threadLinkRefs(markdown, environmentId) : []; + const refsKey = JSON.stringify(refs.map(threadKey)); const titles = useAtomValue( - refs.length === 0 - ? EMPTY_THREAD_LINK_TITLES_ATOM - : threadLinkTitlesAtom(JSON.stringify(refs.map(threadKey))), + refs.length === 0 ? EMPTY_THREAD_LINK_TITLES_ATOM : threadLinkTitlesAtom(refsKey), ); - // Shells leave out idle subagent children and older settled threads, so a cited thread the - // shell lacks is read with its family until it lands; only a settled empty read proves it gone. - const refsKey = JSON.stringify(refs.map(threadKey)); - // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- keeps cited families mounted until the links change; releasing on title arrival removes that title again + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- keeps cited rows mounted for the message's lifetime, independently of the family data it loads useEffect(() => { - const releases = (JSON.parse(refsKey) as ReadonlyArray).map((key) => { + const releases = decodeThreadKeys(JSON.parse(refsKey)).map((key) => { const ref = scopeThreadRef(...decodeThreadKey(JSON.parse(key))); - return threadFamilies.mount(appAtomRegistry, ref.environmentId, ref.threadId); + return threadFamilies.mount(appAtomRegistry, ref.environmentId, ref.threadId, { + onlyWhenMissing: true, + }); }); return () => { for (const release of releases) release(); diff --git a/apps/web/src/components/chat/MarkdownThreadLink.browser.test.tsx b/apps/web/src/components/chat/MarkdownThreadLink.browser.test.tsx new file mode 100644 index 0000000000000..b1d54062bd133 --- /dev/null +++ b/apps/web/src/components/chat/MarkdownThreadLink.browser.test.tsx @@ -0,0 +1,52 @@ +import { RegistryContext, useAtomValue } from "@effect/atom-react"; +import { EnvironmentId, ThreadId, type ScopedThreadRef } from "@t3tools/contracts"; +import { Atom } from "effect/reactivity"; +import { expect, it, vi } from "vite-plus/test"; + +import { render } from "../../../test/render"; +import { appAtomRegistry } from "../../rpc/atomRegistry"; +import { threadFamilies } from "../../state/threads"; +import { MarkdownThreadLink } from "./MarkdownThreadLink"; +import "../../index.css"; + +const environmentId = EnvironmentId.make("local"); +const threadId = ThreadId.make("thread:delegated-task:command%3Amcp%3Achild"); +const child = { environmentId, id: threadId, title: "Code store foundation" }; +const familyLoaded = Atom.make(false); + +vi.mock("../../state/entities", async (original) => ({ + ...(await original()), + useThreadShell: (ref: ScopedThreadRef | null) => { + const loaded = useAtomValue(familyLoaded); + return loaded && ref?.environmentId === environmentId && ref.threadId === threadId + ? child + : null; + }, + useThreadRefs: () => [], + useProject: () => null, +})); + +it("opens a cited idle child's literal ID and keeps its title until the citation unmounts", async () => { + using _family = vi.spyOn(threadFamilies, "mount").mockImplementation((registry, env, id) => { + if (env !== environmentId || id !== threadId) return () => {}; + registry.set(familyLoaded, true); + return () => registry.set(familyLoaded, false); + }); + await using screen = await render( + +

+ The repairs are verified in{" "} + + . +

+ , + ); + const link = screen.getByRole("link", { name: "Thread, Code store foundation" }); + await expect.element(link).toHaveAttribute("href", `/local/${encodeURIComponent(threadId)}`); + await screen.unmount(); + expect(appAtomRegistry.get(familyLoaded)).toBe(false); +}); diff --git a/apps/web/src/components/chat/MarkdownThreadLink.tsx b/apps/web/src/components/chat/MarkdownThreadLink.tsx index b5d3bbb5ac782..cc90b5f995ecc 100644 --- a/apps/web/src/components/chat/MarkdownThreadLink.tsx +++ b/apps/web/src/components/chat/MarkdownThreadLink.tsx @@ -29,17 +29,17 @@ export function MarkdownThreadLink(props: { written === null && decodedId !== null ? scopeThreadRef(environmentId, decodedId) : null, ); const local = written ?? decoded; - // Shells leave out idle subagent children and older settled threads, so a cited thread the - // shell lacks is read with its family until it lands; only a settled empty read proves it gone. - const missingThreadId = local === null ? (decodedId ?? props.link.threadId) : null; - // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- mounts the family read for a cited thread the shell omits while this link shows it - useEffect( - () => - missingThreadId === null - ? undefined - : threadFamilies.mount(appAtomRegistry, environmentId, missingThreadId), - [environmentId, missingThreadId], - ); + const threadId = props.link.threadId; + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- keeps cited rows mounted for the link's lifetime, independently of the family data it loads + useEffect(() => { + const ids = decodedId === null ? [threadId] : [threadId, decodedId]; + const releases = ids.map((id) => + threadFamilies.mount(appAtomRegistry, environmentId, id, { onlyWhenMissing: true }), + ); + return () => { + for (const release of releases) release(); + }; + }, [environmentId, threadId, decodedId]); const guess = local === null && props.link.environmentId === null; const refs = useThreadRefs(guess); const remote = useThreadShell(guess ? guessThreadLinkRef(props.link, refs) : null); diff --git a/packages/client-runtime/src/state/threadFamilies.test.ts b/packages/client-runtime/src/state/threadFamilies.test.ts index b08656975dbd9..df30aef139b2e 100644 --- a/packages/client-runtime/src/state/threadFamilies.test.ts +++ b/packages/client-runtime/src/state/threadFamilies.test.ts @@ -125,8 +125,12 @@ describe("createThreadFamilyAtoms", () => { yield* Queue.unbounded< Effect.Effect >(); + const requests: ThreadId[] = []; const client = { - [ORCHESTRATION_V2_WS_METHODS.getThreadFamily]: () => Effect.flatten(Queue.take(replies)), + [ORCHESTRATION_V2_WS_METHODS.getThreadFamily]: ({ threadId }: { threadId: ThreadId }) => + Effect.sync(() => requests.push(threadId)).pipe( + Effect.andThen(Effect.flatten(Queue.take(replies))), + ), } as unknown as WsRpcProtocolClient; const session = { client, @@ -231,6 +235,91 @@ describe("createThreadFamilyAtoms", () => { yield* settle; expect(threadFamilyRead(registry.get(family))).toBe("settled"); expect(mergedIds()).toEqual(["parent", "idle-child"]); + + const visible = row("visible-citation"); + const omitted = row("omitted-citation"); + registry.set( + shellState, + AsyncResult.success({ + ...shellAt(2), + snapshot: Option.some({ ...shellSnapshot, threads: [parent, visible] }), + }), + ); + const releaseVisible = families.mount(registry, TARGET.environmentId, visible.id, { + onlyWhenMissing: true, + }); + const releaseOmitted = families.mount(registry, TARGET.environmentId, omitted.id, { + onlyWhenMissing: true, + }); + yield* Queue.offerAll(replies, [ + Effect.succeed({ snapshotSequence: 5, threads: [omitted] }), + Effect.succeed({ snapshotSequence: 5, threads: [visible] }), + ]); + yield* AtomRegistry.getResult( + registry, + families.familyAtom(TARGET.environmentId, omitted.id), + { suspendOnWaiting: true }, + ); + expect(requests).toEqual([parent.id, parent.id, parent.id, omitted.id]); + expect(mergedIds()).toEqual([ + "parent", + "visible-citation", + "idle-child", + "omitted-citation", + ]); + + registry.set(shellState, AsyncResult.success(shellAt(2))); + yield* AtomRegistry.getResult( + registry, + families.familyAtom(TARGET.environmentId, visible.id), + { suspendOnWaiting: true }, + ); + expect(requests).toEqual([parent.id, parent.id, parent.id, omitted.id, visible.id]); + expect(mergedIds()).toEqual([ + "parent", + "idle-child", + "visible-citation", + "omitted-citation", + ]); + releaseVisible(); + releaseOmitted(); + expect(mergedIds()).toEqual(["parent", "idle-child"]); + + const unavailable = row("cached-failure"); + const unavailableFamily = families.familyAtom(TARGET.environmentId, unavailable.id); + const releaseFailed = families.mount(registry, TARGET.environmentId, unavailable.id); + yield* Queue.offer( + replies, + Effect.fail(new OrchestrationV2GetShellSnapshotError({ message: "read failed" })), + ); + yield* AtomRegistry.getResult(registry, unavailableFamily, { + suspendOnWaiting: true, + }).pipe(Effect.ignore); + expect(threadFamilyRead(registry.get(unavailableFamily))).toBe("failed"); + releaseFailed(); + registry.set( + shellState, + AsyncResult.success({ + ...shellAt(2), + snapshot: Option.some({ ...shellSnapshot, threads: [parent, unavailable] }), + }), + ); + const releaseRetry = families.mount(registry, TARGET.environmentId, unavailable.id, { + onlyWhenMissing: true, + }); + expect(requests.filter((id) => id === unavailable.id)).toHaveLength(1); + registry.set(shellState, AsyncResult.success(shellAt(2))); + yield* settle; + expect(threadFamilyRead(registry.get(unavailableFamily))).toBe("pending"); + yield* Queue.offer( + replies, + Effect.succeed({ snapshotSequence: 5, threads: [unavailable] }), + ); + yield* AtomRegistry.getResult(registry, unavailableFamily, { suspendOnWaiting: true }); + expect(requests.filter((id) => id === unavailable.id)).toHaveLength(2); + expect(mergedIds()).toEqual(["parent", "idle-child", "cached-failure"]); + releaseRetry(); + expect(mergedIds()).toEqual(["parent", "idle-child"]); }), ), ); diff --git a/packages/client-runtime/src/state/threadFamilies.ts b/packages/client-runtime/src/state/threadFamilies.ts index 0f1bb8d9cd274..7a0f62f75a6de 100644 --- a/packages/client-runtime/src/state/threadFamilies.ts +++ b/packages/client-runtime/src/state/threadFamilies.ts @@ -151,6 +151,26 @@ export function createThreadFamilyAtoms( }).pipe(Atom.withLabel(`environment-thread-family-rows:${key}`)), ); + const shellThreadIdsAtom = Atom.family((environmentId: EnvironmentId) => { + let previous: ReadonlyArray | undefined; + let ids: ReadonlySet = new Set(); + return Atom.make((get) => { + const state = shellStateValue(get(shellStateAtom(environmentId))); + const threads = state === null ? undefined : Option.getOrNull(state.snapshot)?.threads; + if (threads !== previous) { + previous = threads; + ids = new Set(threads?.map(({ id }) => id)); + } + return ids; + }).pipe(Atom.withLabel(`environment-shell-thread-ids:${environmentId}`)); + }); + const missingFamilyRowsAtom = Atom.family((key: string) => { + const [environmentId, threadId] = JSON.parse(key) as [EnvironmentId, ThreadId]; + return Atom.make((get) => + get(shellThreadIdsAtom(environmentId)).has(threadId) ? EMPTY_READS : get(familyRowsAtom(key)), + ).pipe(Atom.withLabel(`environment-missing-thread-family-rows:${key}`)); + }); + /** * Merges `rows` behind the shell rows until the returned release runs. Reads read under `key` * share one entry; each read's `snapshotSequence` decides which removals drop its rows. @@ -186,14 +206,41 @@ export function createThreadFamilyAtoms( registry: AtomRegistry.AtomRegistry, environmentId: EnvironmentId, threadId: ThreadId, + options: { readonly onlyWhenMissing?: boolean } = {}, ): (() => void) => { const key = familyKey(environmentId, threadId); - const release = mountRows(registry, environmentId, `family:${key}`, familyRowsAtom(key)); - // Mounting again after a failure retries instead of keeping the failure until reconnect. - if (threadFamilyRead(registry.get(familyAtom(key))) === "failed") { - refresh(registry, environmentId, threadId); + const onlyWhenMissing = options.onlyWhenMissing === true; + const release = mountRows( + registry, + environmentId, + `${onlyWhenMissing ? "missing-family" : "family"}:${key}`, + onlyWhenMissing ? missingFamilyRowsAtom(key) : familyRowsAtom(key), + ); + const retryFailed = () => { + if (threadFamilyRead(registry.get(familyAtom(key))) === "failed") { + refresh(registry, environmentId, threadId); + } + }; + if (!onlyWhenMissing) { + retryFailed(); + return release; } - return release; + let wasMissing = false; + // A cached failure must retry when the shell first omits a still-mounted citation. + const unsubscribe = registry.subscribe( + shellThreadIdsAtom(environmentId), + (ids) => { + const missing = !ids.has(threadId); + const becameMissing = missing && !wasMissing; + wasMissing = missing; + if (becameMissing) retryFailed(); + }, + { immediate: true }, + ); + return () => { + unsubscribe(); + release(); + }; }; const snapshotAtom = Atom.family((environmentId: EnvironmentId) => {