diff --git a/apps/mobile/src/features/home/HomeScreen.tsx b/apps/mobile/src/features/home/HomeScreen.tsx index d54fb7d4890..da31775912c 100644 --- a/apps/mobile/src/features/home/HomeScreen.tsx +++ b/apps/mobile/src/features/home/HomeScreen.tsx @@ -11,6 +11,10 @@ import { threadSearchMatchKey, type EnvironmentThreadSearchMatch, } from "@t3tools/client-runtime/state/thread-search"; +import { + type ChangeRequestSettlementState, + updateChangeRequestSettlementState, +} from "@t3tools/client-runtime/state/thread-settled"; import type { EnvironmentId, SidebarProjectGroupingMode, @@ -42,6 +46,7 @@ import { ThreadListShowMoreRow, } from "../threads/thread-list-items"; import { + ThreadListV2ChangeRequestLookupPool, ThreadListV2PendingRow, ThreadListV2Row, ThreadListV2SettledShelfHeader, @@ -478,23 +483,16 @@ export function HomeScreen(props: HomeScreenProps) { // Settled threads stay in the live shell stream (settled ≠ archived), so // the partition works directly off live shells — no snapshot merging or // optimistic holds. - // PR states stream in per-row (rows own the VCS subscriptions); a merged or - // closed PR auto-settles its thread on the next partition (mirrors web). + // PR states stream independently of virtualized rows so every branch-backed + // thread can reach a definitive state before the partition settles it. const [changeRequestStateByKey, setChangeRequestStateByKey] = useState< - ReadonlyMap + ReadonlyMap >(() => new Map()); const handleChangeRequestState = useCallback( - (threadKey: string, state: "open" | "closed" | "merged" | null) => { - setChangeRequestStateByKey((current) => { - if ((current.get(threadKey) ?? null) === state) return current; - const next = new Map(current); - if (state === null) { - next.delete(threadKey); - } else { - next.set(threadKey, state); - } - return next; - }); + (stateKey: string, state: ChangeRequestSettlementState) => { + setChangeRequestStateByKey((current) => + updateChangeRequestSettlementState(current, stateKey, state), + ); }, [], ); @@ -1047,6 +1045,14 @@ export function HomeScreen(props: HomeScreenProps) { if (threadListV2Enabled) { return ( + + ReadonlyMap >(() => new Map()); const handleChangeRequestState = useCallback( - (threadKey: string, state: "open" | "closed" | "merged" | null) => { - setChangeRequestStateByKey((current) => { - if ((current.get(threadKey) ?? null) === state) return current; - const next = new Map(current); - if (state === null) { - next.delete(threadKey); - } else { - next.set(threadKey, state); - } - return next; - }); + (stateKey: string, state: ChangeRequestSettlementState) => { + setChangeRequestStateByKey((current) => + updateChangeRequestSettlementState(current, stateKey, state), + ); }, [], ); @@ -1127,6 +1125,16 @@ function ThreadNavigationSidebarPane( : "No threads yet"} ); + const changeRequestLookupPool = threadListV2Enabled ? ( + + ) : null; if (props.nativeChrome) { return ( @@ -1155,6 +1163,7 @@ function ThreadNavigationSidebarPane( }} /> + {changeRequestLookupPool} + {changeRequestLookupPool} diff --git a/apps/mobile/src/features/threads/thread-list-v2-items.tsx b/apps/mobile/src/features/threads/thread-list-v2-items.tsx index 8d6874c7855..5fb61b3fe11 100644 --- a/apps/mobile/src/features/threads/thread-list-v2-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-v2-items.tsx @@ -3,7 +3,15 @@ import type { EnvironmentThreadShell, } from "@t3tools/client-runtime/state/shell"; import type { EnvironmentThreadSearchMatch } from "@t3tools/client-runtime/state/thread-search"; -import { canSnooze, resolveSnoozePresets } from "@t3tools/client-runtime/state/thread-settled"; +import { + canSnooze, + type ChangeRequestSettlementState, + resolveChangeRequestSettlementState, + resolveSnoozePresets, + selectChangeRequestLookupWindow, + threadChangeRequestStateKey, +} from "@t3tools/client-runtime/state/thread-settled"; +import type { EnvironmentId, ProjectId } from "@t3tools/contracts"; import type { MenuAction } from "@react-native-menu/menu"; import { memo, useCallback, useEffect, useMemo, useState, type ComponentProps } from "react"; import { @@ -25,13 +33,17 @@ import { cn } from "../../lib/cn"; import { relativeTime } from "../../lib/time"; import { useThemeColor } from "../../lib/useThemeColor"; import type { PendingNewTask } from "../../state/use-pending-new-tasks"; -import { useThreadPr } from "../../state/use-thread-pr"; +import { useThreadPrLookup, useThreadVcsStatus } from "../../state/use-thread-pr"; import { ThreadSwipeable } from "../home/thread-swipe-actions"; import { + buildThreadListV2ChangeRequestLookupTargets, resolveThreadListV2SnoozeMenuSelection, resolveThreadListV2SnoozeGateExpiryMs, resolveThreadListV2Status, resolveThreadListV2SwipeActions, + THREAD_LIST_V2_CHANGE_REQUEST_LOOKUP_LIMIT, + THREAD_LIST_V2_CHANGE_REQUEST_LOOKUP_WINDOW_MS, + type ThreadListV2ChangeRequestLookupTarget, type ThreadListV2Status, } from "./threadListV2"; import { ThreadSearchMatchExcerpt } from "./thread-search-match"; @@ -300,6 +312,89 @@ export const ThreadListV2PendingRow = memo(function ThreadListV2PendingRow(props ); }); +const ThreadListV2ChangeRequestLookupReporter = memo( + function ThreadListV2ChangeRequestLookupReporter(props: { + readonly target: ThreadListV2ChangeRequestLookupTarget; + readonly onChangeRequestState: (stateKey: string, state: ChangeRequestSettlementState) => void; + }) { + const { target, onChangeRequestState } = props; + const gitStatus = useThreadVcsStatus(target.environmentId, target.cwd); + useEffect(() => { + for (const thread of target.threads) { + onChangeRequestState( + threadChangeRequestStateKey(thread), + resolveChangeRequestSettlementState({ + threadBranch: thread.branch, + gitStatus: gitStatus.data, + gitStatusError: gitStatus.error, + }), + ); + } + }, [gitStatus.data, gitStatus.error, onChangeRequestState, target.threads]); + return null; + }, +); + +export const ThreadListV2ChangeRequestLookupPool = memo( + function ThreadListV2ChangeRequestLookupPool(props: { + readonly threads: ReadonlyArray; + readonly environmentId: EnvironmentId | null; + readonly projectRefs?: ReadonlyArray<{ + readonly environmentId: EnvironmentId; + readonly projectId: ProjectId; + }> | null; + readonly projectCwdByKey: ReadonlyMap; + readonly settlementEnvironmentIds: ReadonlySet; + readonly onChangeRequestState: (stateKey: string, state: ChangeRequestSettlementState) => void; + }) { + const { + threads, + environmentId, + projectRefs, + projectCwdByKey, + settlementEnvironmentIds, + onChangeRequestState, + } = props; + const targets = useMemo( + () => + buildThreadListV2ChangeRequestLookupTargets({ + threads, + environmentId, + projectRefs, + projectCwdByKey, + settlementEnvironmentIds, + }), + [environmentId, projectCwdByKey, projectRefs, settlementEnvironmentIds, threads], + ); + const [windowIndex, setWindowIndex] = useState(0); + useEffect(() => { + if (targets.length <= THREAD_LIST_V2_CHANGE_REQUEST_LOOKUP_LIMIT) return; + const interval = setInterval( + () => setWindowIndex((current) => current + 1), + THREAD_LIST_V2_CHANGE_REQUEST_LOOKUP_WINDOW_MS, + ); + return () => clearInterval(interval); + }, [targets.length]); + const activeTargets = useMemo( + () => + selectChangeRequestLookupWindow({ + targets, + windowIndex, + limit: THREAD_LIST_V2_CHANGE_REQUEST_LOOKUP_LIMIT, + }), + [targets, windowIndex], + ); + + return activeTargets.map((target) => ( + + )); + }, +); + export const ThreadListV2Row = memo(function ThreadListV2Row(props: { readonly thread: EnvironmentThreadShell; readonly variant: "card" | "slim"; @@ -344,12 +439,7 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { readonly snoozeSupported: boolean; readonly onSwipeableWillOpen: (methods: SwipeableMethods) => void; readonly onSwipeableClose: (methods: SwipeableMethods) => void; - /** Reports this row's live PR state up so the partition can auto-settle - merged/closed work (mirrors web's onChangeRequestState). */ - readonly onChangeRequestState?: ( - threadKey: string, - state: "open" | "closed" | "merged" | null, - ) => void; + readonly onChangeRequestState?: (stateKey: string, state: ChangeRequestSettlementState) => void; readonly projectCwd?: string | null; readonly searchMatch?: EnvironmentThreadSearchMatch; readonly searchQuery?: string; @@ -372,12 +462,14 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { } = props; const snoozedRow = props.snoozed === true; - const pr = useThreadPr(thread, props.projectCwd ?? props.project?.workspaceRoot ?? null); - const prState = pr?.state ?? null; - const threadKey = `${thread.environmentId}:${thread.id}`; + const { changeRequestState, pr } = useThreadPrLookup( + thread, + props.projectCwd ?? props.project?.workspaceRoot ?? null, + ); + const changeRequestStateKey = threadChangeRequestStateKey(thread); useEffect(() => { - onChangeRequestState?.(threadKey, prState); - }, [onChangeRequestState, prState, threadKey]); + onChangeRequestState?.(changeRequestStateKey, changeRequestState); + }, [changeRequestState, changeRequestStateKey, onChangeRequestState]); const screenColor = useThemeColor("--color-screen"); const drawerColor = useThemeColor("--color-drawer"); diff --git a/apps/mobile/src/features/threads/threadListV2.test.ts b/apps/mobile/src/features/threads/threadListV2.test.ts index 99b5700f7b0..3ff07496ab1 100644 --- a/apps/mobile/src/features/threads/threadListV2.test.ts +++ b/apps/mobile/src/features/threads/threadListV2.test.ts @@ -1,6 +1,10 @@ import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell"; import { threadSearchMatchKey } from "@t3tools/client-runtime/state/thread-search"; -import { resolveSnoozePresets } from "@t3tools/client-runtime/state/thread-settled"; +import { + resolveSnoozePresets, + selectChangeRequestLookupWindow, + threadChangeRequestStateKey, +} from "@t3tools/client-runtime/state/thread-settled"; import { CommandId, EnvironmentId, @@ -14,6 +18,7 @@ import { describe, expect, it } from "vite-plus/test"; import type { PendingNewTask } from "../../state/use-pending-new-tasks"; import { + buildThreadListV2ChangeRequestLookupTargets, buildThreadListV2Items, buildThreadListV2ListItems, resolveThreadListV2Enabled, @@ -457,6 +462,120 @@ describe("buildThreadListV2Items", () => { expect(layout.settledShelfHeaderIndex).toBe(1); }); + it("keeps stale branch threads active until their PR state is known", () => { + const thread = makeThread({ + id: ThreadId.make("unknown-pr"), + title: "Unknown PR", + branch: "feature/unknown-pr", + latestUserMessageAt: "2026-05-01T00:00:00.000Z", + }); + const unresolved = buildThreadListV2Items({ + threads: [thread], + environmentId: null, + searchQuery: "", + now: NOW, + }); + + expect(unresolved.items.map((item) => [item.thread.id, item.variant])).toEqual([ + ["unknown-pr", "card"], + ]); + expect(unresolved.settledCount).toBe(0); + + const confirmedNoPr = buildThreadListV2Items({ + threads: [thread], + environmentId: null, + searchQuery: "", + changeRequestStateByKey: new Map([[threadChangeRequestStateKey(thread), "none"]]), + now: NOW, + }); + + expect(confirmedNoPr.items.map((item) => [item.thread.id, item.variant])).toEqual([ + ["unknown-pr", "slim"], + ]); + expect(confirmedNoPr.settledCount).toBe(1); + }); + + it("does not reuse a PR state after a thread branch changes", () => { + const previous = makeThread({ + id: ThreadId.make("changed-branch"), + title: "Changed branch", + branch: "feature/previous", + latestUserMessageAt: "2026-05-01T00:00:00.000Z", + }); + const current = { ...previous, branch: "feature/current" }; + const layout = buildThreadListV2Items({ + threads: [current], + environmentId: null, + searchQuery: "", + changeRequestStateByKey: new Map([[threadChangeRequestStateKey(previous), "none"]]), + now: NOW, + }); + + expect(threadChangeRequestStateKey(current)).not.toBe(threadChangeRequestStateKey(previous)); + expect(layout.items.map((item) => [item.thread.id, item.variant])).toEqual([ + ["changed-branch", "card"], + ]); + expect(layout.settledCount).toBe(0); + }); + + it("bounds and fairly advances off-screen PR lookup targets", () => { + const threads = Array.from({ length: 18 }, (_, index) => + makeThread({ + id: ThreadId.make(`lookup-${index}`), + title: `Lookup ${index}`, + branch: `feature/lookup-${index}`, + worktreePath: `/repo/worktrees/${index}`, + }), + ); + const targets = buildThreadListV2ChangeRequestLookupTargets({ + threads, + environmentId: null, + projectCwdByKey: new Map(), + settlementEnvironmentIds: new Set([environmentId]), + }); + const firstBatch = selectChangeRequestLookupWindow({ + targets, + windowIndex: 0, + limit: 16, + }); + + expect(firstBatch).toHaveLength(16); + const secondBatch = selectChangeRequestLookupWindow({ + targets, + windowIndex: 1, + limit: 16, + }); + expect(secondBatch.slice(0, 2).map((target) => target.key)).toEqual( + targets.slice(16).map((target) => target.key), + ); + expect(new Set([...firstBatch, ...secondBatch].map((target) => target.key)).size).toBe(18); + }); + + it("deduplicates off-screen PR lookups by environment and cwd", () => { + const targets = buildThreadListV2ChangeRequestLookupTargets({ + threads: [ + makeThread({ + id: ThreadId.make("shared-cwd-a"), + title: "Shared cwd A", + branch: "feature/shared-a", + worktreePath: "/repo/shared", + }), + makeThread({ + id: ThreadId.make("shared-cwd-b"), + title: "Shared cwd B", + branch: "feature/shared-b", + worktreePath: "/repo/shared", + }), + ], + environmentId: null, + projectCwdByKey: new Map(), + settlementEnvironmentIds: new Set([environmentId]), + }); + + expect(targets).toHaveLength(1); + expect(targets[0]?.threads).toHaveLength(2); + }); + it("collapses settled threads to a counted shelf header", () => { const layout = buildThreadListV2Items({ threads: [ diff --git a/apps/mobile/src/features/threads/threadListV2.ts b/apps/mobile/src/features/threads/threadListV2.ts index c88aff4ec02..3d7b2aa27c9 100644 --- a/apps/mobile/src/features/threads/threadListV2.ts +++ b/apps/mobile/src/features/threads/threadListV2.ts @@ -5,8 +5,12 @@ import { QUEUED_TURN_START_GRACE_MS, resolveSnoozePresets, snoozeWakeLabel, + threadChangeRequestStateKey, +} from "@t3tools/client-runtime/state/thread-settled"; +import type { + ChangeRequestSettlementState, + SnoozePreset, } from "@t3tools/client-runtime/state/thread-settled"; -import type { SnoozePreset } from "@t3tools/client-runtime/state/thread-settled"; import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell"; import { threadSearchMatchKey } from "@t3tools/client-runtime/state/thread-search"; import type { EnvironmentId, ProjectId } from "@t3tools/contracts"; @@ -25,6 +29,8 @@ export { snoozeWakeLabel }; */ export type ThreadListV2Status = "approval" | "input" | "working" | "failed" | "ready"; export type ThreadListV2SwipeAction = "archive" | "settle" | "unsettle" | "snooze" | "unsnooze"; +export const THREAD_LIST_V2_CHANGE_REQUEST_LOOKUP_LIMIT = 16; +export const THREAD_LIST_V2_CHANGE_REQUEST_LOOKUP_WINDOW_MS = 30_000; export function resolveThreadListV2SnoozeMenuSelection(input: { readonly event: string; @@ -175,6 +181,68 @@ export function sortThreadsForListV2; +} + +export function threadChangeRequestLookupTargetKey( + environmentId: EnvironmentId, + cwd: string, +): string { + return JSON.stringify([environmentId, cwd]); +} + +export function buildThreadListV2ChangeRequestLookupTargets(input: { + readonly threads: ReadonlyArray; + readonly environmentId: EnvironmentId | null; + readonly projectRefs?: ReadonlyArray<{ + readonly environmentId: EnvironmentId; + readonly projectId: ProjectId; + }> | null; + readonly projectCwdByKey: ReadonlyMap; + readonly settlementEnvironmentIds: ReadonlySet; +}): ReadonlyArray { + const projectKeys = input.projectRefs + ? new Set(input.projectRefs.map((ref) => `${ref.environmentId}:${ref.projectId}`)) + : null; + const groups = new Map< + string, + { + environmentId: EnvironmentId; + cwd: string; + threads: EnvironmentThreadShell[]; + } + >(); + + // Keep confirmed states in the rotation: a branch can gain a PR later, and + // a closed PR can reopen while its thread remains off-screen. + for (const thread of sortThreadsForListV2(input.threads)) { + if (thread.archivedAt !== null || thread.branch === null) continue; + if (!input.settlementEnvironmentIds.has(thread.environmentId)) continue; + if (input.environmentId !== null && thread.environmentId !== input.environmentId) continue; + const projectKey = `${thread.environmentId}:${thread.projectId}`; + if (projectKeys !== null && !projectKeys.has(projectKey)) continue; + const cwd = thread.worktreePath ?? input.projectCwdByKey.get(projectKey) ?? null; + if (cwd === null) continue; + const key = threadChangeRequestLookupTargetKey(thread.environmentId, cwd); + const current = groups.get(key); + if (current) { + current.threads.push(thread); + } else { + groups.set(key, { + environmentId: thread.environmentId, + cwd, + threads: [thread], + }); + } + } + + return [...groups].map(([key, group]) => ({ key, ...group })); +} + export interface ThreadListV2Item { readonly thread: EnvironmentThreadShell; readonly variant: "card" | "slim"; @@ -316,8 +384,8 @@ export function buildThreadListV2Items(input: { }> | null; readonly searchQuery: string; readonly matchedThreadKeys?: ReadonlySet; - /** Per-row PR state reported up by visible rows ("env:threadId" keys). */ - readonly changeRequestStateByKey?: ReadonlyMap; + /** PR state reported by branch-aware lookup reporters. */ + readonly changeRequestStateByKey?: ReadonlyMap; /** Environments whose server supports thread.settle/unsettle. Threads on other environments never classify as settled — the user could neither un-settle nor pin them. Absent = no gating (tests). */ @@ -377,7 +445,9 @@ export function buildThreadListV2Items(input: { const supportsSettlement = input.settlementEnvironmentIds?.has(thread.environmentId) ?? true; const supportsSnooze = input.snoozeEnvironmentIds?.has(thread.environmentId) ?? true; const changeRequestState = - input.changeRequestStateByKey?.get(`${thread.environmentId}:${thread.id}`) ?? null; + thread.branch === null + ? "none" + : (input.changeRequestStateByKey?.get(threadChangeRequestStateKey(thread)) ?? "unknown"); // Visibility parity with web: a snoozed thread leaves the list until it // wakes (or raises its hand — effectiveSnoozed refuses blocked/failed // work). Snooze outranks settled classification, same as web. diff --git a/apps/mobile/src/state/use-thread-pr.ts b/apps/mobile/src/state/use-thread-pr.ts index a3440cd4848..b79b60535f1 100644 --- a/apps/mobile/src/state/use-thread-pr.ts +++ b/apps/mobile/src/state/use-thread-pr.ts @@ -1,8 +1,10 @@ import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell"; +import { resolveChangeRequestSettlementState } from "@t3tools/client-runtime/state/thread-settled"; +import type { EnvironmentId } from "@t3tools/contracts"; import { useEnvironmentQuery } from "./query"; import { presentThreadPr, type ThreadPrPresentation } from "./thread-pr-presentation"; -import { vcsEnvironment } from "./vcs"; +import { threadVcsEnvironment } from "./vcs"; export { presentThreadPr, @@ -10,32 +12,42 @@ export { type ThreadPrPresentation, } from "./thread-pr-presentation"; +export function useThreadVcsStatus(environmentId: EnvironmentId, cwd: string | null) { + return useEnvironmentQuery( + cwd === null + ? null + : threadVcsEnvironment.status({ + environmentId, + input: { cwd }, + }), + ); +} + /** * Live PR status for a thread's branch. Subscriptions are deduplicated per - * (environmentId, cwd) by the atom family, so many rows on the same worktree - * or project root share one stream — and virtualization means only visible - * rows subscribe at all. + * (environmentId, cwd) by the atom family, so visible rows and Thread List + * v2's bounded off-screen lookup pool share one stream per worktree/project. */ +export function useThreadPrLookup(thread: EnvironmentThreadShell, projectCwd: string | null) { + const cwd = thread.worktreePath ?? projectCwd; + const gitStatus = useThreadVcsStatus(thread.environmentId, thread.branch === null ? null : cwd); + + const status = gitStatus.data; + const changeRequestState = resolveChangeRequestSettlementState({ + threadBranch: thread.branch, + gitStatus: status, + gitStatusError: gitStatus.error, + }); + const pr = + status !== null && thread.branch !== null && status.refName === thread.branch && status.pr + ? presentThreadPr(status.pr, status.sourceControlProvider) + : null; + return { changeRequestState, pr }; +} + export function useThreadPr( thread: EnvironmentThreadShell, projectCwd: string | null, ): ThreadPrPresentation | null { - const cwd = thread.worktreePath ?? projectCwd; - const gitStatus = useEnvironmentQuery( - thread.branch !== null && cwd !== null - ? vcsEnvironment.status({ - environmentId: thread.environmentId, - input: { cwd }, - }) - : null, - ); - - const status = gitStatus.data; - if (status === null || thread.branch === null || status.refName !== thread.branch) { - return null; - } - if (!status.pr) { - return null; - } - return presentThreadPr(status.pr, status.sourceControlProvider); + return useThreadPrLookup(thread, projectCwd).pr; } diff --git a/apps/mobile/src/state/vcs.ts b/apps/mobile/src/state/vcs.ts index dc8c251149f..b0c9783b009 100644 --- a/apps/mobile/src/state/vcs.ts +++ b/apps/mobile/src/state/vcs.ts @@ -6,4 +6,10 @@ import { import { connectionAtomRuntime } from "../connection/runtime"; export const vcsEnvironment = createVcsEnvironmentAtoms(connectionAtomRuntime); +export const threadVcsEnvironment = createVcsEnvironmentAtoms(connectionAtomRuntime, { + // Thread List v2 time-slices off-screen PR lookups. Dispose each status + // stream as soon as its last row/reporter unmounts so the pool's cap also + // bounds server pollers without changing caching for other mobile Git UI. + statusIdleTtlMs: 0, +}); export const vcsActionManager = createVcsActionManager(connectionAtomRuntime); diff --git a/apps/server/src/git/GitManager.test.ts b/apps/server/src/git/GitManager.test.ts index 695c7b76f64..2c80fd185bf 100644 --- a/apps/server/src/git/GitManager.test.ts +++ b/apps/server/src/git/GitManager.test.ts @@ -1315,6 +1315,7 @@ it.layer(GitManagerTestLayer)("GitManager", (it) => { const status = yield* manager.status({ cwd: repoDir }); expect(status.refName).toBe("feature/status-no-gh"); expect(status.pr).toBeNull(); + expect(status.prLookupFailed).toBe(true); }), ); @@ -1356,6 +1357,7 @@ it.layer(GitManagerTestLayer)("GitManager", (it) => { yield* manager.invalidateStatus(repoDir); const second = yield* manager.status({ cwd: repoDir }); expect(second.pr?.number).toBe(214); + expect(second.prLookupFailed).toBe(true); }), ); diff --git a/apps/server/src/git/GitManager.ts b/apps/server/src/git/GitManager.ts index da002df5e6c..e1f680c8111 100644 --- a/apps/server/src/git/GitManager.ts +++ b/apps/server/src/git/GitManager.ts @@ -995,7 +995,7 @@ export const make = Effect.gen(function* () { }), ), ), - Effect.map(({ pr }) => pr), + Effect.map(({ pr }) => ({ pr, lookupFailed: false as const })), Effect.catch((error) => Effect.logWarning("PR lookup failed; keeping last known PR state.").pipe( Effect.annotateLogs({ @@ -1007,14 +1007,15 @@ export const make = Effect.gen(function* () { : typeof error, }), Effect.andThen(resolveBranchHeadContext(cwd, details)), - Effect.map((headContext) => - resolveLastKnownPr(branchKey, { + Effect.map((headContext) => ({ + pr: resolveLastKnownPr(branchKey, { upstreamRef: details.upstreamRef, headBranch: headContext.headBranch, remoteName: headContext.remoteName, headRemoteUrlKey: headContext.headRemoteUrlKey, }), - ), + lookupFailed: true as const, + })), ), ), ); @@ -1030,21 +1031,22 @@ export const make = Effect.gen(function* () { return null; } - const pr = + const prLookup = details.branch !== null ? yield* lookupStatusPr(cwd, { branch: details.branch, upstreamRef: details.upstreamRef, isDefaultBranch: details.isDefaultBranch, }) - : null; + : { pr: null, lookupFailed: false as const }; return { hasUpstream: details.hasUpstream, aheadCount: details.aheadCount, behindCount: details.behindCount, aheadOfDefaultCount: details.aheadOfDefaultCount, - pr, + pr: prLookup.pr, + ...(prLookup.lookupFailed ? { prLookupFailed: true } : {}), } satisfies VcsStatusRemoteResult; }); const remoteStatusResultCache = yield* Cache.makeWith((cwd: string) => readRemoteStatus(cwd), { diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 6820a29e2c8..10a1c7c9037 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -67,7 +67,7 @@ const baseStatus: VcsStatusResult = { ...baseRemoteStatus, }; -function makeTestLayer(state: { +interface VcsTestState { currentLocalStatus: VcsStatusLocalResult; currentRemoteStatus: VcsStatusRemoteResult | null; localStatusCalls: number; @@ -75,7 +75,27 @@ function makeTestLayer(state: { localInvalidationCalls: number; remoteInvalidationCalls: number; remoteStatusRefreshUpstreamValues?: Array; -}) { + failRemoteStatus?: boolean; + dieRemoteStatus?: boolean; + dieInvalidateStatus?: boolean; + blockRemoteStatusAtCall?: number; + remoteStatusStarted?: Deferred.Deferred | null; + remoteStatusRelease?: Deferred.Deferred | null; +} + +function makeTestState(overrides: Partial = {}): VcsTestState { + return { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + ...overrides, + }; +} + +function makeTestLayer(state: VcsTestState) { return VcsStatusBroadcaster.layer.pipe( Layer.provideMerge(NodeServices.layer), Layer.provide(makeBackgroundPolicyLayer(() => true)), @@ -87,9 +107,30 @@ function makeTestLayer(state: { return state.currentLocalStatus; }), remoteStatus: (_input, options) => - Effect.sync(() => { + Effect.gen(function* () { state.remoteStatusCalls += 1; state.remoteStatusRefreshUpstreamValues?.push(options?.refreshUpstream); + if ( + state.blockRemoteStatusAtCall !== undefined && + state.remoteStatusCalls === state.blockRemoteStatusAtCall && + state.remoteStatusStarted !== null && + state.remoteStatusStarted !== undefined && + state.remoteStatusRelease !== null && + state.remoteStatusRelease !== undefined + ) { + yield* Deferred.succeed(state.remoteStatusStarted, undefined); + yield* Deferred.await(state.remoteStatusRelease); + } + if (state.failRemoteStatus === true) { + return yield* new GitManagerError({ + operation: "VcsStatusBroadcaster.test", + cwd: "/repo", + detail: "remote status failed", + }); + } + if (state.dieRemoteStatus === true) { + return yield* Effect.die(new Error("remote status defect")); + } return state.currentRemoteStatus; }), invalidateLocalStatus: () => @@ -101,7 +142,10 @@ function makeTestLayer(state: { state.remoteInvalidationCalls += 1; }), invalidateStatus: () => - Effect.sync(() => { + Effect.gen(function* () { + if (state.dieInvalidateStatus === true) { + return yield* Effect.die(new Error("status invalidation defect")); + } state.localInvalidationCalls += 1; state.remoteInvalidationCalls += 1; }), @@ -208,7 +252,7 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(makeTestLayer(state))); }); - it.effect("keeps the cached snapshot unchanged when a refresh branch fails", () => { + it.effect("reloads remote status after a full refresh fails", () => { const state = { currentLocalStatus: baseLocalStatus, currentRemoteStatus: baseRemoteStatus, @@ -273,17 +317,22 @@ describe("VcsStatusBroadcaster", () => { state.failRemoteStatus = true; const refreshExit = yield* broadcaster.refreshStatus("/repo").pipe(Effect.exit); + state.failRemoteStatus = false; const cached = yield* broadcaster.getStatus({ cwd: "/repo" }); assert.isTrue(Exit.isFailure(refreshExit)); - assert.deepStrictEqual(cached, baseStatus); + assert.deepStrictEqual(cached, { + ...baseLocalStatus, + ...state.currentRemoteStatus, + }); + assert.equal(state.remoteStatusCalls, 3); }).pipe(Effect.provide(testLayer)); }); - it.effect("refreshes only the cached local snapshot when requested", () => { + it.effect("reloads remote status after the local branch changes", () => { const state = { currentLocalStatus: baseLocalStatus, - currentRemoteStatus: baseRemoteStatus, + currentRemoteStatus: remoteStatusWithPr, localStatusCalls: 0, remoteStatusCalls: 0, localInvalidationCalls: 0, @@ -299,18 +348,22 @@ describe("VcsStatusBroadcaster", () => { refName: "feature/local-only-refresh", hasWorkingTreeChanges: true, }; + state.currentRemoteStatus = baseRemoteStatus; const refreshedLocal = yield* broadcaster.refreshLocalStatus("/repo"); const cached = yield* broadcaster.getStatus({ cwd: "/repo" }); - assert.deepStrictEqual(initial, baseStatus); + assert.deepStrictEqual(initial, { + ...baseLocalStatus, + ...remoteStatusWithPr, + }); assert.deepStrictEqual(refreshedLocal, state.currentLocalStatus); assert.deepStrictEqual(cached, { ...state.currentLocalStatus, ...baseRemoteStatus, }); assert.equal(state.localStatusCalls, 2); - assert.equal(state.remoteStatusCalls, 1); + assert.equal(state.remoteStatusCalls, 2); assert.equal(state.localInvalidationCalls, 1); assert.equal(state.remoteInvalidationCalls, 0); }).pipe(Effect.provide(makeTestLayer(state))); @@ -410,10 +463,12 @@ describe("VcsStatusBroadcaster", () => { _tag: "snapshot", local: baseLocalStatus, remote: null, + remoteLoaded: false, } satisfies VcsStatusStreamEvent); assert.deepStrictEqual(remoteUpdated, { _tag: "remoteUpdated", remote: baseRemoteStatus, + remoteLoaded: true, } satisfies VcsStatusStreamEvent); }).pipe(Effect.provide(makeTestLayer(state))); }); @@ -457,10 +512,12 @@ describe("VcsStatusBroadcaster", () => { _tag: "snapshot", local: baseLocalStatus, remote: null, + remoteLoaded: false, } satisfies VcsStatusStreamEvent); assert.deepStrictEqual(remoteUpdated, { _tag: "remoteUpdated", remote: remoteStatusWithPr, + remoteLoaded: true, } satisfies VcsStatusStreamEvent); assert.equal(state.remoteStatusCalls, 1); assert.equal(state.remoteInvalidationCalls, 0); @@ -578,6 +635,7 @@ describe("VcsStatusBroadcaster", () => { assert.deepStrictEqual(remoteUpdated, { _tag: "remoteUpdated", remote: remoteStatusWithPr, + remoteLoaded: true, } satisfies VcsStatusStreamEvent); assert.equal(state.remoteStatusCalls, 2); assert.equal(state.remoteInvalidationCalls, 0); @@ -598,7 +656,7 @@ describe("VcsStatusBroadcaster", () => { it.effect("delays automatic refresh when a cached remote snapshot is available", () => { const state = { currentLocalStatus: baseLocalStatus, - currentRemoteStatus: baseRemoteStatus, + currentRemoteStatus: null, localStatusCalls: 0, remoteStatusCalls: 0, localInvalidationCalls: 0, @@ -621,7 +679,13 @@ describe("VcsStatusBroadcaster", () => { : Effect.void, ).pipe(Effect.forkIn(scope)); - yield* Deferred.await(snapshotDeferred); + const snapshot = yield* Deferred.await(snapshotDeferred); + assert.deepStrictEqual(snapshot, { + _tag: "snapshot", + local: baseLocalStatus, + remote: null, + remoteLoaded: true, + } satisfies VcsStatusStreamEvent); assert.equal(state.remoteStatusCalls, 1); assert.equal(state.remoteInvalidationCalls, 0); @@ -637,6 +701,330 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(Layer.merge(makeTestLayer(state), TestClock.layer()))); }); + it.effect("marks cached remote status unresolved across refresh failures", () => { + const state = makeTestState({ failRemoteStatus: false }); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const scope = yield* Scope.make(); + const snapshotDeferred = yield* Deferred.make(); + const unavailableDeferred = yield* Deferred.make(); + const recoveredDeferred = yield* Deferred.make(); + yield* Stream.runForEach( + broadcaster.streamStatus( + { cwd: "/repo" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.minutes(1)) }, + ), + (event) => { + if (event._tag === "snapshot") { + return Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore); + } + if (event._tag !== "remoteUpdated") return Effect.void; + return event.remoteLoaded === false + ? Deferred.succeed(unavailableDeferred, event).pipe(Effect.ignore) + : Deferred.succeed(recoveredDeferred, event).pipe(Effect.ignore); + }, + ).pipe(Effect.forkIn(scope)); + + yield* Deferred.await(snapshotDeferred); + state.failRemoteStatus = true; + yield* TestClock.adjust(Duration.minutes(1)); + yield* Effect.yieldNow; + const unavailable = yield* Deferred.poll(unavailableDeferred); + assert.isTrue(Option.isSome(unavailable)); + if (Option.isSome(unavailable)) { + assert.deepStrictEqual(yield* Deferred.await(unavailableDeferred), { + _tag: "remoteUpdated", + remote: baseRemoteStatus, + remoteLoaded: false, + } satisfies VcsStatusStreamEvent); + } + + state.failRemoteStatus = false; + yield* TestClock.adjust(Duration.minutes(1)); + yield* Effect.yieldNow; + const recovered = yield* Deferred.poll(recoveredDeferred); + assert.isTrue(Option.isSome(recovered)); + if (Option.isSome(recovered)) { + assert.deepStrictEqual(yield* Deferred.await(recoveredDeferred), { + _tag: "remoteUpdated", + remote: baseRemoteStatus, + remoteLoaded: true, + } satisfies VcsStatusStreamEvent); + } + + yield* Scope.close(scope, Exit.void); + }).pipe(Effect.provide(Layer.merge(makeTestLayer(state), TestClock.layer()))); + }); + + it.effect("marks cached remote status unresolved across refresh defects", () => { + const state = makeTestState({ dieRemoteStatus: false }); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const snapshotDeferred = yield* Deferred.make(); + const unavailableDeferred = yield* Deferred.make(); + yield* Stream.runForEach( + broadcaster.streamStatus( + { cwd: "/repo" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.minutes(1)) }, + ), + (event) => { + if (event._tag === "snapshot") { + return Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore); + } + return event._tag === "remoteUpdated" && event.remoteLoaded === false + ? Deferred.succeed(unavailableDeferred, event).pipe(Effect.ignore) + : Effect.void; + }, + ).pipe(Effect.forkScoped); + + yield* Deferred.await(snapshotDeferred); + state.dieRemoteStatus = true; + yield* TestClock.adjust(Duration.minutes(1)); + yield* Effect.yieldNow; + assert.deepStrictEqual(yield* Deferred.await(unavailableDeferred), { + _tag: "remoteUpdated", + remote: baseRemoteStatus, + remoteLoaded: false, + } satisfies VcsStatusStreamEvent); + }).pipe(Effect.provide(Layer.merge(makeTestLayer(state), TestClock.layer()))); + }); + + it.effect("keeps provider-level PR lookup fallbacks unresolved", () => { + const lookupFallback = { + ...baseRemoteStatus, + prLookupFailed: true, + } satisfies VcsStatusRemoteResult; + const state = makeTestState({ + currentRemoteStatus: lookupFallback as VcsStatusRemoteResult, + }); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const unavailableDeferred = yield* Deferred.make(); + const recoveredDeferred = yield* Deferred.make(); + yield* Stream.runForEach( + broadcaster.streamStatus( + { cwd: "/repo" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) }, + ), + (event) => { + if (event._tag !== "remoteUpdated") return Effect.void; + return event.remoteLoaded === false + ? Deferred.succeed(unavailableDeferred, event).pipe(Effect.ignore) + : Deferred.succeed(recoveredDeferred, event).pipe(Effect.ignore); + }, + ).pipe(Effect.forkScoped); + + assert.deepStrictEqual(yield* Deferred.await(unavailableDeferred), { + _tag: "remoteUpdated", + remote: lookupFallback, + remoteLoaded: false, + } satisfies VcsStatusStreamEvent); + assert.equal(state.remoteStatusCalls, 1); + + yield* TestClock.adjust(Duration.seconds(30)); + yield* Effect.yieldNow; + assert.equal(state.remoteStatusCalls, 2); + state.currentRemoteStatus = baseRemoteStatus; + yield* TestClock.adjust(Duration.seconds(59)); + yield* Effect.yieldNow; + assert.equal(state.remoteStatusCalls, 2); + yield* TestClock.adjust(Duration.seconds(1)); + assert.deepStrictEqual(yield* Deferred.await(recoveredDeferred), { + _tag: "remoteUpdated", + remote: baseRemoteStatus, + remoteLoaded: true, + } satisfies VcsStatusStreamEvent); + assert.equal(state.remoteStatusCalls, 3); + }).pipe(Effect.provide(Layer.merge(makeTestLayer(state), TestClock.layer()))); + }); + + it.effect("marks cached remote status unresolved when an explicit refresh fails", () => { + const state = makeTestState({ failRemoteStatus: false }); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const snapshotDeferred = yield* Deferred.make(); + const unavailableDeferred = yield* Deferred.make(); + yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => { + if (event._tag === "snapshot") { + return Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore); + } + return event._tag === "remoteUpdated" && event.remoteLoaded === false + ? Deferred.succeed(unavailableDeferred, event).pipe(Effect.ignore) + : Effect.void; + }).pipe(Effect.forkScoped); + + yield* Deferred.await(snapshotDeferred); + state.failRemoteStatus = true; + const refreshExit = yield* broadcaster.refreshStatus("/repo").pipe(Effect.exit); + assert.isTrue(Exit.isFailure(refreshExit)); + assert.deepStrictEqual(yield* Deferred.await(unavailableDeferred), { + _tag: "remoteUpdated", + remote: baseRemoteStatus, + remoteLoaded: false, + } satisfies VcsStatusStreamEvent); + }).pipe(Effect.provide(makeTestLayer(state))); + }); + + it.effect("marks cached remote status unresolved when full refresh invalidation defects", () => { + const state = makeTestState({ dieInvalidateStatus: false }); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const snapshotDeferred = yield* Deferred.make(); + const unavailableDeferred = yield* Deferred.make(); + yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => { + if (event._tag === "snapshot") { + return Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore); + } + return event._tag === "remoteUpdated" && event.remoteLoaded === false + ? Deferred.succeed(unavailableDeferred, event).pipe(Effect.ignore) + : Effect.void; + }).pipe(Effect.forkScoped); + + yield* Deferred.await(snapshotDeferred); + state.dieInvalidateStatus = true; + const refreshExit = yield* broadcaster.refreshStatus("/repo").pipe(Effect.exit); + assert.isTrue(Exit.isFailure(refreshExit)); + assert.deepStrictEqual(yield* Deferred.await(unavailableDeferred), { + _tag: "remoteUpdated", + remote: baseRemoteStatus, + remoteLoaded: false, + } satisfies VcsStatusStreamEvent); + }).pipe(Effect.provide(makeTestLayer(state))); + }); + + it.effect("returns cached status when a full refresh is dropped after a ref change", () => { + const state = makeTestState({ + currentRemoteStatus: remoteStatusWithPr, + blockRemoteStatusAtCall: 2, + remoteStatusStarted: null, + remoteStatusRelease: null, + }); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const scope = yield* Scope.make(); + const snapshotDeferred = yield* Deferred.make(); + yield* Stream.runForEach( + broadcaster.streamStatus( + { cwd: "/repo" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) }, + ), + (event) => + event._tag === "snapshot" + ? Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore) + : Effect.void, + ).pipe(Effect.forkIn(scope)); + yield* Deferred.await(snapshotDeferred); + + state.remoteStatusStarted = yield* Deferred.make(); + state.remoteStatusRelease = yield* Deferred.make(); + const refreshResult = yield* Deferred.make(); + yield* broadcaster.refreshStatus("/repo").pipe( + Effect.flatMap((status) => Deferred.succeed(refreshResult, status)), + Effect.forkIn(scope), + ); + yield* Deferred.await(state.remoteStatusStarted); + + state.currentLocalStatus = { + ...baseLocalStatus, + refName: "feature/new-ref", + }; + yield* broadcaster.refreshLocalStatus("/repo"); + yield* Deferred.succeed(state.remoteStatusRelease, undefined); + assert.deepStrictEqual(yield* Deferred.await(refreshResult), { + ...state.currentLocalStatus, + hasUpstream: false, + aheadCount: 0, + behindCount: 0, + aheadOfDefaultCount: 0, + pr: null, + } satisfies VcsStatusResult); + + const latestSnapshot = yield* Stream.runHead( + broadcaster.streamStatus( + { cwd: "/repo" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) }, + ), + ); + assert.isTrue(Option.isSome(latestSnapshot)); + if (Option.isSome(latestSnapshot)) { + assert.deepStrictEqual(latestSnapshot.value, { + _tag: "snapshot", + local: state.currentLocalStatus, + remote: remoteStatusWithPr, + remoteLoaded: false, + } satisfies VcsStatusStreamEvent); + } + + yield* Scope.close(scope, Exit.void); + }).pipe(Effect.provide(makeTestLayer(state))); + }); + + it.effect("ignores a stale refresh failure after the new ref has loaded", () => { + const state = makeTestState({ + failRemoteStatus: false, + blockRemoteStatusAtCall: 2, + remoteStatusStarted: null, + remoteStatusRelease: null, + }); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const scope = yield* Scope.make(); + state.remoteStatusStarted = yield* Deferred.make(); + state.remoteStatusRelease = yield* Deferred.make(); + const staleRefreshDone = yield* Deferred.make(); + yield* broadcaster + .refreshStatus("/repo") + .pipe( + Effect.ensuring(Deferred.succeed(staleRefreshDone, undefined).pipe(Effect.ignore)), + Effect.forkIn(scope), + ); + yield* Deferred.await(state.remoteStatusStarted); + + state.currentLocalStatus = { + ...baseLocalStatus, + refName: "feature/new-ref", + }; + yield* broadcaster.refreshLocalStatus("/repo"); + yield* broadcaster.refreshStatus("/repo"); + + state.failRemoteStatus = true; + yield* Deferred.succeed(state.remoteStatusRelease, undefined); + yield* Deferred.await(staleRefreshDone); + state.failRemoteStatus = false; + + const latestSnapshot = yield* Stream.runHead( + broadcaster.streamStatus( + { cwd: "/repo" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) }, + ), + ); + assert.isTrue(Option.isSome(latestSnapshot)); + if (Option.isSome(latestSnapshot)) { + assert.deepStrictEqual(latestSnapshot.value, { + _tag: "snapshot", + local: state.currentLocalStatus, + remote: baseRemoteStatus, + remoteLoaded: true, + } satisfies VcsStatusStreamEvent); + } + + yield* Scope.close(scope, Exit.void); + }).pipe(Effect.provide(makeTestLayer(state))); + }); + it("backs off remote refresh failures exponentially and honors larger configured intervals", () => { assert.equal( Duration.toMillis(VcsStatusBroadcaster.remoteRefreshFailureDelay(1, Duration.seconds(1))), diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index f28069f6d8b..ab73f7d50ca 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -127,8 +127,17 @@ interface CachedValue { interface CachedVcsStatus { readonly local: CachedValue | null; readonly remote: CachedValue | null; + readonly remoteLoaded: boolean; + readonly refGeneration: number; } +const EMPTY_CACHED_VCS_STATUS: CachedVcsStatus = { + local: null, + remote: null, + remoteLoaded: false, + refGeneration: 0, +}; + interface ActiveRemotePoller { readonly fiber: Fiber.Fiber; readonly subscriberCount: number; @@ -207,11 +216,14 @@ export const make = Effect.gen(function* () { value: local, } satisfies CachedValue; const shouldPublish = yield* Ref.modify(cacheRef, (cache) => { - const previous = cache.get(cwd) ?? { local: null, remote: null }; + const previous = cache.get(cwd) ?? EMPTY_CACHED_VCS_STATUS; + const refChanged = previous.local?.value.refName !== local.refName; const nextCache = new Map(cache); nextCache.set(cwd, { ...previous, local: nextLocal, + remoteLoaded: refChanged ? false : previous.remoteLoaded, + refGeneration: refChanged ? previous.refGeneration + 1 : previous.refGeneration, }); return [previous.local?.fingerprint !== nextLocal.fingerprint, nextCache] as const; }); @@ -231,40 +243,94 @@ export const make = Effect.gen(function* () { ); const updateCachedRemoteStatus = Effect.fn("VcsStatusBroadcaster.updateCachedRemoteStatus")( - function* (cwd: string, remote: VcsStatusRemoteResult | null, options?: { publish?: boolean }) { + function* ( + cwd: string, + remote: VcsStatusRemoteResult | null, + options?: { + publish?: boolean; + expectedRefGeneration?: number; + remoteLoaded?: boolean; + }, + ) { const nextRemote = { fingerprint: fingerprintStatusPart(remote), value: remote, } satisfies CachedValue; - const shouldPublish = yield* Ref.modify(cacheRef, (cache) => { - const previous = cache.get(cwd) ?? { local: null, remote: null }; - const nextCache = new Map(cache); - nextCache.set(cwd, { - ...previous, - remote: nextRemote, - }); - return [previous.remote?.fingerprint !== nextRemote.fingerprint, nextCache] as const; - }); + const remoteLoaded = options?.remoteLoaded ?? remote?.prLookupFailed !== true; + const update = yield* Ref.modify( + cacheRef, + ( + cache, + ): readonly [ + { readonly accepted: boolean; readonly shouldPublish: boolean }, + Map, + ] => { + const previous = cache.get(cwd) ?? EMPTY_CACHED_VCS_STATUS; + if ( + options?.expectedRefGeneration !== undefined && + previous.refGeneration !== options.expectedRefGeneration + ) { + return [{ accepted: false, shouldPublish: false }, cache]; + } + const nextCache = new Map(cache); + nextCache.set(cwd, { + ...previous, + remote: nextRemote, + remoteLoaded, + }); + return [ + { + accepted: true, + shouldPublish: + previous.remote?.fingerprint !== nextRemote.fingerprint || + previous.remoteLoaded !== remoteLoaded, + }, + nextCache, + ]; + }, + ); - if (options?.publish && shouldPublish) { + if (options?.publish && update.shouldPublish) { yield* PubSub.publish(changesPubSub, { cwd, event: { _tag: "remoteUpdated", remote, + remoteLoaded, }, }); } - return remote; + return { loaded: update.accepted && remoteLoaded, remote }; + }, + ); + + const markCachedRemoteUnavailable = Effect.fn("VcsStatusBroadcaster.markCachedRemoteUnavailable")( + function* (cwd: string, expectedRefGeneration: number) { + const cached = yield* getCachedStatus(cwd); + if (cached?.remoteLoaded !== true) return; + yield* updateCachedRemoteStatus(cwd, cached.remote?.value ?? null, { + publish: true, + expectedRefGeneration, + remoteLoaded: false, + }); }, ); + const markRemoteUnavailableForCause = ( + cwd: string, + expectedRefGeneration: number, + cause: Cause.Cause, + ) => + cause.reasons.some((reason) => !Cause.isInterruptReason(reason)) + ? markCachedRemoteUnavailable(cwd, expectedRefGeneration) + : Effect.void; + const updateCachedStatus = Effect.fn("VcsStatusBroadcaster.updateCachedStatus")(function* ( cwd: string, local: VcsStatusLocalResult, remote: VcsStatusRemoteResult | null, - options?: { publish?: boolean }, + options?: { publish?: boolean; expectedRefGeneration?: number }, ) { const nextLocal = { fingerprint: fingerprintStatusPart(local), @@ -274,32 +340,59 @@ export const make = Effect.gen(function* () { fingerprint: fingerprintStatusPart(remote), value: remote, } satisfies CachedValue; - const shouldPublish = yield* Ref.modify(cacheRef, (cache) => { - const previous = cache.get(cwd) ?? { local: null, remote: null }; + const remoteLoaded = remote?.prLookupFailed !== true; + const update = yield* Ref.modify(cacheRef, (cache) => { + const previous = cache.get(cwd) ?? EMPTY_CACHED_VCS_STATUS; + if ( + options?.expectedRefGeneration !== undefined && + previous.refGeneration !== options.expectedRefGeneration + ) { + return [ + { + shouldPublish: false, + status: previous.local + ? mergeGitStatusParts( + previous.local.value, + previous.remoteLoaded ? (previous.remote?.value ?? null) : null, + ) + : mergeGitStatusParts(local, remote), + }, + cache, + ] as const; + } + const refChanged = previous.local?.value.refName !== local.refName; const nextCache = new Map(cache); nextCache.set(cwd, { local: nextLocal, remote: nextRemote, + remoteLoaded, + refGeneration: refChanged ? previous.refGeneration + 1 : previous.refGeneration, }); return [ - previous.local?.fingerprint !== nextLocal.fingerprint || - previous.remote?.fingerprint !== nextRemote.fingerprint, + { + shouldPublish: + previous.local?.fingerprint !== nextLocal.fingerprint || + previous.remote?.fingerprint !== nextRemote.fingerprint || + previous.remoteLoaded !== remoteLoaded, + status: mergeGitStatusParts(local, remote), + }, nextCache, ] as const; }); - if (options?.publish && shouldPublish) { + if (options?.publish && update.shouldPublish) { yield* PubSub.publish(changesPubSub, { cwd, event: { _tag: "snapshot", local, remote, + remoteLoaded, }, }); } - return mergeGitStatusParts(local, remote); + return update.status; }); const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* ( @@ -326,13 +419,15 @@ export const make = Effect.gen(function* () { )(function* (input) { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); const cached = yield* getCachedStatus(cwd); - if (cached?.local && cached.remote) { + if (cached?.local && cached.remote && cached.remoteLoaded) { return mergeGitStatusParts(cached.local.value, cached.remote.value); } const [local, remote] = yield* Effect.all( [ cached?.local ? Effect.succeed(cached.local.value) : workflow.localStatus({ cwd }), - cached?.remote ? Effect.succeed(cached.remote.value) : workflow.remoteStatus({ cwd }), + cached?.remote && cached.remoteLoaded + ? Effect.succeed(cached.remote.value) + : workflow.remoteStatus({ cwd }), ], { concurrency: "unbounded" }, ); @@ -358,25 +453,41 @@ export const make = Effect.gen(function* () { cwd: string, options?: { readonly refreshUpstream?: boolean }, ) { - if (options?.refreshUpstream !== false) { - yield* workflow.invalidateRemoteStatus(cwd); - } - const remote = yield* workflow.remoteStatus({ cwd }, options); - return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); + const expectedRefGeneration = (yield* getCachedStatus(cwd))?.refGeneration ?? 0; + return yield* Effect.gen(function* () { + if (options?.refreshUpstream !== false) { + yield* workflow.invalidateRemoteStatus(cwd); + } + const remote = yield* workflow.remoteStatus({ cwd }, options); + return yield* updateCachedRemoteStatus(cwd, remote, { + publish: true, + expectedRefGeneration, + }); + }).pipe( + Effect.tapCause((cause) => markRemoteUnavailableForCause(cwd, expectedRefGeneration, cause)), + ); }); const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn( "VcsStatusBroadcaster.refreshStatus", )(function* (rawCwd) { const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); - // invalidateStatus (not the two partial invalidations) so an explicit - // refresh also bypasses GitManager's slow PR-lookup cache. - yield* workflow.invalidateStatus(cwd); - const [local, remote] = yield* Effect.all( - [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd })], - { concurrency: "unbounded" }, + const expectedRefGeneration = (yield* getCachedStatus(cwd))?.refGeneration ?? 0; + return yield* Effect.gen(function* () { + // invalidateStatus (not the two partial invalidations) so an explicit + // refresh also bypasses GitManager's slow PR-lookup cache. + yield* workflow.invalidateStatus(cwd); + const [local, remote] = yield* Effect.all( + [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd })], + { concurrency: "unbounded" }, + ); + return yield* updateCachedStatus(cwd, local, remote, { + publish: true, + expectedRefGeneration, + }); + }).pipe( + Effect.tapCause((cause) => markRemoteUnavailableForCause(cwd, expectedRefGeneration, cause)), ); - return yield* updateCachedStatus(cwd, local, remote, { publish: true }); }); const makeRemoteRefreshLoop = ( @@ -393,7 +504,11 @@ export const make = Effect.gen(function* () { const activeInterval = Duration.isZero(configuredInterval) ? DEFAULT_VCS_STATUS_REFRESH_INTERVAL : configuredInterval; - const needsInitialRefresh = yield* Ref.get(needsInitialRefreshRef); + const cachedRemoteLoaded = (yield* getCachedStatus(cwd))?.remoteLoaded ?? false; + const needsInitialRefresh = (yield* Ref.get(needsInitialRefreshRef)) || !cachedRemoteLoaded; + if (needsInitialRefresh) { + yield* Ref.set(needsInitialRefreshRef, true); + } if (Duration.isZero(configuredInterval) && !needsInitialRefresh) { return activeInterval; } @@ -418,9 +533,16 @@ export const make = Effect.gen(function* () { refreshUpstream: !Duration.isZero(configuredInterval), }).pipe(Effect.exit); if (Exit.isSuccess(exit)) { - yield* Ref.set(needsInitialRefreshRef, false); - yield* Ref.set(consecutiveFailuresRef, 0); - return activeInterval; + yield* Ref.set(needsInitialRefreshRef, !exit.value.loaded); + if (exit.value.loaded) { + yield* Ref.set(consecutiveFailuresRef, 0); + return activeInterval; + } + const unresolvedAttempts = yield* Ref.updateAndGet( + consecutiveFailuresRef, + (count) => count + 1, + ); + return remoteRefreshFailureDelay(unresolvedAttempts, activeInterval); } const interruptionReasons = exit.cause.reasons.filter(Cause.isInterruptReason); @@ -561,12 +683,13 @@ export const make = Effect.gen(function* () { const initialLocal = yield* getOrLoadLocalStatus(cwd); const cachedStatus = yield* getCachedStatus(cwd); const initialRemote = cachedStatus?.remote?.value ?? null; + const initialRemoteLoaded = cachedStatus?.remoteLoaded ?? false; yield* retainRemotePoller( cwd, input.cwd, options?.automaticRemoteRefreshInterval ?? Effect.succeed(DEFAULT_VCS_STATUS_REFRESH_INTERVAL), - cachedStatus?.remote === null || cachedStatus?.remote === undefined, + !initialRemoteLoaded, ); const release = releaseRemotePoller(cwd, input.cwd).pipe(Effect.ignore, Effect.asVoid); @@ -576,6 +699,7 @@ export const make = Effect.gen(function* () { _tag: "snapshot" as const, local: initialLocal, remote: initialRemote, + remoteLoaded: initialRemoteLoaded, }), Stream.fromSubscription(subscription).pipe( Stream.filter((event) => event.cwd === cwd), diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 2b9eda1a787..a8e265daca8 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -25,7 +25,11 @@ import { connectionStatusTitle, type EnvironmentConnectionPresentation, } from "@t3tools/client-runtime/connection"; -import { effectiveSettled, effectiveSnoozed } from "@t3tools/client-runtime/state/thread-settled"; +import { + effectiveSettled, + effectiveSnoozed, + resolveChangeRequestSettlementState, +} from "@t3tools/client-runtime/state/thread-settled"; import { parseScopedThreadKey, scopedThreadKey, @@ -236,7 +240,6 @@ import { shouldShowProviderStatusBanner, } from "./chat/ProviderStatusBanner"; import { ThreadErrorBanner } from "./chat/ThreadErrorBanner"; -import { resolveThreadPr } from "./ThreadStatusIndicators"; import { ComposerBannerStack, type ComposerBannerStackItem } from "./chat/ComposerBannerStack"; import { ThreadSyncStatusPill } from "./chat/ThreadSyncStatusPill"; import { @@ -3962,9 +3965,10 @@ function ChatViewContent(props: ChatViewProps) { // so the banner and the sidebar row never disagree. const activeThreadShell = useThreadShell(isServerThread ? activeThreadRef : null); const autoSettleAfterDays = useClientSettings((settings) => settings.sidebarAutoSettleAfterDays); - const activeThreadPr = resolveThreadPr({ + const activeThreadChangeRequestState = resolveChangeRequestSettlementState({ threadBranch: activeThread?.branch ?? null, - gitStatus: gitStatusQuery.data ?? null, + gitStatus: gitStatusQuery.data, + gitStatusError: gitStatusQuery.error, }); const supportsSettlement = serverConfig?.environment.capabilities.threadSettlement === true; const supportsSnooze = serverConfig?.environment.capabilities.threadSnooze === true; @@ -3990,10 +3994,10 @@ function ChatViewContent(props: ChatViewProps) { return effectiveSettled(activeThreadShell, { now: `${nowMinute}:00.000Z`, autoSettleAfterDays, - changeRequestState: activeThreadPr?.state ?? null, + changeRequestState: activeThreadChangeRequestState, }); }, [ - activeThreadPr?.state, + activeThreadChangeRequestState, activeThreadShell, autoSettleAfterDays, nowMinute, diff --git a/apps/web/src/components/SidebarV2.tsx b/apps/web/src/components/SidebarV2.tsx index ee73b570514..745f8fd783f 100644 --- a/apps/web/src/components/SidebarV2.tsx +++ b/apps/web/src/components/SidebarV2.tsx @@ -2,9 +2,14 @@ import { autoAnimate } from "@formkit/auto-animate"; import { useAtomValue } from "@effect/atom-react"; import { canSnooze, + type ChangeRequestSettlementState, effectiveSettled, effectiveSnoozed, + resolveChangeRequestSettlementState, + selectChangeRequestLookupWindow, + threadChangeRequestStateKey, threadWokeAt, + updateChangeRequestSettlementState, } from "@t3tools/client-runtime/state/thread-settled"; import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/models"; import { @@ -93,7 +98,7 @@ import { useNowMinute } from "../hooks/useNowMinute"; import { useEnvironments, usePrimaryEnvironmentId } from "../state/environments"; import { useProjects, useThreadShells } from "../state/entities"; import { environmentServerConfigsAtom, primaryServerKeybindingsAtom } from "../state/server"; -import { vcsEnvironment } from "../state/vcs"; +import { threadVcsEnvironment, vcsEnvironment } from "../state/vcs"; import { threadEnvironment } from "../state/threads"; import { projectEnvironment } from "../state/projects"; import { useEnvironmentQuery } from "../state/query"; @@ -167,6 +172,8 @@ import { useComposerDraftStore } from "../composerDraftStore"; // stays behind an explicit Show more. const SETTLED_TAIL_INITIAL_COUNT = 10; const SETTLED_TAIL_PAGE_COUNT = 25; +const CHANGE_REQUEST_LOOKUP_LIMIT = 16; +const CHANGE_REQUEST_LOOKUP_WINDOW_MS = 30_000; const PROJECT_GROUPING_MODE_LABELS: Record = { repository: "Group by repository", repository_path: "Group by repository path", @@ -383,6 +390,43 @@ function SnoozePopoverButton(props: { ); } +function useSidebarV2ChangeRequestState( + thread: SidebarThreadSummary, + projectCwd: string | null, + onChangeRequestState: (stateKey: string, state: ChangeRequestSettlementState) => void, +) { + const gitCwd = thread.worktreePath ?? projectCwd; + const gitStatus = useEnvironmentQuery( + (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null + ? threadVcsEnvironment.status({ + environmentId: thread.environmentId, + input: { cwd: gitCwd }, + }) + : null, + ); + const changeRequestState = resolveChangeRequestSettlementState({ + threadBranch: thread.branch, + gitStatus: gitStatus.data, + gitStatusError: gitStatus.error, + }); + const stateKey = threadChangeRequestStateKey(thread); + useEffect(() => { + onChangeRequestState(stateKey, changeRequestState); + }, [changeRequestState, onChangeRequestState, stateKey]); + return gitStatus; +} + +const SidebarV2ChangeRequestStateReporter = memo( + function SidebarV2ChangeRequestStateReporter(props: { + thread: SidebarThreadSummary; + projectCwd: string | null; + onChangeRequestState: (stateKey: string, state: ChangeRequestSettlementState) => void; + }) { + useSidebarV2ChangeRequestState(props.thread, props.projectCwd, props.onChangeRequestState); + return null; + }, +); + const SidebarV2Row = memo(function SidebarV2Row(props: { thread: SidebarThreadSummary; variant: "card" | "slim"; @@ -419,7 +463,7 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { onUnsettle: (threadRef: ScopedThreadRef) => void; onSnooze: (threadRef: ScopedThreadRef, preset: SnoozePreset) => void; onUnsnooze: (threadRef: ScopedThreadRef) => void; - onChangeRequestState: (threadKey: string, state: "open" | "closed" | "merged" | null) => void; + onChangeRequestState: (stateKey: string, state: ChangeRequestSettlementState) => void; }) { const { isRenaming, @@ -522,15 +566,7 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { } : null; - const gitCwd = thread.worktreePath ?? props.projectCwd; - const gitStatus = useEnvironmentQuery( - (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null - ? vcsEnvironment.status({ - environmentId: thread.environmentId, - input: { cwd: gitCwd }, - }) - : null, - ); + const gitStatus = useSidebarV2ChangeRequestState(thread, props.projectCwd, onChangeRequestState); const branchMismatch = resolveLocalCheckoutBranchMismatch({ effectiveEnvMode: thread.worktreePath === null ? "local" : "worktree", activeWorktreePath: thread.worktreePath, @@ -543,13 +579,6 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { }); const prStatus = prStatusIndicator(pr, gitStatus.data?.sourceControlProvider); const settledPrHoverClass = pr ? settledPrHoverColorClass(pr.state) : undefined; - // Report the PR state up: the parent partitions rows with effectiveSettled, - // and a merged/closed PR auto-settles a thread — data only rows have. - const prState = pr?.state ?? null; - useEffect(() => { - onChangeRequestState(threadKey, prState); - }, [onChangeRequestState, prState, threadKey]); - const modelInstanceId = thread.session?.providerInstanceId ?? thread.modelSelection.instanceId; const providerEntry = props.providerEntryByInstanceId.get(modelInstanceId) ?? null; const driverKind = providerEntry?.driverKind ?? null; @@ -1350,20 +1379,13 @@ export default function SidebarV2() { // PR states stream in per-row (rows own the VCS subscriptions); a merged or // closed PR auto-settles its thread on the next partition. const [changeRequestStateByKey, setChangeRequestStateByKey] = useState< - ReadonlyMap + ReadonlyMap >(() => new Map()); const handleChangeRequestState = useCallback( - (threadKey: string, state: "open" | "closed" | "merged" | null) => { - setChangeRequestStateByKey((current) => { - if ((current.get(threadKey) ?? null) === state) return current; - const next = new Map(current); - if (state === null) { - next.delete(threadKey); - } else { - next.set(threadKey, state); - } - return next; - }); + (stateKey: string, state: ChangeRequestSettlementState) => { + setChangeRequestStateByKey((current) => + updateChangeRequestSettlementState(current, stateKey, state), + ); }, [], ); @@ -1583,8 +1605,10 @@ export default function SidebarV2() { serverConfigs.get(thread.environmentId)?.environment.capabilities.threadSettlement === true; const supportsSnooze = serverConfigs.get(thread.environmentId)?.environment.capabilities.threadSnooze === true; - const threadKey = scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)); - const changeRequestState = changeRequestStateByKey.get(threadKey) ?? null; + const changeRequestState = + thread.branch === null + ? "none" + : (changeRequestStateByKey.get(threadChangeRequestStateKey(thread)) ?? "unknown"); // Snooze outranks settled classification: an explicitly snoozed thread // belongs to the shelf even if it would also auto-settle (the shelf's // wake time is a stronger statement about when it matters again). @@ -1733,6 +1757,35 @@ export default function SidebarV2() { () => [...activeThreads, ...visibleSnoozedThreads, ...renderedSettledThreads], [activeThreads, visibleSnoozedThreads, renderedSettledThreads], ); + // Collapsed and paged shelf rows unmount, so revisit their PR state in a + // bounded pool instead of letting a cached no-PR/closed result live forever. + const backgroundChangeRequestThreads = useMemo(() => { + const renderedStateKeys = new Set( + (isSearchingThreads ? [] : orderedThreads).map(threadChangeRequestStateKey), + ); + return [...activeThreads, ...snoozedThreads, ...settledThreads].filter( + (thread) => + thread.branch !== null && !renderedStateKeys.has(threadChangeRequestStateKey(thread)), + ); + }, [activeThreads, isSearchingThreads, orderedThreads, settledThreads, snoozedThreads]); + const [changeRequestLookupWindowIndex, setChangeRequestLookupWindowIndex] = useState(0); + useEffect(() => { + if (backgroundChangeRequestThreads.length <= CHANGE_REQUEST_LOOKUP_LIMIT) return; + const interval = window.setInterval( + () => setChangeRequestLookupWindowIndex((current) => current + 1), + CHANGE_REQUEST_LOOKUP_WINDOW_MS, + ); + return () => window.clearInterval(interval); + }, [backgroundChangeRequestThreads.length]); + const backgroundChangeRequestLookupThreads = useMemo( + () => + selectChangeRequestLookupWindow({ + targets: backgroundChangeRequestThreads, + windowIndex: changeRequestLookupWindowIndex, + limit: CHANGE_REQUEST_LOOKUP_LIMIT, + }), + [backgroundChangeRequestThreads, changeRequestLookupWindowIndex], + ); const orderedThreadKeys = useMemo( () => orderedThreads.map((thread) => @@ -2583,6 +2636,14 @@ export default function SidebarV2() { return ( <> + {backgroundChangeRequestLookupThreads.map((thread) => ( + + ))}