diff --git a/apps/mobile/src/state/use-thread-pr.ts b/apps/mobile/src/state/use-thread-pr.ts index a3440cd4848..85af767469c 100644 --- a/apps/mobile/src/state/use-thread-pr.ts +++ b/apps/mobile/src/state/use-thread-pr.ts @@ -15,6 +15,10 @@ export { * (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. + * + * Keyed by the thread's recorded branch, not the current checkout: the + * streamed status is the fast path while the checkout matches, and a + * branch-keyed lookup keeps the badge alive once the checkout moves away. */ export function useThreadPr( thread: EnvironmentThreadShell, @@ -29,13 +33,25 @@ export function useThreadPr( }) : null, ); + const branchPr = useEnvironmentQuery( + thread.branch !== null && cwd !== null + ? vcsEnvironment.branchPr({ + environmentId: thread.environmentId, + input: { cwd, branch: thread.branch }, + }) + : null, + ); const status = gitStatus.data; - if (status === null || thread.branch === null || status.refName !== thread.branch) { - return null; + // When the status describes the thread's own branch it is authoritative + // about that branch's PR — including when it reports none, so this must not + // fall through to a warm branch lookup that would resurrect a stale badge. + if (status !== null && thread.branch !== null && status.refName === thread.branch) { + return status.pr ? presentThreadPr(status.pr, status.sourceControlProvider) : null; } - if (!status.pr) { + const pr = branchPr.data?.pr; + if (!pr) { return null; } - return presentThreadPr(status.pr, status.sourceControlProvider); + return presentThreadPr(pr, status?.sourceControlProvider); } diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index c1d9f786adb..aca3b3b7516 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -64,6 +64,7 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.subscribeVcsStatus]: AuthOrchestrationReadScope, [WS_METHODS.subscribeResourceTelemetry]: AuthOrchestrationReadScope, [WS_METHODS.vcsRefreshStatus]: AuthOrchestrationReadScope, + [WS_METHODS.vcsBranchPr]: AuthOrchestrationReadScope, [WS_METHODS.vcsPull]: AuthOrchestrationOperateScope, [WS_METHODS.gitRunStackedAction]: AuthOrchestrationOperateScope, [WS_METHODS.gitResolvePullRequest]: AuthOrchestrationOperateScope, diff --git a/apps/server/src/git/GitManager.ts b/apps/server/src/git/GitManager.ts index da002df5e6c..a356e9db2ac 100644 --- a/apps/server/src/git/GitManager.ts +++ b/apps/server/src/git/GitManager.ts @@ -23,6 +23,8 @@ import { GitRunStackedActionInput, GitRunStackedActionResult, GitStackedAction, + type VcsBranchPrInput, + type VcsBranchPrResult, VcsStatusInput, type VcsStatusLocalResult, type VcsStatusRemoteResult, @@ -87,6 +89,9 @@ export class GitManager extends Context.Service< input: VcsStatusInput, options?: GitVcsDriver.GitRemoteStatusOptions, ) => Effect.Effect; + readonly branchPr: ( + input: VcsBranchPrInput, + ) => Effect.Effect; readonly invalidateLocalStatus: (cwd: string) => Effect.Effect; readonly invalidateRemoteStatus: (cwd: string) => Effect.Effect; readonly invalidateStatus: (cwd: string) => Effect.Effect; @@ -1684,6 +1689,43 @@ export const make = Effect.gen(function* () { }); return mergeGitStatusParts(local, remote); }); + // PR lookup for a branch that is not necessarily checked out. Threads keep + // a recorded branch, and their PR badge must survive the checkout moving + // elsewhere — so this derives the branch's upstream from git config instead + // of HEAD and funnels into the same cached lookup as status. + const branchPr: GitManager["Service"]["branchPr"] = Effect.fn("branchPr")(function* (input) { + const cwd = yield* normalizeStatusCacheKey(input.cwd); + const branch = input.branch; + const [remoteName, mergeRef, originHead] = yield* Effect.all( + [ + readConfigValueNullable(cwd, `branch.${branch}.remote`), + readConfigValueNullable(cwd, `branch.${branch}.merge`), + gitCore + .execute({ + operation: "GitManager.branchPr.defaultBranch", + cwd, + args: ["symbolic-ref", "--short", "refs/remotes/origin/HEAD"], + allowNonZeroExit: true, + }) + .pipe(Effect.orElseSucceed(() => null)), + ], + { concurrency: "unbounded" }, + ); + const upstreamRef = + remoteName !== null && mergeRef !== null && mergeRef.startsWith("refs/heads/") + ? `${remoteName}/${mergeRef.slice("refs/heads/".length)}` + : null; + const originHeadBranch = + originHead !== null && originHead.exitCode === 0 + ? (originHead.stdout.trim().split("/").slice(1).join("/") ?? "") + : ""; + const isDefaultBranch = + originHeadBranch.length > 0 + ? branch === originHeadBranch + : branch === "main" || branch === "master"; + const pr = yield* lookupStatusPr(cwd, { branch, upstreamRef, isDefaultBranch }); + return { branch, pr } satisfies VcsBranchPrResult; + }); const invalidateLocalStatus: GitManager["Service"]["invalidateLocalStatus"] = Effect.fn( "invalidateLocalStatus", )(function* (cwd) { @@ -2136,6 +2178,7 @@ export const make = Effect.gen(function* () { return GitManager.of({ localStatus, remoteStatus, + branchPr, status, invalidateLocalStatus, invalidateRemoteStatus, diff --git a/apps/server/src/git/GitWorkflowService.ts b/apps/server/src/git/GitWorkflowService.ts index 100b9beadba..f5c37d3ae29 100644 --- a/apps/server/src/git/GitWorkflowService.ts +++ b/apps/server/src/git/GitWorkflowService.ts @@ -22,6 +22,8 @@ import { type GitResolvePullRequestResult, type GitRunStackedActionInput, type GitRunStackedActionResult, + type VcsBranchPrInput, + type VcsBranchPrResult, type VcsStatusInput, type VcsStatusLocalResult, type VcsStatusRemoteResult, @@ -45,6 +47,9 @@ export class GitWorkflowService extends Context.Service< input: VcsStatusInput, options?: GitVcsDriver.GitRemoteStatusOptions, ) => Effect.Effect; + readonly branchPr: ( + input: VcsBranchPrInput, + ) => Effect.Effect; readonly invalidateLocalStatus: (cwd: string) => Effect.Effect; readonly invalidateRemoteStatus: (cwd: string) => Effect.Effect; readonly invalidateStatus: (cwd: string) => Effect.Effect; @@ -270,6 +275,14 @@ export const make = Effect.gen(function* () { isGitRepository ? gitManager.remoteStatus(input, options) : Effect.succeed(null), ), ), + branchPr: (input) => + detectGitRepositoryForStatus("GitWorkflowService.branchPr", input.cwd).pipe( + Effect.flatMap((isGitRepository) => + isGitRepository + ? gitManager.branchPr(input) + : Effect.succeed({ branch: input.branch, pr: null }), + ), + ), invalidateLocalStatus: gitManager.invalidateLocalStatus, invalidateRemoteStatus: gitManager.invalidateRemoteStatus, invalidateStatus: gitManager.invalidateStatus, diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 6820a29e2c8..3d8f04ff335 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -6,6 +6,7 @@ import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; @@ -208,7 +209,7 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(makeTestLayer(state))); }); - it.effect("keeps the cached snapshot unchanged when a refresh branch fails", () => { + it.effect("keeps local progress and recovers the remote half after a failed refresh", () => { const state = { currentLocalStatus: baseLocalStatus, currentRemoteStatus: baseRemoteStatus, @@ -273,14 +274,104 @@ describe("VcsStatusBroadcaster", () => { state.failRemoteStatus = true; const refreshExit = yield* broadcaster.refreshStatus("/repo").pipe(Effect.exit); - const cached = yield* broadcaster.getStatus({ cwd: "/repo" }); - assert.isTrue(Exit.isFailure(refreshExit)); - assert.deepStrictEqual(cached, baseStatus); + + // The local half landed even though the refresh failed. The remote half + // was computed for the previous checkout, so the refName change dropped + // it rather than pairing the old branch's data with the new branch; it + // repopulates on the next successful read. + state.failRemoteStatus = false; + const cached = yield* broadcaster.getStatus({ cwd: "/repo" }); + assert.deepStrictEqual(cached, { + ...state.currentLocalStatus, + ...state.currentRemoteStatus, + }); }).pipe(Effect.provide(testLayer)); }); - it.effect("refreshes only the cached local snapshot when requested", () => { + it.effect("returns the new checkout when one lands mid-refresh", () => { + const state = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const remoteEntered = yield* Deferred.make(); + const releaseRemote = yield* Deferred.make(); + // Armed only around the refresh under test, so neither the priming read + // nor the follow-up refresh that the checkout change kicks off blocks. + const gate = { armed: false }; + const testLayer = VcsStatusBroadcaster.layer.pipe( + Layer.provideMerge(NodeServices.layer), + Layer.provide(makeBackgroundPolicyLayer(() => true)), + Layer.provide( + Layer.mock(GitWorkflowService.GitWorkflowService)({ + localStatus: () => + Effect.sync(() => { + state.localStatusCalls += 1; + return state.currentLocalStatus; + }), + remoteStatus: () => + Effect.suspend(() => { + state.remoteStatusCalls += 1; + if (!gate.armed) { + return Effect.succeed(state.currentRemoteStatus); + } + gate.armed = false; + return Deferred.succeed(remoteEntered, undefined).pipe( + Effect.andThen(Deferred.await(releaseRemote)), + Effect.as(state.currentRemoteStatus), + ); + }), + invalidateLocalStatus: () => + Effect.sync(() => { + state.localInvalidationCalls += 1; + }), + invalidateRemoteStatus: () => + Effect.sync(() => { + state.remoteInvalidationCalls += 1; + }), + invalidateStatus: () => + Effect.sync(() => { + state.localInvalidationCalls += 1; + state.remoteInvalidationCalls += 1; + }), + }), + ), + ); + + return yield* Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + + gate.armed = true; + const refreshFiber = yield* Effect.forkChild(broadcaster.refreshStatus("/repo"), { + startImmediately: true, + }); + yield* Deferred.await(remoteEntered); + + // A checkout lands while the slow remote read is still in flight. + state.currentLocalStatus = { + ...baseLocalStatus, + refName: "feature/checked-out-mid-refresh", + }; + yield* broadcaster.refreshLocalStatus("/repo"); + + yield* Deferred.succeed(releaseRemote, undefined); + const refreshed = yield* Fiber.join(refreshFiber); + + // The refresh must not hand back the branch that was checked out when + // it started; its own remote result was discarded as superseded. + assert.equal(refreshed.refName, "feature/checked-out-mid-refresh"); + }).pipe(Effect.provide(testLayer)); + }); + }); + + it.effect("refreshes only the cached local snapshot when the checkout is unchanged", () => { const state = { currentLocalStatus: baseLocalStatus, currentRemoteStatus: baseRemoteStatus, @@ -296,7 +387,6 @@ describe("VcsStatusBroadcaster", () => { state.currentLocalStatus = { ...baseLocalStatus, - refName: "feature/local-only-refresh", hasWorkingTreeChanges: true, }; @@ -316,6 +406,53 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(makeTestLayer(state))); }); + it.effect("repopulates the remote half when a local refresh changes the checkout", () => { + const state = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const initial = yield* broadcaster.getStatus({ cwd: "/repo" }); + + state.currentLocalStatus = { + ...baseLocalStatus, + refName: "feature/local-only-refresh", + hasWorkingTreeChanges: true, + }; + state.currentRemoteStatus = { + ...baseRemoteStatus, + aheadCount: 4, + }; + + const refreshedLocal = yield* broadcaster.refreshLocalStatus("/repo"); + + // The refName change drops the stale remote half and forks a remote + // refresh for the new branch; let that fiber settle. + for (let i = 0; i < 100 && state.remoteStatusCalls < 2; i++) { + yield* Effect.yieldNow; + } + + const cached = yield* broadcaster.getStatus({ cwd: "/repo" }); + + assert.deepStrictEqual(initial, baseStatus); + assert.deepStrictEqual(refreshedLocal, state.currentLocalStatus); + assert.deepStrictEqual(cached, { + ...state.currentLocalStatus, + ...state.currentRemoteStatus, + }); + assert.equal(state.localStatusCalls, 2); + assert.equal(state.remoteStatusCalls, 2); + assert.equal(state.localInvalidationCalls, 1); + assert.equal(state.remoteInvalidationCalls, 1); + }).pipe(Effect.provide(makeTestLayer(state))); + }); + it.effect("normalizes symlinked CWDs before cache lookup and workflow calls", () => { const seenCwds: string[] = []; const state = { diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index f28069f6d8b..c26f05c6031 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -10,6 +10,7 @@ import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schedule from "effect/Schedule"; import * as Scope from "effect/Scope"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SynchronizedRef from "effect/SynchronizedRef"; import type { @@ -194,6 +195,35 @@ export const make = Effect.gen(function* () { const cacheRef = yield* Ref.make(new Map()); const pollersRef = yield* SynchronizedRef.make(new Map()); + // Serialize refreshes per cwd so a read that started before a mutation (a + // checkout, a pull) cannot finish after the post-mutation refresh and + // publish stale state on top of fresh state. Local and remote halves get + // separate locks: local reads are cheap and must never queue behind an + // in-flight remote read (PR lookups hit the network for seconds). The only + // path that holds both acquires remote before local. + interface CwdLocks { + readonly local: Semaphore.Semaphore; + readonly remote: Semaphore.Semaphore; + } + const locksRef = yield* SynchronizedRef.make(new Map()); + const locksFor = (cwd: string) => + SynchronizedRef.modifyEffect(locksRef, (locks) => { + const existing = locks.get(cwd); + if (existing) { + return Effect.succeed([existing, locks] as const); + } + return Effect.all([Semaphore.make(1), Semaphore.make(1)]).pipe( + Effect.map(([local, remote]) => { + const created: CwdLocks = { local, remote }; + return [created, new Map(locks).set(cwd, created)] as const; + }), + ); + }); + const withLocalLock = (cwd: string, effect: Effect.Effect) => + locksFor(cwd).pipe(Effect.flatMap((locks) => locks.local.withPermits(1)(effect))); + const withRemoteLock = (cwd: string, effect: Effect.Effect) => + locksFor(cwd).pipe(Effect.flatMap((locks) => locks.remote.withPermits(1)(effect))); + const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* ( cwd: string, ) { @@ -206,27 +236,45 @@ export const make = Effect.gen(function* () { fingerprint: fingerprintStatusPart(local), value: local, } satisfies CachedValue; - const shouldPublish = yield* Ref.modify(cacheRef, (cache) => { + const { shouldPublish, checkoutChanged } = yield* Ref.modify(cacheRef, (cache) => { const previous = cache.get(cwd) ?? { local: null, remote: null }; + const changedRefName = + previous.local !== null && previous.local.value.refName !== local.refName; const nextCache = new Map(cache); nextCache.set(cwd, { - ...previous, local: nextLocal, + // The cached remote half (ahead/behind, PR) was computed for the + // previous checkout; carrying it across a refName change would pair + // the new branch with the old branch's PR. Drop it and let the + // remote refresh repopulate it for the new branch. + remote: changedRefName ? null : previous.remote, }); - return [previous.local?.fingerprint !== nextLocal.fingerprint, nextCache] as const; + return [ + { + shouldPublish: previous.local?.fingerprint !== nextLocal.fingerprint, + checkoutChanged: changedRefName, + }, + nextCache, + ] as const; }); if (options?.publish && shouldPublish) { yield* PubSub.publish(changesPubSub, { cwd, - event: { - _tag: "localUpdated", - local, - }, + event: checkoutChanged + ? { + _tag: "snapshot", + local, + remote: null, + } + : { + _tag: "localUpdated", + local, + }, }); } - return local; + return { local, checkoutChanged }; }, ); @@ -305,8 +353,13 @@ export const make = Effect.gen(function* () { const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* ( cwd: string, ) { - const local = yield* workflow.localStatus({ cwd }); - return yield* updateCachedLocalStatus(cwd, local); + return yield* withLocalLock( + cwd, + Effect.gen(function* () { + const local = yield* workflow.localStatus({ cwd }); + return (yield* updateCachedLocalStatus(cwd, local)).local; + }), + ); }); const getOrLoadLocalStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")(function* ( @@ -329,21 +382,52 @@ export const make = Effect.gen(function* () { if (cached?.local && cached.remote) { 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 }), - ], - { concurrency: "unbounded" }, + return yield* withRemoteLock( + cwd, + withLocalLock( + cwd, + Effect.gen(function* () { + // Re-check under the locks: a refresh may have populated the cache + // while this call was waiting for them. + const current = yield* getCachedStatus(cwd); + if (current?.local && current.remote) { + return mergeGitStatusParts(current.local.value, current.remote.value); + } + const [local, remote] = yield* Effect.all( + [ + current?.local ? Effect.succeed(current.local.value) : workflow.localStatus({ cwd }), + current?.remote + ? Effect.succeed(current.remote.value) + : workflow.remoteStatus({ cwd }), + ], + { concurrency: "unbounded" }, + ); + return yield* updateCachedStatus(cwd, local, remote); + }), + ), ); - return yield* updateCachedStatus(cwd, local, remote); }); const refreshLocalStatusCore = Effect.fn("VcsStatusBroadcaster.refreshLocalStatusCore")( function* (cwd: string) { - yield* workflow.invalidateLocalStatus(cwd); - const local = yield* workflow.localStatus({ cwd }); - return yield* updateCachedLocalStatus(cwd, local, { publish: true }); + const result = yield* withLocalLock( + cwd, + Effect.gen(function* () { + yield* workflow.invalidateLocalStatus(cwd); + const local = yield* workflow.localStatus({ cwd }); + return yield* updateCachedLocalStatus(cwd, local, { publish: true }); + }), + ); + if (result.checkoutChanged) { + // The checkout moved (a switch, or a `git checkout` outside T3's + // commands). The remote half was dropped above; repopulate it for the + // new branch without waiting for the next poll tick. + yield* refreshRemoteStatus(cwd).pipe( + Effect.ignoreCause({ log: true }), + Effect.forkIn(broadcasterScope), + ); + } + return result.local; }, ); @@ -356,27 +440,65 @@ export const make = Effect.gen(function* () { const refreshRemoteStatus = Effect.fn("VcsStatusBroadcaster.refreshRemoteStatus")(function* ( cwd: string, - options?: { readonly refreshUpstream?: boolean }, + options?: { readonly refreshUpstream?: boolean; readonly invalidate?: boolean }, ) { - if (options?.refreshUpstream !== false) { - yield* workflow.invalidateRemoteStatus(cwd); - } - const remote = yield* workflow.remoteStatus({ cwd }, options); - return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); + return yield* withRemoteLock( + cwd, + Effect.gen(function* () { + const refNameBefore = (yield* getCachedStatus(cwd))?.local?.value.refName; + if (options?.invalidate !== false && options?.refreshUpstream !== false) { + yield* workflow.invalidateRemoteStatus(cwd); + } + const remote = yield* workflow.remoteStatus( + { cwd }, + options?.refreshUpstream === undefined + ? undefined + : { refreshUpstream: options.refreshUpstream }, + ); + // A checkout that lands while the slow remote read is in flight makes + // this result describe the previous branch. Publishing it would + // attach the old branch's PR/divergence to the new refName — discard + // it; the refName change itself already kicked a follow-up refresh. + const refNameAfter = (yield* getCachedStatus(cwd))?.local?.value.refName; + if (refNameBefore !== undefined && refNameAfter !== refNameBefore) { + return (yield* getCachedStatus(cwd))?.remote?.value ?? null; + } + return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); + }), + ); }); - const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn( - "VcsStatusBroadcaster.refreshStatus", - )(function* (rawCwd) { - const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); + const refreshStatusCore = Effect.fn("VcsStatusBroadcaster.refreshStatusCore")(function* ( + cwd: string, + ) { // 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 local = yield* withLocalLock( + cwd, + Effect.gen(function* () { + const localResult = yield* workflow.localStatus({ cwd }); + return (yield* updateCachedLocalStatus(cwd, localResult, { publish: true })).local; + }), ); - return yield* updateCachedStatus(cwd, local, remote, { publish: true }); + const remote = yield* refreshRemoteStatus(cwd, { invalidate: false }); + // A checkout can land between the two reads. refreshRemoteStatus discards + // a superseded remote result, which would leave `local` describing the old + // branch — returning that pair would hand the caller a status the cache + // itself no longer agrees with. Prefer the cache's halves, which the + // concurrent refresh has already reconciled for the new branch. + const cached = yield* getCachedStatus(cwd); + return mergeGitStatusParts( + cached?.local?.value ?? local, + cached?.remote ? cached.remote.value : remote, + ); + }); + + const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn( + "VcsStatusBroadcaster.refreshStatus", + )(function* (rawCwd) { + const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); + return yield* refreshStatusCore(cwd); }); const makeRemoteRefreshLoop = ( diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 4ece17cc1d5..b1061434b52 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1763,6 +1763,10 @@ const makeWsRpcLayer = ( "rpc.aggregate": "vcs", }, ), + [WS_METHODS.vcsBranchPr]: (input) => + observeRpcEffect(WS_METHODS.vcsBranchPr, gitWorkflow.branchPr(input), { + "rpc.aggregate": "vcs", + }), [WS_METHODS.vcsPull]: (input) => observeRpcEffect( WS_METHODS.vcsPull, @@ -1839,7 +1843,18 @@ const makeWsRpcLayer = ( [WS_METHODS.vcsSwitchRef]: (input) => observeRpcEffect( WS_METHODS.vcsSwitchRef, - gitWorkflow.switchRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), + gitWorkflow.switchRef(input).pipe( + // Publish the new refName before the RPC resolves so a client + // that awaited the switch can never observe the old branch from + // the status stream afterwards. The full refresh (remote half, + // PR lookup) is slow and stays forked. + Effect.tap(() => + vcsStatusBroadcaster + .refreshLocalStatus(input.cwd) + .pipe(Effect.ignoreCause({ log: true })), + ), + Effect.tap(() => refreshGitStatus(input.cwd)), + ), { "rpc.aggregate": "vcs" }, ), [WS_METHODS.vcsInit]: (input) => diff --git a/apps/web/src/components/BranchToolbar.logic.test.ts b/apps/web/src/components/BranchToolbar.logic.test.ts index 76336f1ef1f..86ced8deaf5 100644 --- a/apps/web/src/components/BranchToolbar.logic.test.ts +++ b/apps/web/src/components/BranchToolbar.logic.test.ts @@ -10,7 +10,6 @@ import { resolveEffectiveEnvMode, resolveEnvModeLabel, resolveBranchTriggerLabel, - resolveBranchToolbarPrBranch, resolveBranchToolbarValue, resolveLockedWorkspaceLabel, resolveLocalCheckoutBranchMismatch, @@ -270,35 +269,6 @@ describe("resolveBranchTriggerLabel", () => { }); }); -describe("resolveBranchToolbarPrBranch", () => { - it("uses the explicit thread branch when it matches the displayed branch", () => { - expect( - resolveBranchToolbarPrBranch({ - activeThreadBranch: "feature/current", - resolvedActiveBranch: "feature/current", - }), - ).toBe("feature/current"); - }); - - it("hides PR state while an optimistic branch switch is in flight", () => { - expect( - resolveBranchToolbarPrBranch({ - activeThreadBranch: "feature/current", - resolvedActiveBranch: "feature/next", - }), - ).toBeNull(); - }); - - it("does not infer PR state without an explicit thread branch", () => { - expect( - resolveBranchToolbarPrBranch({ - activeThreadBranch: null, - resolvedActiveBranch: "feature/current", - }), - ).toBeNull(); - }); -}); - describe("resolveLocalCheckoutBranchMismatch", () => { it("detects when a local thread is associated with a different branch than the checkout", () => { expect( diff --git a/apps/web/src/components/BranchToolbar.logic.ts b/apps/web/src/components/BranchToolbar.logic.ts index d9737f17a32..dd3300c0cbd 100644 --- a/apps/web/src/components/BranchToolbar.logic.ts +++ b/apps/web/src/components/BranchToolbar.logic.ts @@ -183,13 +183,6 @@ export function resolveBranchTriggerLabel(input: { return resolvedActiveBranch; } -export function resolveBranchToolbarPrBranch(input: { - activeThreadBranch: string | null; - resolvedActiveBranch: string | null; -}): string | null { - return input.activeThreadBranch === input.resolvedActiveBranch ? input.activeThreadBranch : null; -} - export function resolveLocalCheckoutBranchMismatch(input: { effectiveEnvMode: EnvMode; activeWorktreePath: string | null; diff --git a/apps/web/src/components/BranchToolbarBranchSelector.tsx b/apps/web/src/components/BranchToolbarBranchSelector.tsx index bbd27f65ab0..8c3eb8669f6 100644 --- a/apps/web/src/components/BranchToolbarBranchSelector.tsx +++ b/apps/web/src/components/BranchToolbarBranchSelector.tsx @@ -37,7 +37,6 @@ import { getSourceControlPresentation } from "../sourceControlPresentation"; import { deriveLocalBranchNameFromRemoteRef, resolveBranchTriggerLabel, - resolveBranchToolbarPrBranch, resolveBranchSelectionTarget, resolveBranchToolbarValue, resolveDraftEnvModeAfterBranchChange, @@ -47,7 +46,7 @@ import { import { ChangeRequestStatusIcon, prStatusIndicator, - resolveThreadPr, + selectBranchPr, } from "./ThreadStatusIndicators"; import { Button } from "./ui/button"; import { Switch } from "./ui/switch"; @@ -601,13 +600,28 @@ export function BranchToolbarBranchSelector({ startFromOrigin, }); - // PR pill shown next to the branch selector when the active branch has one. - const branchPr = resolveThreadPr({ - threadBranch: resolveBranchToolbarPrBranch({ - activeThreadBranch, - resolvedActiveBranch, - }), + // PR pill shown next to the branch selector when the displayed ref has one. + // The stream-fed status is the fast path while the checkout matches; the + // branch-keyed query keeps the pill correct when the checkout has moved + // elsewhere. Draft "From " selectors intentionally show none — the + // base branch's PR is not this thread's. + const branchPrLookupBranch = + (effectiveEnvMode === "worktree" && !activeWorktreePath) || + resolvedActiveBranchIsRemote === true + ? null + : resolvedActiveBranch; + const branchPrQuery = useEnvironmentQuery( + branchCwd !== null && branchPrLookupBranch !== null + ? vcsEnvironment.branchPr({ + environmentId, + input: { cwd: branchCwd, branch: branchPrLookupBranch }, + }) + : null, + ); + const branchPr = selectBranchPr({ + branch: branchPrLookupBranch, gitStatus: branchStatusQuery.data ?? null, + branchPr: branchPrQuery.data?.pr, }); const branchPrStatus = prStatusIndicator(branchPr, branchStatusQuery.data?.sourceControlProvider); // Action-oriented tooltip (the pill opens the PR), distinct from the sidebar's diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index ad704d9e1b2..262c7ec07b5 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -238,7 +238,7 @@ import { shouldShowProviderStatusBanner, } from "./chat/ProviderStatusBanner"; import { ThreadErrorBanner } from "./chat/ThreadErrorBanner"; -import { resolveThreadPr } from "./ThreadStatusIndicators"; +import { useThreadPr } from "./ThreadStatusIndicators"; import { ComposerBannerStack, type ComposerBannerStackItem } from "./chat/ComposerBannerStack"; import { ThreadSyncStatusPill } from "./chat/ThreadSyncStatusPill"; import { @@ -3964,7 +3964,9 @@ 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 activeThreadPr = useThreadPr({ + environmentId, + cwd: gitStatusCwd, threadBranch: activeThread?.branch ?? null, gitStatus: gitStatusQuery.data ?? null, }); diff --git a/apps/web/src/components/GitActionsControl.tsx b/apps/web/src/components/GitActionsControl.tsx index 20e5c349790..3451b841d08 100644 --- a/apps/web/src/components/GitActionsControl.tsx +++ b/apps/web/src/components/GitActionsControl.tsx @@ -1099,6 +1099,12 @@ export default function GitActionsControl({ // Default to true while loading so we don't flash init controls. const isRepo = gitStatus?.isRepo ?? true; const hasPrimaryRemote = gitStatus?.hasPrimaryRemote ?? false; + // Deliberately the checkout's status, not the active thread's branch: every + // action here (commit, push, create) runs against whatever is checked out at + // `gitCwd`, so gating them on another branch's PR would offer "Create PR" + // for a branch that already has one — or hide it for one that doesn't. + // Thread-scoped PR state lives on the sidebar badge and the composer pill, + // which are branch-keyed; the branch-mismatch banner explains the split. const gitStatusForActions = gitStatus; const allFiles = gitStatusForActions?.workingTree.files ?? []; diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index cffab8bd577..27e1b9269fc 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -16,8 +16,8 @@ import { ChangeRequestStatusIcon, prStatusIndicator, PrStatusTooltipContent, - resolveThreadPr, terminalStatusFromRunningIds, + useThreadPr, ThreadStatusLabel, ThreadWorktreeIndicator, } from "./ThreadStatusIndicators"; @@ -449,7 +449,9 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr lastVisitedAt, }, }); - const pr = resolveThreadPr({ + const pr = useThreadPr({ + environmentId: thread.environmentId, + cwd: gitCwd, threadBranch: thread.branch, gitStatus: gitStatus.data, }); diff --git a/apps/web/src/components/SidebarV2.tsx b/apps/web/src/components/SidebarV2.tsx index e8f7f8bbb12..c54b3160c38 100644 --- a/apps/web/src/components/SidebarV2.tsx +++ b/apps/web/src/components/SidebarV2.tsx @@ -127,8 +127,8 @@ import { import { resolveLocalCheckoutBranchMismatch } from "./BranchToolbar.logic"; import { prStatusIndicator, - resolveThreadPr, settledPrHoverColorClass, + useThreadPr, terminalStatusFromRunningIds, type TerminalStatusIndicator, } from "./ThreadStatusIndicators"; @@ -543,7 +543,9 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { activeThreadBranch: thread.branch, currentGitBranch: gitStatus.data?.refName ?? null, }); - const pr = resolveThreadPr({ + const pr = useThreadPr({ + environmentId: thread.environmentId, + cwd: gitCwd, threadBranch: thread.branch, gitStatus: gitStatus.data, }); diff --git a/apps/web/src/components/ThreadStatusIndicators.test.ts b/apps/web/src/components/ThreadStatusIndicators.test.ts index 3eb8e4f710f..e5b35771410 100644 --- a/apps/web/src/components/ThreadStatusIndicators.test.ts +++ b/apps/web/src/components/ThreadStatusIndicators.test.ts @@ -3,7 +3,7 @@ import { describe, expect, it } from "vite-plus/test"; import { prStatusIndicator, - resolveThreadPr, + selectBranchPr, settledPrHoverColorClass, } from "./ThreadStatusIndicators"; @@ -30,43 +30,54 @@ function status(overrides: Partial = {}): VcsStatusResult { }; } -describe("resolveThreadPr", () => { - it("keeps local-checkout PR indicators scoped to the stored thread branch", () => { - expect( - resolveThreadPr({ - threadBranch: "feature/other", - gitStatus: status(), - }), - ).toBeNull(); +const otherBranchPr = { + number: 7, + title: "Branch-keyed PR", + url: "https://github.com/pingdotgg/t3code/pull/7", + baseRef: "main", + headRef: "feature/other", + state: "open", +} as const; + +describe("selectBranchPr", () => { + it("shows the PR when the live checkout matches the requested branch", () => { + const gitStatus = status(); + + expect(selectBranchPr({ branch: "feature/current", gitStatus, branchPr: undefined })).toBe( + gitStatus.pr, + ); }); - it("hides PR indicators when a dedicated worktree has switched away from the thread branch", () => { + it("falls back to the branch-keyed lookup when the checkout moved elsewhere", () => { expect( - resolveThreadPr({ - threadBranch: "stack/base", + selectBranchPr({ + branch: "feature/other", gitStatus: status(), + branchPr: otherBranchPr, }), - ).toBeNull(); + ).toBe(otherBranchPr); }); - it("hides PR indicators when thread branch metadata is missing", () => { + it("treats an empty PR on the matching checkout as authoritative", () => { + // Regression: a warm branch-keyed result must not resurrect a badge the + // status stream just cleared (e.g. immediately after a merge). expect( - resolveThreadPr({ - threadBranch: null, - gitStatus: status(), + selectBranchPr({ + branch: "feature/current", + gitStatus: status({ pr: null }), + branchPr: otherBranchPr, }), ).toBeNull(); }); - it("shows the PR when the live checkout matches the stored thread branch", () => { - const gitStatus = status(); + it("returns nothing when the branch is unknown and no lookup resolved", () => { + expect(selectBranchPr({ branch: null, gitStatus: status(), branchPr: undefined })).toBeNull(); + }); + it("returns nothing while the branch-keyed lookup is unresolved", () => { expect( - resolveThreadPr({ - threadBranch: "feature/current", - gitStatus, - }), - ).toBe(gitStatus.pr); + selectBranchPr({ branch: "feature/other", gitStatus: status(), branchPr: undefined }), + ).toBeNull(); }); }); diff --git a/apps/web/src/components/ThreadStatusIndicators.tsx b/apps/web/src/components/ThreadStatusIndicators.tsx index af53d1a78b2..b468d7b5b51 100644 --- a/apps/web/src/components/ThreadStatusIndicators.tsx +++ b/apps/web/src/components/ThreadStatusIndicators.tsx @@ -3,7 +3,7 @@ import { scopedThreadKey, scopeThreadRef, } from "@t3tools/client-runtime/environment"; -import type { VcsStatusResult } from "@t3tools/contracts"; +import type { EnvironmentId, VcsStatusResult } from "@t3tools/contracts"; import { CloudIcon, FolderGit2Icon, GitPullRequestIcon, TerminalIcon } from "lucide-react"; import { useMemo } from "react"; import { useEnvironment, usePrimaryEnvironmentId } from "../state/environments"; @@ -110,20 +110,60 @@ export function PrStatusTooltipContent({ status }: { status: PrStatusIndicator } ); } -export function resolveThreadPr(input: { - threadBranch: string | null; +/** + * True when the streamed status describes the thread's own branch, and so is + * authoritative about that branch's PR — including when it reports none. + */ +function statusCoversBranch( + gitStatus: VcsStatusResult | null, + threadBranch: string | null, +): gitStatus is VcsStatusResult { + return gitStatus !== null && threadBranch !== null && gitStatus.refName === threadBranch; +} + +/** + * Pick the PR for `branch` from the two available sources. The streamed + * status wins whenever it describes that branch — including when it reports + * no PR, which is why this short-circuits instead of falling through with + * `??`: a warm branch-keyed result must never resurrect a badge the + * authoritative status just cleared (e.g. right after a merge). + */ +export function selectBranchPr(input: { + branch: string | null; gitStatus: VcsStatusResult | null; + branchPr: ThreadPr | null | undefined; }): ThreadPr | null { - const { threadBranch, gitStatus } = input; - if (gitStatus === null) { - return null; - } - - if (threadBranch === null || gitStatus.refName !== threadBranch) { - return null; + if (statusCoversBranch(input.gitStatus, input.branch)) { + return input.gitStatus.pr ?? null; } + return input.branchPr ?? null; +} - return gitStatus.pr ?? null; +/** + * PR state for a thread, keyed by the thread's recorded branch rather than + * the current checkout. The stream-fed status is the fast path while the + * checkout matches the thread's branch; the branch-keyed query keeps the + * badge alive when the checkout has moved elsewhere. + */ +export function useThreadPr(input: { + environmentId: EnvironmentId | null; + cwd: string | null; + threadBranch: string | null; + gitStatus: VcsStatusResult | null; +}): ThreadPr | null { + const branchPr = useEnvironmentQuery( + input.environmentId !== null && input.threadBranch !== null && input.cwd !== null + ? vcsEnvironment.branchPr({ + environmentId: input.environmentId, + input: { cwd: input.cwd, branch: input.threadBranch }, + }) + : null, + ); + return selectBranchPr({ + branch: input.threadBranch, + gitStatus: input.gitStatus, + branchPr: branchPr.data?.pr, + }); } export function terminalStatusFromRunningIds( @@ -250,7 +290,9 @@ export function ThreadRowLeadingStatus({ thread }: { thread: SidebarThreadSummar }) : null, ); - const pr = resolveThreadPr({ + const pr = useThreadPr({ + environmentId: thread.environmentId, + cwd: gitCwd, threadBranch: thread.branch, gitStatus: gitStatus.data, }); diff --git a/packages/client-runtime/src/state/vcs.ts b/packages/client-runtime/src/state/vcs.ts index a0d4510be7f..337ecc9b4e1 100644 --- a/packages/client-runtime/src/state/vcs.ts +++ b/packages/client-runtime/src/state/vcs.ts @@ -15,7 +15,11 @@ import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom, AtomRegistry } from "effect/unstable/reactivity"; -import { createEnvironmentRpcCommand, createEnvironmentSubscriptionAtomFamily } from "./runtime.ts"; +import { + createEnvironmentRpcCommand, + createEnvironmentRpcQueryAtomFamily, + createEnvironmentSubscriptionAtomFamily, +} from "./runtime.ts"; import type { EnvironmentRegistry } from "../connection/registry.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; @@ -270,6 +274,16 @@ export function createVcsEnvironmentAtoms( return { listRefs, + // PR state per (cwd, branch), independent of the current checkout. The + // server funnels this into the same cached PR lookup that feeds status, + // so a warm badge costs no extra provider calls. + branchPr: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:vcs:branch-pr", + tag: WS_METHODS.vcsBranchPr, + staleTimeMs: 60_000, + idleTtlMs: 5 * 60_000, + refreshIntervalMs: 5 * 60_000, + }), status: createEnvironmentSubscriptionAtomFamily(runtime, { label: "environment-data:vcs:status", subscribe: (input: EnvironmentRpcInput) => diff --git a/packages/contracts/src/git.ts b/packages/contracts/src/git.ts index 2e0552740a6..0883e1983aa 100644 --- a/packages/contracts/src/git.ts +++ b/packages/contracts/src/git.ts @@ -188,9 +188,15 @@ export const VcsInitInput = Schema.Struct({ }); export type VcsInitInput = typeof VcsInitInput.Type; +export const VcsBranchPrInput = Schema.Struct({ + cwd: TrimmedNonEmptyStringSchema, + branch: TrimmedNonEmptyStringSchema, +}); +export type VcsBranchPrInput = typeof VcsBranchPrInput.Type; + // RPC Results -const VcsStatusChangeRequest = Schema.Struct({ +export const VcsStatusChangeRequest = Schema.Struct({ number: PositiveInt, title: TrimmedNonEmptyStringSchema, url: Schema.String, @@ -198,6 +204,7 @@ const VcsStatusChangeRequest = Schema.Struct({ headRef: TrimmedNonEmptyStringSchema, state: VcsStatusChangeRequestState, }); +export type VcsStatusChangeRequest = typeof VcsStatusChangeRequest.Type; const VcsStatusLocalShape = { isRepo: Schema.Boolean, @@ -253,6 +260,17 @@ export const VcsStatusStreamEvent = Schema.Union([ ]); export type VcsStatusStreamEvent = typeof VcsStatusStreamEvent.Type; +/** + * PR/MR state for one branch of a repository, independent of what is + * currently checked out at the cwd. This is what lets a thread keep its PR + * badge after the user (or another thread) switches the checkout elsewhere. + */ +export const VcsBranchPrResult = Schema.Struct({ + branch: TrimmedNonEmptyStringSchema, + pr: Schema.NullOr(VcsStatusChangeRequest), +}); +export type VcsBranchPrResult = typeof VcsBranchPrResult.Type; + export const VcsListRefsResult = Schema.Struct({ refs: Schema.Array(VcsRef), isRepo: Schema.Boolean, diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 42cafae31fc..5fc75faf9b8 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -21,6 +21,8 @@ import { import { AssetAccessError, AssetCreateUrlInput, AssetCreateUrlResult } from "./assets.ts"; import { GitActionProgressEvent, + VcsBranchPrInput, + VcsBranchPrResult, VcsSwitchRefInput, VcsSwitchRefResult, GitCommandError, @@ -191,6 +193,7 @@ export const WS_METHODS = { vcsCreateRef: "vcs.createRef", vcsSwitchRef: "vcs.switchRef", vcsInit: "vcs.init", + vcsBranchPr: "vcs.branchPr", // Git workflow methods gitRunStackedAction: "git.runStackedAction", @@ -558,6 +561,12 @@ export const WsVcsInitRpc = Rpc.make(WS_METHODS.vcsInit, { error: Schema.Union([VcsError, EnvironmentAuthorizationError]), }); +export const WsVcsBranchPrRpc = Rpc.make(WS_METHODS.vcsBranchPr, { + payload: VcsBranchPrInput, + success: VcsBranchPrResult, + error: Schema.Union([GitManagerServiceError, EnvironmentAuthorizationError]), +}); + /** * Ephemeral live diff preview for compact/mobile surfaces. * Not the persisted T3 Review model. Future review sessions should use @@ -838,6 +847,7 @@ export const WsRpcGroup = RpcGroup.make( WsVcsCreateRefRpc, WsVcsSwitchRefRpc, WsVcsInitRpc, + WsVcsBranchPrRpc, WsReviewGetDiffPreviewRpc, WsReviewGetDiffFileContentsRpc, WsTerminalOpenRpc,