| author | |
| committer | |
| log | 4e547318b60031eb546d8cf2b84ad9fa0785a87a |
| tree | a20383626dd241d39906145c022220b20c18efc2 |
| parent | 1641b4aba58ab495cad7a9800b173af3416e16dd |
| signature | Signed by PGP key B5690EEEBB952194 |
Refresh missing PR associations after agent turns on the thread's current branch. Preserve background policy, known PR caches, and failed-lookup backoff. Serialize status loads and refreshes to prevent stale responses from hiding a PR. Find branches pushed under their own name while still tracking the default branch.
Original work by Theo Browne with Claude Fable 5.1 in Claude Code. Takeover fixes created with GPT-6 Astra (preview) in Codex.
Co-authored-by: Theo Browne <me@t3.gg>
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>11 files changed, 788 insertions(+), 60 deletions(-)
apps/server/integration/OrchestrationEngineHarness.integration.ts+2| ... | ... | @@ -363,6 +363,8 @@ export const makeOrchestrationIntegrationHarness = ( |
| 363 | 363 | workingTree: { files: [], insertions: 0, deletions: 0 }, |
| 364 | 364 | }), |
| 365 | 365 | refreshStatus: () => Effect.die("refreshStatus should not be called in this test"), |
| 366 | refreshPullRequestStatus: () => | |
| 367 | Effect.die("refreshPullRequestStatus should not be called in this test"), | |
| 366 | 368 | streamStatus: () => Stream.empty, |
| 367 | 369 | }), |
| 368 | 370 | ), |
apps/server/src/git/GitManager.test.ts+279| ... | ... | @@ -999,6 +999,80 @@ it.layer(GitManagerTestLayer)("GitManager", (it) => { |
| 999 | 999 | }), |
| 1000 | 1000 | ); |
| 1001 | 1001 | |
| 1002 | it.effect("turn-end refresh finds a new PR and keeps known PRs cached", () => | |
| 1003 | Effect.gen(function* () { | |
| 1004 | const repoDir = yield* makeTempDir("t3code-git-manager-"); | |
| 1005 | yield* initRepo(repoDir); | |
| 1006 | const remoteDir = yield* createBareRemote(); | |
| 1007 | yield* runGit(repoDir, ["remote", "add", "origin", remoteDir]); | |
| 1008 | yield* runGit(repoDir, ["push", "-u", "origin", "main"]); | |
| 1009 | yield* runGit(repoDir, ["checkout", "-b", "feature/turn-refresh", "origin/main"]); | |
| 1010 | yield* runGit(repoDir, ["push", "origin", "feature/turn-refresh"]); | |
| 1011 | ||
| 1012 | const { manager, ghCalls } = yield* makeManager({ | |
| 1013 | ghScenario: { | |
| 1014 | prListSequence: [ | |
| 1015 | "[]", | |
| 1016 | // Fake gh returns raw JSON stdout, matching the CLI boundary under test. | |
| 1017 | // @effect-diagnostics-next-line preferSchemaOverJson:off | |
| 1018 | JSON.stringify([ | |
| 1019 | { | |
| 1020 | number: 114, | |
| 1021 | title: "Opened during the turn", | |
| 1022 | url: "https://github.com/pingdotgg/codething-mvp/pull/114", | |
| 1023 | baseRefName: "main", | |
| 1024 | headRefName: "feature/turn-refresh", | |
| 1025 | }, | |
| 1026 | ]), | |
| 1027 | ], | |
| 1028 | }, | |
| 1029 | }); | |
| 1030 | expect((yield* manager.remoteStatus({ cwd: repoDir }))?.pr).toBeNull(); | |
| 1031 | expect( | |
| 1032 | (yield* manager.remoteStatus({ cwd: repoDir }, { refreshUpstream: false }))?.pr, | |
| 1033 | ).toBeNull(); | |
| 1034 | ||
| 1035 | const refreshed = yield* manager.remoteStatus( | |
| 1036 | { cwd: repoDir }, | |
| 1037 | { refreshUpstream: false, refreshMissingPullRequest: true }, | |
| 1038 | ); | |
| 1039 | expect(refreshed?.pr?.number).toBe(114); | |
| 1040 | yield* manager.remoteStatus( | |
| 1041 | { cwd: repoDir }, | |
| 1042 | { refreshUpstream: false, refreshMissingPullRequest: true }, | |
| 1043 | ); | |
| 1044 | expect(ghCalls.filter((call) => call.startsWith("pr list "))).toHaveLength(2); | |
| 1045 | }), | |
| 1046 | ); | |
| 1047 | ||
| 1048 | it.effect("turn-end refresh preserves failed PR lookup backoff", () => | |
| 1049 | Effect.gen(function* () { | |
| 1050 | const repoDir = yield* makeTempDir("t3code-git-manager-"); | |
| 1051 | yield* initRepo(repoDir); | |
| 1052 | const remoteDir = yield* createBareRemote(); | |
| 1053 | yield* runGit(repoDir, ["remote", "add", "origin", remoteDir]); | |
| 1054 | yield* runGit(repoDir, ["checkout", "-b", "feature/rate-limited"]); | |
| 1055 | yield* runGit(repoDir, ["push", "-u", "origin", "feature/rate-limited"]); | |
| 1056 | const { manager, ghCalls } = yield* makeManager({ | |
| 1057 | ghScenario: { | |
| 1058 | failWith: new GitHubCli.GitHubCliUnavailableError({ | |
| 1059 | command: "gh", | |
| 1060 | cwd: repoDir, | |
| 1061 | cause: new Error("rate limited"), | |
| 1062 | }), | |
| 1063 | }, | |
| 1064 | }); | |
| 1065 | yield* manager.remoteStatus({ cwd: repoDir }); | |
| 1066 | const callsAfterFailure = ghCalls.length; | |
| 1067 | yield* manager.remoteStatus( | |
| 1068 | { cwd: repoDir }, | |
| 1069 | { refreshUpstream: false, refreshMissingPullRequest: true }, | |
| 1070 | ); | |
| 1071 | expect(callsAfterFailure).toBeGreaterThan(0); | |
| 1072 | expect(ghCalls).toHaveLength(callsAfterFailure); | |
| 1073 | }), | |
| 1074 | ); | |
| 1075 | ||
| 1002 | 1076 | it.effect("status skips the provider lookup for a branch that was never pushed", () => |
| 1003 | 1077 | Effect.gen(function* () { |
| 1004 | 1078 | const repoDir = yield* makeTempDir("t3code-git-manager-"); |
| ... | ... | @@ -1909,6 +1983,211 @@ it.layer(GitManagerTestLayer)("GitManager", (it) => { |
| 1909 | 1983 | }), |
| 1910 | 1984 | ); |
| 1911 | 1985 | |
| 1986 | it.effect("status finds a PR pushed under the branch's own name despite a default upstream", () => | |
| 1987 | Effect.gen(function* () { | |
| 1988 | const repoDir = yield* makeTempDir("t3code-git-manager-"); | |
| 1989 | yield* initRepo(repoDir); | |
| 1990 | const remoteDir = yield* createBareRemote(); | |
| 1991 | yield* runGit(repoDir, ["remote", "add", "origin", remoteDir]); | |
| 1992 | yield* runGit(repoDir, ["push", "-u", "origin", "main"]); | |
| 1993 | yield* runGit(repoDir, ["remote", "set-head", "origin", "main"]); | |
| 1994 | yield* runGit(repoDir, ["checkout", "-b", "feature/pushed-plain", "origin/main"]); | |
| 1995 | // A plain push (no -u) leaves the upstream on origin/main. | |
| 1996 | yield* runGit(repoDir, ["push", "origin", "feature/pushed-plain"]); | |
| 1997 | ||
| 1998 | const { manager, ghCalls } = yield* makeManager({ | |
| 1999 | ghScenario: { | |
| 2000 | prListByHeadSelector: { | |
| 2001 | // Fake gh returns raw JSON stdout, matching the CLI boundary under test. | |
| 2002 | // @effect-diagnostics-next-line preferSchemaOverJson:off | |
| 2003 | "feature/pushed-plain": JSON.stringify([ | |
| 2004 | { | |
| 2005 | number: 88, | |
| 2006 | title: "Pushed without -u", | |
| 2007 | url: "https://github.com/pingdotgg/codething-mvp/pull/88", | |
| 2008 | baseRefName: "main", | |
| 2009 | headRefName: "feature/pushed-plain", | |
| 2010 | state: "OPEN", | |
| 2011 | updatedAt: "2026-05-01T10:00:00Z", | |
| 2012 | }, | |
| 2013 | ]), | |
| 2014 | }, | |
| 2015 | }, | |
| 2016 | }); | |
| 2017 | ||
| 2018 | const status = yield* manager.status({ cwd: repoDir }); | |
| 2019 | expect(status.refName).toBe("feature/pushed-plain"); | |
| 2020 | expect(status.pr?.number).toBe(88); | |
| 2021 | expect(ghCalls.some((call) => call.includes("--head main"))).toBe(false); | |
| 2022 | }), | |
| 2023 | ); | |
| 2024 | ||
| 2025 | it.effect( | |
| 2026 | "status finds a fork PR pushed under the branch's own name despite a default upstream", | |
| 2027 | () => | |
| 2028 | Effect.gen(function* () { | |
| 2029 | const repoDir = yield* makeTempDir("t3code-git-manager-"); | |
| 2030 | yield* initRepo(repoDir); | |
| 2031 | const originDir = yield* createBareRemote(); | |
| 2032 | const forkDir = yield* createBareRemote(); | |
| 2033 | yield* runGit(repoDir, ["remote", "add", "origin", originDir]); | |
| 2034 | yield* runGit(repoDir, ["push", "-u", "origin", "main"]); | |
| 2035 | yield* runGit(repoDir, ["remote", "set-head", "origin", "main"]); | |
| 2036 | yield* configureRemote(repoDir, "team/fork", forkDir, "team/fork"); | |
| 2037 | yield* runGit(repoDir, ["checkout", "-b", "feature/fork-plain", "origin/main"]); | |
| 2038 | // Pushed to the fork without -u: upstream stays origin/main. | |
| 2039 | yield* runGit(repoDir, ["push", "team/fork", "feature/fork-plain"]); | |
| 2040 | yield* configureVisibleRemoteUrlWithLocalRewrite( | |
| 2041 | repoDir, | |
| 2042 | "origin", | |
| 2043 | "git@github.com:pingdotgg/codething-mvp.git", | |
| 2044 | originDir, | |
| 2045 | ); | |
| 2046 | yield* configureVisibleRemoteUrlWithLocalRewrite( | |
| 2047 | repoDir, | |
| 2048 | "team/fork", | |
| 2049 | "git@github.com:contributor/codething-mvp.git", | |
| 2050 | forkDir, | |
| 2051 | ); | |
| 2052 | ||
| 2053 | const { manager, ghCalls } = yield* makeManager({ | |
| 2054 | ghScenario: { | |
| 2055 | prListByHeadSelector: { | |
| 2056 | // Fake gh returns raw JSON stdout, matching the CLI boundary under test. | |
| 2057 | // @effect-diagnostics-next-line preferSchemaOverJson:off | |
| 2058 | "contributor:feature/fork-plain": JSON.stringify([ | |
| 2059 | { | |
| 2060 | number: 89, | |
| 2061 | title: "Fork PR pushed without -u", | |
| 2062 | url: "https://github.com/pingdotgg/codething-mvp/pull/89", | |
| 2063 | baseRefName: "main", | |
| 2064 | headRefName: "feature/fork-plain", | |
| 2065 | state: "OPEN", | |
| 2066 | updatedAt: "2026-05-01T10:00:00Z", | |
| 2067 | isCrossRepository: true, | |
| 2068 | headRepository: { nameWithOwner: "contributor/codething-mvp" }, | |
| 2069 | headRepositoryOwner: { login: "contributor" }, | |
| 2070 | }, | |
| 2071 | ]), | |
| 2072 | }, | |
| 2073 | }, | |
| 2074 | }); | |
| 2075 | ||
| 2076 | const status = yield* manager.status({ cwd: repoDir }); | |
| 2077 | expect(status.pr?.number).toBe(89); | |
| 2078 | expect(ghCalls.some((call) => call.includes("--head contributor:feature/fork-plain"))).toBe( | |
| 2079 | true, | |
| 2080 | ); | |
| 2081 | expect(ghCalls.some((call) => call.includes("--head main"))).toBe(false); | |
| 2082 | }), | |
| 2083 | ); | |
| 2084 | ||
| 2085 | it.effect("branch PR lookup verifies identity on the fork that holds the own-name ref", () => | |
| 2086 | Effect.gen(function* () { | |
| 2087 | const repoDir = yield* makeTempDir("t3code-git-manager-"); | |
| 2088 | yield* initRepo(repoDir); | |
| 2089 | const originDir = yield* createBareRemote(); | |
| 2090 | const forkDir = yield* createBareRemote(); | |
| 2091 | yield* runGit(repoDir, ["remote", "add", "origin", originDir]); | |
| 2092 | yield* runGit(repoDir, ["push", "-u", "origin", "main"]); | |
| 2093 | yield* runGit(repoDir, ["remote", "set-head", "origin", "main"]); | |
| 2094 | yield* configureRemote(repoDir, "team/fork", forkDir, "team/fork"); | |
| 2095 | yield* runGit(repoDir, ["checkout", "-b", "feature/fork-settle", "origin/main"]); | |
| 2096 | yield* runGit(repoDir, ["push", "team/fork", "feature/fork-settle"]); | |
| 2097 | yield* configureVisibleRemoteUrlWithLocalRewrite( | |
| 2098 | repoDir, | |
| 2099 | "origin", | |
| 2100 | "git@github.com:pingdotgg/codething-mvp.git", | |
| 2101 | originDir, | |
| 2102 | ); | |
| 2103 | yield* configureVisibleRemoteUrlWithLocalRewrite( | |
| 2104 | repoDir, | |
| 2105 | "team/fork", | |
| 2106 | "git@github.com:contributor/codething-mvp.git", | |
| 2107 | forkDir, | |
| 2108 | ); | |
| 2109 | ||
| 2110 | const { manager } = yield* makeManager({ | |
| 2111 | ghScenario: { | |
| 2112 | prListByHeadSelector: { | |
| 2113 | // Fake gh returns raw JSON stdout, matching the CLI boundary under test. | |
| 2114 | // @effect-diagnostics-next-line preferSchemaOverJson:off | |
| 2115 | "contributor:feature/fork-settle": JSON.stringify([ | |
| 2116 | { | |
| 2117 | number: 91, | |
| 2118 | title: "Fork PR to settle", | |
| 2119 | url: "https://github.com/pingdotgg/codething-mvp/pull/91", | |
| 2120 | baseRefName: "main", | |
| 2121 | headRefName: "feature/fork-settle", | |
| 2122 | state: "MERGED", | |
| 2123 | updatedAt: "2026-05-02T10:00:00Z", | |
| 2124 | isCrossRepository: true, | |
| 2125 | headRepository: { nameWithOwner: "contributor/codething-mvp" }, | |
| 2126 | headRepositoryOwner: { login: "contributor" }, | |
| 2127 | }, | |
| 2128 | ]), | |
| 2129 | }, | |
| 2130 | }, | |
| 2131 | }); | |
| 2132 | ||
| 2133 | const pullRequest = yield* manager.branchPullRequest({ | |
| 2134 | cwd: repoDir, | |
| 2135 | branch: "feature/fork-settle", | |
| 2136 | }); | |
| 2137 | ||
| 2138 | expect(pullRequest).toEqual({ | |
| 2139 | state: "merged", | |
| 2140 | updatedAt: "2026-05-02T10:00:00.000Z", | |
| 2141 | }); | |
| 2142 | }), | |
| 2143 | ); | |
| 2144 | ||
| 2145 | it.effect("status keeps an own-name PR when a later lookup fails on a default upstream", () => | |
| 2146 | Effect.gen(function* () { | |
| 2147 | const repoDir = yield* makeTempDir("t3code-git-manager-"); | |
| 2148 | yield* initRepo(repoDir); | |
| 2149 | const remoteDir = yield* createBareRemote(); | |
| 2150 | yield* runGit(repoDir, ["remote", "add", "origin", remoteDir]); | |
| 2151 | yield* runGit(repoDir, ["push", "-u", "origin", "main"]); | |
| 2152 | yield* runGit(repoDir, ["remote", "set-head", "origin", "main"]); | |
| 2153 | yield* runGit(repoDir, ["checkout", "-b", "feature/sticky-plain", "origin/main"]); | |
| 2154 | yield* runGit(repoDir, ["push", "origin", "feature/sticky-plain"]); | |
| 2155 | ||
| 2156 | const { manager } = yield* makeManager({ | |
| 2157 | ghScenario: { | |
| 2158 | prListByHeadSelector: { | |
| 2159 | // Fake gh returns raw JSON stdout, matching the CLI boundary under test. | |
| 2160 | // @effect-diagnostics-next-line preferSchemaOverJson:off | |
| 2161 | "feature/sticky-plain": JSON.stringify([ | |
| 2162 | { | |
| 2163 | number: 90, | |
| 2164 | title: "Sticky own-name PR", | |
| 2165 | url: "https://github.com/pingdotgg/codething-mvp/pull/90", | |
| 2166 | baseRefName: "main", | |
| 2167 | headRefName: "feature/sticky-plain", | |
| 2168 | state: "OPEN", | |
| 2169 | updatedAt: "2026-05-01T10:00:00Z", | |
| 2170 | }, | |
| 2171 | ]), | |
| 2172 | }, | |
| 2173 | failWith: new GitHubCli.GitHubCliUnavailableError({ | |
| 2174 | command: "gh", | |
| 2175 | cwd: repoDir, | |
| 2176 | cause: new Error("rate limited"), | |
| 2177 | }), | |
| 2178 | failAfterCalls: 1, | |
| 2179 | }, | |
| 2180 | }); | |
| 2181 | ||
| 2182 | const first = yield* manager.status({ cwd: repoDir }); | |
| 2183 | expect(first.pr?.number).toBe(90); | |
| 2184 | ||
| 2185 | yield* manager.invalidateStatus(repoDir); | |
| 2186 | const second = yield* manager.status({ cwd: repoDir }); | |
| 2187 | expect(second.pr?.number).toBe(90); | |
| 2188 | }), | |
| 2189 | ); | |
| 2190 | ||
| 1912 | 2191 | it.effect("status prefers open PR when merged PR has newer updatedAt", () => |
| 1913 | 2192 | Effect.gen(function* () { |
| 1914 | 2193 | const repoDir = yield* makeTempDir("t3code-git-manager-"); |
apps/server/src/git/GitManager.ts+130-28| ... | ... | @@ -72,6 +72,11 @@ export interface GitRunStackedActionOptions { |
| 72 | 72 | readonly progressReporter?: GitActionProgressReporter; |
| 73 | 73 | } |
| 74 | 74 | |
| 75 | export interface GitRemoteStatusOptions extends GitVcsDriver.GitRemoteStatusOptions { | |
| 76 | /** Retry a cached missing PR without clearing known PRs or failed lookup backoff. */ | |
| 77 | readonly refreshMissingPullRequest?: boolean; | |
| 78 | } | |
| 79 | ||
| 75 | 80 | interface SourceControlTextGenerationSettings { |
| 76 | 81 | readonly modelSelection: ModelSelection; |
| 77 | 82 | readonly style: SourceControlWritingStyleSettings; |
| ... | ... | @@ -88,7 +93,7 @@ export class GitManager extends Context.Service< |
| 88 | 93 | ) => Effect.Effect<VcsStatusLocalResult, GitManagerServiceError>; |
| 89 | 94 | readonly remoteStatus: ( |
| 90 | 95 | input: VcsStatusInput, |
| 91 | options?: GitVcsDriver.GitRemoteStatusOptions, | |
| 96 | options?: GitRemoteStatusOptions, | |
| 92 | 97 | ) => Effect.Effect<VcsStatusRemoteResult | null, GitManagerServiceError>; |
| 93 | 98 | /** Resolve the PR for a saved branch without changing the current checkout. */ |
| 94 | 99 | readonly branchPullRequest: (input: { |
| ... | ... | @@ -1009,20 +1014,8 @@ export const make = Effect.gen(function* () { |
| 1009 | 1014 | ...(remoteName.length > 0 ? { remoteName } : {}), |
| 1010 | 1015 | }; |
| 1011 | 1016 | return Effect.gen(function* () { |
| 1012 | const headContext = yield* resolveBranchHeadContext(cwd, details); | |
| 1013 | const upstreamHeadIsDefault = | |
| 1014 | headContext.headBranch === details.defaultBranch || | |
| 1015 | (details.defaultBranch === null && | |
| 1016 | (headContext.headBranch === "main" || headContext.headBranch === "master")); | |
| 1017 | // `git worktree add -b feature origin/main` makes the new local branch | |
| 1018 | // track origin/main. That upstream is the branch's base, not its | |
| 1019 | // published PR head. Looking up PRs for it can attach an old reverse | |
| 1020 | // merge from main and auto-settle an unrelated feature thread. | |
| 1021 | if ( | |
| 1022 | headContext.headBranch !== details.branch && | |
| 1023 | upstreamHeadIsDefault && | |
| 1024 | !headContext.isCrossRepository | |
| 1025 | ) { | |
| 1017 | const { headContext, lookup } = yield* resolveLookupHeadContext(cwd, details); | |
| 1018 | if (!lookup) { | |
| 1026 | 1019 | return { latest: null, headContext }; |
| 1027 | 1020 | } |
| 1028 | 1021 | // Only skip when the branch is untracked as well: anything carrying an |
| ... | ... | @@ -1115,11 +1108,21 @@ export const make = Effect.gen(function* () { |
| 1115 | 1108 | defaultBranch: string | null; |
| 1116 | 1109 | isDefaultBranch: boolean; |
| 1117 | 1110 | }, |
| 1111 | refreshMissingPullRequest = false, | |
| 1118 | 1112 | ) { |
| 1119 | 1113 | // Keyed by (cwd, branch) only: the upstream ref changing (e.g. a first |
| 1120 | 1114 | // `push -u`) must not orphan the fallback value for the same branch. |
| 1121 | 1115 | const branchKey = `${cwd}\u0000${details.branch}`; |
| 1122 | return yield* Cache.get(prLookupCache, prLookupCacheKey(cwd, details)).pipe( | |
| 1116 | const cacheKey = prLookupCacheKey(cwd, details); | |
| 1117 | if (refreshMissingPullRequest) { | |
| 1118 | const cached = yield* Cache.getOption(prLookupCache, cacheKey).pipe( | |
| 1119 | Effect.orElseSucceed(() => Option.none()), | |
| 1120 | ); | |
| 1121 | if (Option.isSome(cached) && cached.value.latest === null) { | |
| 1122 | yield* Cache.invalidate(prLookupCache, cacheKey); | |
| 1123 | } | |
| 1124 | } | |
| 1125 | return yield* Cache.get(prLookupCache, cacheKey).pipe( | |
| 1123 | 1126 | Effect.map(({ latest, headContext }) => { |
| 1124 | 1127 | if (!latest) return { pr: null, headContext }; |
| 1125 | 1128 | // On the default branch, only surface open PRs. |
| ... | ... | @@ -1159,8 +1162,8 @@ export const make = Effect.gen(function* () { |
| 1159 | 1162 | } |
| 1160 | 1163 | : {}), |
| 1161 | 1164 | }), |
| 1162 | Effect.andThen(resolveBranchHeadContext(cwd, details)), | |
| 1163 | Effect.map((headContext) => | |
| 1165 | Effect.andThen(resolveLookupHeadContext(cwd, details)), | |
| 1166 | Effect.map(({ headContext }) => | |
| 1164 | 1167 | resolveLastKnownPr(branchKey, { |
| 1165 | 1168 | upstreamRef: details.upstreamRef, |
| 1166 | 1169 | headBranch: headContext.headBranch, |
| ... | ... | @@ -1174,7 +1177,7 @@ export const make = Effect.gen(function* () { |
| 1174 | 1177 | }); |
| 1175 | 1178 | const readRemoteStatus = Effect.fn("readRemoteStatus")(function* ( |
| 1176 | 1179 | cwd: string, |
| 1177 | options?: GitVcsDriver.GitRemoteStatusOptions, | |
| 1180 | options?: GitRemoteStatusOptions, | |
| 1178 | 1181 | ) { |
| 1179 | 1182 | const details = yield* gitCore |
| 1180 | 1183 | .statusDetailsRemote(cwd, options) |
| ... | ... | @@ -1185,12 +1188,16 @@ export const make = Effect.gen(function* () { |
| 1185 | 1188 | |
| 1186 | 1189 | const pr = |
| 1187 | 1190 | details.branch !== null |
| 1188 | ? yield* lookupStatusPr(cwd, { | |
| 1189 | branch: details.branch, | |
| 1190 | upstreamRef: details.upstreamRef, | |
| 1191 | defaultBranch: details.defaultBranch, | |
| 1192 | isDefaultBranch: details.isDefaultBranch, | |
| 1193 | }) | |
| 1191 | ? yield* lookupStatusPr( | |
| 1192 | cwd, | |
| 1193 | { | |
| 1194 | branch: details.branch, | |
| 1195 | upstreamRef: details.upstreamRef, | |
| 1196 | defaultBranch: details.defaultBranch, | |
| 1197 | isDefaultBranch: details.isDefaultBranch, | |
| 1198 | }, | |
| 1199 | options?.refreshMissingPullRequest, | |
| 1200 | ) | |
| 1194 | 1201 | : null; |
| 1195 | 1202 | |
| 1196 | 1203 | return { |
| ... | ... | @@ -1346,6 +1353,96 @@ export const make = Effect.gen(function* () { |
| 1346 | 1353 | } satisfies BranchHeadContext; |
| 1347 | 1354 | }); |
| 1348 | 1355 | |
| 1356 | // The remote that holds a ref named after the local branch, or null when | |
| 1357 | // none does. Remote names may contain slashes, so refs are matched literally | |
| 1358 | // per remote instead of with a glob. When several remotes hold the name, the | |
| 1359 | // preferred remote wins, then origin, then the first configured remote. | |
| 1360 | const findRemoteTrackingRemote = Effect.fn("findRemoteTrackingRemote")(function* ( | |
| 1361 | cwd: string, | |
| 1362 | branch: string, | |
| 1363 | preferredRemoteName: string | null, | |
| 1364 | ) { | |
| 1365 | if (branch.length === 0) return null; | |
| 1366 | return yield* Effect.gen(function* () { | |
| 1367 | const remoteNames = (yield* gitCore.execute({ | |
| 1368 | operation: "GitManager.findRemoteTrackingRemote.remotes", | |
| 1369 | cwd, | |
| 1370 | args: ["remote"], | |
| 1371 | timeoutMs: 5_000, | |
| 1372 | })).stdout | |
| 1373 | .split("\n") | |
| 1374 | .map((name) => name.trim()) | |
| 1375 | .filter((name) => name.length > 0); | |
| 1376 | if (remoteNames.length === 0) return null; | |
| 1377 | const refs = new Set( | |
| 1378 | (yield* gitCore.execute({ | |
| 1379 | operation: "GitManager.findRemoteTrackingRemote.refs", | |
| 1380 | cwd, | |
| 1381 | args: [ | |
| 1382 | "for-each-ref", | |
| 1383 | "--format=%(refname)", | |
| 1384 | ...remoteNames.map((name) => `refs/remotes/${name}/${branch}`), | |
| 1385 | ], | |
| 1386 | timeoutMs: 5_000, | |
| 1387 | })).stdout | |
| 1388 | .split("\n") | |
| 1389 | .map((ref) => ref.trim()) | |
| 1390 | .filter((ref) => ref.length > 0), | |
| 1391 | ); | |
| 1392 | const matching = remoteNames.filter((name) => refs.has(`refs/remotes/${name}/${branch}`)); | |
| 1393 | if (preferredRemoteName !== null && matching.includes(preferredRemoteName)) { | |
| 1394 | return preferredRemoteName; | |
| 1395 | } | |
| 1396 | if (matching.includes("origin")) return "origin"; | |
| 1397 | return matching[0] ?? null; | |
| 1398 | }).pipe(Effect.orElseSucceed(() => null)); | |
| 1399 | }); | |
| 1400 | ||
| 1401 | // `git worktree add -b feature origin/main` makes the new local branch track | |
| 1402 | // origin/main. That upstream is the branch's base, not its published PR | |
| 1403 | // head. Looking up PRs for it can attach an old reverse merge from main and | |
| 1404 | // auto-settle an unrelated feature thread. | |
| 1405 | // | |
| 1406 | // The branch may still have been pushed under its own name by a plain | |
| 1407 | // `git push <remote> feature` that never moved the upstream. When a remote | |
| 1408 | // holds a ref for the local name, look the PR up by that name on that | |
| 1409 | // remote. Without such a ref there is nothing to ask the host about, so | |
| 1410 | // `lookup` is false and no API call is spent. Both the cached lookup and the | |
| 1411 | // failure fallback resolve through here so the last-known PR compares | |
| 1412 | // against the same head branch. | |
| 1413 | const resolveLookupHeadContext = Effect.fn("resolveLookupHeadContext")(function* ( | |
| 1414 | cwd: string, | |
| 1415 | details: { | |
| 1416 | branch: string; | |
| 1417 | upstreamRef: string | null; | |
| 1418 | defaultBranch: string | null; | |
| 1419 | remoteName?: string; | |
| 1420 | }, | |
| 1421 | ) { | |
| 1422 | const headContext = yield* resolveBranchHeadContext(cwd, details); | |
| 1423 | const upstreamHeadIsDefault = | |
| 1424 | headContext.headBranch === details.defaultBranch || | |
| 1425 | (details.defaultBranch === null && | |
| 1426 | (headContext.headBranch === "main" || headContext.headBranch === "master")); | |
| 1427 | if ( | |
| 1428 | headContext.headBranch === details.branch || | |
| 1429 | !upstreamHeadIsDefault || | |
| 1430 | headContext.isCrossRepository | |
| 1431 | ) { | |
| 1432 | return { headContext, lookup: true }; | |
| 1433 | } | |
| 1434 | const remoteName = yield* findRemoteTrackingRemote(cwd, details.branch, headContext.remoteName); | |
| 1435 | if (remoteName === null) { | |
| 1436 | return { headContext, lookup: false }; | |
| 1437 | } | |
| 1438 | const ownNameContext = yield* resolveBranchHeadContext(cwd, { | |
| 1439 | branch: details.branch, | |
| 1440 | upstreamRef: null, | |
| 1441 | remoteName, | |
| 1442 | }); | |
| 1443 | return { headContext: ownNameContext, lookup: true }; | |
| 1444 | }); | |
| 1445 | ||
| 1349 | 1446 | /** |
| 1350 | 1447 | * Whether git has no record of this branch on any remote, so a change request |
| 1351 | 1448 | * cannot exist for it and asking the provider is a guaranteed-empty API call. |
| ... | ... | @@ -1911,7 +2008,7 @@ export const make = Effect.gen(function* () { |
| 1911 | 2008 | const remoteStatus: GitManager["Service"]["remoteStatus"] = Effect.fn("remoteStatus")( |
| 1912 | 2009 | function* (input, options) { |
| 1913 | 2010 | const cacheKey = yield* normalizeStatusCacheKey(input.cwd); |
| 1914 | if (options?.refreshUpstream === false) { | |
| 2011 | if (options?.refreshUpstream === false || options?.refreshMissingPullRequest) { | |
| 1915 | 2012 | return yield* readRemoteStatus(cacheKey, options); |
| 1916 | 2013 | } |
| 1917 | 2014 | return yield* Cache.get(remoteStatusResultCache, cacheKey); |
| ... | ... | @@ -2006,10 +2103,15 @@ export const make = Effect.gen(function* () { |
| 2006 | 2103 | ...(localBranchExists ? {} : { remoteName }), |
| 2007 | 2104 | }); |
| 2008 | 2105 | let cached = yield* Cache.get(prLookupCache, cacheKey); |
| 2106 | // The cached head context may have resolved on a different remote than | |
| 2107 | // the saved upstream: a branch tracking origin/main but pushed to a fork | |
| 2108 | // is looked up on the fork. Verify against the remote the lookup used. | |
| 2109 | const identityRemoteName = (headContext: BranchHeadContext) => | |
| 2110 | headContext.remoteName ?? remoteName ?? undefined; | |
| 2009 | 2111 | const currentIdentity = yield* resolvePrLookupRepositoryIdentity( |
| 2010 | 2112 | cacheCwd, |
| 2011 | 2113 | branch, |
| 2012 | remoteName ?? undefined, | |
| 2114 | identityRemoteName(cached.headContext), | |
| 2013 | 2115 | ); |
| 2014 | 2116 | const canVerifyIdentity = (headContext: BranchHeadContext, identity: typeof currentIdentity) => |
| 2015 | 2117 | !( |
| ... | ... | @@ -2032,7 +2134,7 @@ export const make = Effect.gen(function* () { |
| 2032 | 2134 | const refreshedIdentity = yield* resolvePrLookupRepositoryIdentity( |
| 2033 | 2135 | cacheCwd, |
| 2034 | 2136 | branch, |
| 2035 | remoteName ?? undefined, | |
| 2137 | identityRemoteName(cached.headContext), | |
| 2036 | 2138 | ); |
| 2037 | 2139 | if ( |
| 2038 | 2140 | !canVerifyIdentity(cached.headContext, refreshedIdentity) || |
apps/server/src/git/GitWorkflowService.ts+1-1| ... | ... | @@ -43,7 +43,7 @@ export class GitWorkflowService extends Context.Service< |
| 43 | 43 | ) => Effect.Effect<VcsStatusLocalResult, GitManagerServiceError>; |
| 44 | 44 | readonly remoteStatus: ( |
| 45 | 45 | input: VcsStatusInput, |
| 46 | options?: GitVcsDriver.GitRemoteStatusOptions, | |
| 46 | options?: GitManager.GitRemoteStatusOptions, | |
| 47 | 47 | ) => Effect.Effect<VcsStatusRemoteResult | null, GitManagerServiceError>; |
| 48 | 48 | readonly invalidateLocalStatus: (cwd: string) => Effect.Effect<void, never>; |
| 49 | 49 | readonly invalidateRemoteStatus: (cwd: string) => Effect.Effect<void, never>; |
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+92-2| ... | ... | @@ -295,6 +295,7 @@ describe("CheckpointReactor", () => { |
| 295 | 295 | readonly providerSessionCwd?: string; |
| 296 | 296 | readonly providerName?: ProviderDriverKind; |
| 297 | 297 | readonly gitStatusRefreshCalls?: Array<string>; |
| 298 | readonly pullRequestRefreshCalls?: Array<string>; | |
| 298 | 299 | }) { |
| 299 | 300 | const cwd = createGitRepository(); |
| 300 | 301 | tempDirs.push(cwd); |
| ... | ... | @@ -333,7 +334,8 @@ describe("CheckpointReactor", () => { |
| 333 | 334 | Effect.as({ |
| 334 | 335 | isRepo: true, |
| 335 | 336 | hasPrimaryRemote: false, |
| 336 | isDefaultRef: true, | |
| 337 | isDefaultRef: | |
| 338 | options?.localStatusRefName === undefined || options.localStatusRefName === "main", | |
| 337 | 339 | refName: |
| 338 | 340 | options?.localStatusRefName !== undefined ? options.localStatusRefName : "main", |
| 339 | 341 | hasWorkingTreeChanges: false, |
| ... | ... | @@ -341,6 +343,10 @@ describe("CheckpointReactor", () => { |
| 341 | 343 | }), |
| 342 | 344 | ), |
| 343 | 345 | refreshStatus: () => Effect.die("refreshStatus should not be called in this test"), |
| 346 | refreshPullRequestStatus: (cwd: string) => | |
| 347 | Effect.sync(() => { | |
| 348 | options?.pullRequestRefreshCalls?.push(cwd); | |
| 349 | }).pipe(Effect.as(null)), | |
| 344 | 350 | streamStatus: () => Stream.empty, |
| 345 | 351 | }); |
| 346 | 352 | |
| ... | ... | @@ -561,6 +567,78 @@ describe("CheckpointReactor", () => { |
| 561 | 567 | expect(gitStatusRefreshCalls).toEqual([harness.cwd]); |
| 562 | 568 | }); |
| 563 | 569 | |
| 570 | it("re-asks for the pull request at turn end when the thread branch is checked out", async () => { | |
| 571 | const pullRequestRefreshCalls: string[] = []; | |
| 572 | const harness = await createHarness({ | |
| 573 | seedFilesystemCheckpoints: false, | |
| 574 | threadBranch: "t3code/feature", | |
| 575 | localStatusRefName: "t3code/feature", | |
| 576 | pullRequestRefreshCalls, | |
| 577 | }); | |
| 578 | ||
| 579 | harness.provider.emit({ | |
| 580 | type: "turn.completed", | |
| 581 | eventId: EventId.make("evt-turn-completed-refresh-pr"), | |
| 582 | provider: ProviderDriverKind.make("codex"), | |
| 583 | createdAt: "2026-01-01T00:00:00.000Z", | |
| 584 | threadId: ThreadId.make("thread-1"), | |
| 585 | turnId: asTurnId("turn-refresh-pr"), | |
| 586 | payload: { state: "completed" }, | |
| 587 | }); | |
| 588 | ||
| 589 | await harness.drain(); | |
| 590 | ||
| 591 | expect(pullRequestRefreshCalls).toEqual([harness.cwd]); | |
| 592 | }); | |
| 593 | ||
| 594 | it("re-asks for the pull request after adopting a drifted checkout", async () => { | |
| 595 | const pullRequestRefreshCalls: string[] = []; | |
| 596 | const harness = await createHarness({ | |
| 597 | seedFilesystemCheckpoints: false, | |
| 598 | threadBranch: "t3code/original-branch", | |
| 599 | localStatusRefName: "t3code/renamed-by-agent", | |
| 600 | pullRequestRefreshCalls, | |
| 601 | }); | |
| 602 | ||
| 603 | harness.provider.emit({ | |
| 604 | type: "turn.completed", | |
| 605 | eventId: EventId.make("evt-turn-completed-drift-pr"), | |
| 606 | provider: ProviderDriverKind.make("codex"), | |
| 607 | createdAt: "2026-01-01T00:00:00.000Z", | |
| 608 | threadId: ThreadId.make("thread-1"), | |
| 609 | turnId: asTurnId("turn-drift-pr"), | |
| 610 | payload: { state: "completed" }, | |
| 611 | }); | |
| 612 | ||
| 613 | await harness.drain(); | |
| 614 | ||
| 615 | expect(pullRequestRefreshCalls).toEqual([harness.cwd]); | |
| 616 | }); | |
| 617 | ||
| 618 | it("does not re-ask for the pull request at turn end on the default branch", async () => { | |
| 619 | const pullRequestRefreshCalls: string[] = []; | |
| 620 | const harness = await createHarness({ | |
| 621 | seedFilesystemCheckpoints: false, | |
| 622 | threadBranch: "main", | |
| 623 | localStatusRefName: "main", | |
| 624 | pullRequestRefreshCalls, | |
| 625 | }); | |
| 626 | ||
| 627 | harness.provider.emit({ | |
| 628 | type: "turn.completed", | |
| 629 | eventId: EventId.make("evt-turn-completed-no-pr-refresh"), | |
| 630 | provider: ProviderDriverKind.make("codex"), | |
| 631 | createdAt: "2026-01-01T00:00:00.000Z", | |
| 632 | threadId: ThreadId.make("thread-1"), | |
| 633 | turnId: asTurnId("turn-no-pr-refresh"), | |
| 634 | payload: { state: "completed" }, | |
| 635 | }); | |
| 636 | ||
| 637 | await harness.drain(); | |
| 638 | ||
| 639 | expect(pullRequestRefreshCalls).toEqual([]); | |
| 640 | }); | |
| 641 | ||
| 564 | 642 | it("adopts a drifted checkout as the thread branch on a dedicated worktree", async () => { |
| 565 | 643 | const harness = await createHarness({ |
| 566 | 644 | seedFilesystemCheckpoints: false, |
| ... | ... | @@ -593,11 +671,13 @@ describe("CheckpointReactor", () => { |
| 593 | 671 | }); |
| 594 | 672 | |
| 595 | 673 | it("does not adopt a drifted checkout when the worktree is shared by another thread", async () => { |
| 674 | const pullRequestRefreshCalls: string[] = []; | |
| 596 | 675 | const harness = await createHarness({ |
| 597 | 676 | seedFilesystemCheckpoints: false, |
| 598 | 677 | threadBranch: "t3code/original-branch", |
| 599 | 678 | localStatusRefName: "t3code/renamed-by-agent", |
| 600 | 679 | secondThreadSharingWorktree: true, |
| 680 | pullRequestRefreshCalls, | |
| 601 | 681 | }); |
| 602 | 682 | |
| 603 | 683 | harness.provider.emit({ |
| ... | ... | @@ -615,6 +695,7 @@ describe("CheckpointReactor", () => { |
| 615 | 695 | const snapshot = await harness.readModel(); |
| 616 | 696 | const thread = snapshot.threads.find((entry) => entry.id === ThreadId.make("thread-1")); |
| 617 | 697 | expect(thread?.branch).toBe("t3code/original-branch"); |
| 698 | expect(pullRequestRefreshCalls).toEqual([]); | |
| 618 | 699 | }); |
| 619 | 700 | |
| 620 | 701 | it("does not adopt a temporary placeholder checkout as the thread branch", async () => { |
| ... | ... | @@ -642,7 +723,13 @@ describe("CheckpointReactor", () => { |
| 642 | 723 | }); |
| 643 | 724 | |
| 644 | 725 | it("ignores auxiliary thread turn completion while primary turn is active", async () => { |
| 645 | const harness = await createHarness({ seedFilesystemCheckpoints: false }); | |
| 726 | const pullRequestRefreshCalls: string[] = []; | |
| 727 | const harness = await createHarness({ | |
| 728 | seedFilesystemCheckpoints: false, | |
| 729 | threadBranch: "t3code/feature", | |
| 730 | localStatusRefName: "t3code/feature", | |
| 731 | pullRequestRefreshCalls, | |
| 732 | }); | |
| 646 | 733 | const createdAt = "2026-01-01T00:00:00.000Z"; |
| 647 | 734 | |
| 648 | 735 | await Effect.runPromise( |
| ... | ... | @@ -694,6 +781,7 @@ describe("CheckpointReactor", () => { |
| 694 | 781 | const midReadModel = await harness.readModel(); |
| 695 | 782 | const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); |
| 696 | 783 | expect(midThread?.checkpoints).toHaveLength(0); |
| 784 | expect(pullRequestRefreshCalls).toEqual([]); | |
| 697 | 785 | |
| 698 | 786 | harness.provider.emit({ |
| 699 | 787 | type: "turn.completed", |
| ... | ... | @@ -711,6 +799,8 @@ describe("CheckpointReactor", () => { |
| 711 | 799 | (entry) => entry.latestTurn?.turnId === "turn-main" && entry.checkpoints.length === 1, |
| 712 | 800 | ); |
| 713 | 801 | expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1); |
| 802 | await harness.drain(); | |
| 803 | expect(pullRequestRefreshCalls).toEqual([harness.cwd]); | |
| 714 | 804 | }); |
| 715 | 805 | |
| 716 | 806 | it("captures pre-turn and completion checkpoints for claude runtime events", async () => { |
apps/server/src/orchestration/Layers/CheckpointReactor.ts+35-3| ... | ... | @@ -553,9 +553,42 @@ const make = Effect.gen(function* () { |
| 553 | 553 | cwd: sessionRuntime.value.cwd, |
| 554 | 554 | local, |
| 555 | 555 | }); |
| 556 | yield* refreshPullRequestAfterTurn({ | |
| 557 | threadId: event.threadId, | |
| 558 | turnId: toTurnId(event.turnId), | |
| 559 | cwd: sessionRuntime.value.cwd, | |
| 560 | local, | |
| 561 | }); | |
| 556 | 562 | } |
| 557 | 563 | }); |
| 558 | 564 | |
| 565 | // Retry a missing PR after the agent finishes its push and PR creation. | |
| 566 | // Re-read the projected branch after drift adoption. A rejected metadata | |
| 567 | // update must not let this thread refresh another thread's checkout. | |
| 568 | const refreshPullRequestAfterTurn = Effect.fn("refreshPullRequestAfterTurn")(function* (input: { | |
| 569 | readonly threadId: ThreadId; | |
| 570 | readonly turnId: TurnId | null; | |
| 571 | readonly cwd: string; | |
| 572 | readonly local: VcsStatusLocalResult; | |
| 573 | }) { | |
| 574 | const checkedOutBranch = input.local.refName; | |
| 575 | if (checkedOutBranch === null || input.local.isDefaultRef) return; | |
| 576 | const thread = yield* projectionSnapshotQuery | |
| 577 | .getThreadShellById(input.threadId) | |
| 578 | .pipe(Effect.map(Option.getOrUndefined)); | |
| 579 | if (!thread || thread.branch !== checkedOutBranch) return; | |
| 580 | if (thread.session?.activeTurnId && !sameId(thread.session.activeTurnId, input.turnId)) return; | |
| 581 | yield* vcsStatusBroadcaster.refreshPullRequestStatus(input.cwd).pipe( | |
| 582 | Effect.catch((error) => | |
| 583 | Effect.logWarning("failed to refresh pull request status after turn completion", { | |
| 584 | threadId: input.threadId, | |
| 585 | cwd: input.cwd, | |
| 586 | detail: error.message, | |
| 587 | }), | |
| 588 | ), | |
| 589 | ); | |
| 590 | }); | |
| 591 | ||
| 559 | 592 | // A `git checkout` run inside a thread's dedicated worktree (by an agent or |
| 560 | 593 | // the user) bypasses T3's commands, so the thread's recorded branch goes |
| 561 | 594 | // stale. Since #4460 the client only attributes PR state to a thread when |
| ... | ... | @@ -843,9 +876,8 @@ const make = Effect.gen(function* () { |
| 843 | 876 | |
| 844 | 877 | // When ProviderRuntimeIngestion creates a placeholder checkpoint (status "missing") |
| 845 | 878 | // from a turn.diff.updated runtime event, capture the real git checkpoint to |
| 846 | // replace it. The providerService.streamEvents PubSub does not reliably deliver | |
| 847 | // turn.completed runtime events to this reactor (shared subscription), so | |
| 848 | // reacting to the domain event is the reliable path. | |
| 879 | // replace it. ProviderService broadcasts runtime events to each subscriber. | |
| 880 | // This domain-event path also captures checkpoints from turn diff updates. | |
| 849 | 881 | if (event.type === "thread.turn-diff-completed") { |
| 850 | 882 | yield* captureCheckpointFromPlaceholder(event).pipe( |
| 851 | 883 | Effect.catch((error) => |
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+2| ... | ... | @@ -461,6 +461,8 @@ describe("ProviderCommandReactor", () => { |
| 461 | 461 | refreshLocalStatus: () => |
| 462 | 462 | Effect.die("refreshLocalStatus should not be called in this test"), |
| 463 | 463 | refreshStatus, |
| 464 | refreshPullRequestStatus: () => | |
| 465 | Effect.die("refreshPullRequestStatus should not be called in this test"), | |
| 464 | 466 | streamStatus: () => Stream.die("streamStatus should not be called in this test"), |
| 465 | 467 | }), |
| 466 | 468 | ), |
apps/server/src/vcs/VcsStatusBroadcaster.test.ts+146-1| ... | ... | @@ -5,6 +5,7 @@ import * as Deferred from "effect/Deferred"; |
| 5 | 5 | import * as DateTime from "effect/DateTime"; |
| 6 | 6 | import * as Duration from "effect/Duration"; |
| 7 | 7 | import * as Effect from "effect/Effect"; |
| 8 | import * as Fiber from "effect/Fiber"; | |
| 8 | 9 | import * as Exit from "effect/Exit"; |
| 9 | 10 | import * as FileSystem from "effect/FileSystem"; |
| 10 | 11 | import * as Layer from "effect/Layer"; |
| ... | ... | @@ -75,10 +76,11 @@ function makeTestLayer(state: { |
| 75 | 76 | localInvalidationCalls: number; |
| 76 | 77 | remoteInvalidationCalls: number; |
| 77 | 78 | remoteStatusRefreshUpstreamValues?: Array<boolean | undefined>; |
| 79 | backgroundWorkEnabled?: boolean; | |
| 78 | 80 | }) { |
| 79 | 81 | return VcsStatusBroadcaster.layer.pipe( |
| 80 | 82 | Layer.provideMerge(NodeServices.layer), |
| 81 | Layer.provide(makeBackgroundPolicyLayer(() => true)), | |
| 83 | Layer.provide(makeBackgroundPolicyLayer(() => state.backgroundWorkEnabled !== false)), | |
| 82 | 84 | Layer.provide( |
| 83 | 85 | Layer.mock(GitWorkflowService.GitWorkflowService)({ |
| 84 | 86 | localStatus: () => |
| ... | ... | @@ -228,6 +230,149 @@ describe("VcsStatusBroadcaster", () => { |
| 228 | 230 | }).pipe(Effect.provide(makeTestLayer(state))); |
| 229 | 231 | }); |
| 230 | 232 | |
| 233 | it.effect("refreshes a loaded cwd without reusing a previous branch's PR", () => { | |
| 234 | const state = { | |
| 235 | currentLocalStatus: baseLocalStatus, | |
| 236 | currentRemoteStatus: baseRemoteStatus, | |
| 237 | localStatusCalls: 0, | |
| 238 | remoteStatusCalls: 0, | |
| 239 | localInvalidationCalls: 0, | |
| 240 | remoteInvalidationCalls: 0, | |
| 241 | }; | |
| 242 | ||
| 243 | return Effect.gen(function* () { | |
| 244 | const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; | |
| 245 | ||
| 246 | // Nobody loaded this cwd yet: no host request is spent. | |
| 247 | assert.isNull(yield* broadcaster.refreshPullRequestStatus("/repo")); | |
| 248 | assert.equal(state.remoteStatusCalls, 0); | |
| 249 | ||
| 250 | yield* broadcaster.getStatus({ cwd: "/repo" }); | |
| 251 | assert.equal(state.remoteStatusCalls, 1); | |
| 252 | ||
| 253 | // Loaded and no PR known: ask GitManager to retry the missing PR. | |
| 254 | state.currentRemoteStatus = remoteStatusWithPr; | |
| 255 | const refreshed = yield* broadcaster.refreshPullRequestStatus("/repo"); | |
| 256 | assert.deepStrictEqual(refreshed, remoteStatusWithPr); | |
| 257 | assert.equal(state.remoteStatusCalls, 2); | |
| 258 | assert.equal(state.remoteInvalidationCalls, 0); | |
| 259 | ||
| 260 | // The agent switches branches. The previous branch's PR must not block a read. | |
| 261 | state.currentLocalStatus = { ...baseLocalStatus, refName: "feature/next" }; | |
| 262 | state.currentRemoteStatus = baseRemoteStatus; | |
| 263 | yield* broadcaster.refreshLocalStatus("/repo"); | |
| 264 | const refreshedBranch = yield* broadcaster.refreshPullRequestStatus("/repo"); | |
| 265 | assert.deepStrictEqual(refreshedBranch, baseRemoteStatus); | |
| 266 | assert.equal(state.remoteStatusCalls, 3); | |
| 267 | }).pipe(Effect.provide(makeTestLayer(state))); | |
| 268 | }); | |
| 269 | ||
| 270 | it.effect("a poll that started before the turn-end refresh cannot overwrite its PR", () => { | |
| 271 | const releaseFirstPoll = Deferred.makeUnsafe<void>(); | |
| 272 | const firstPollStarted = Deferred.makeUnsafe<void>(); | |
| 273 | let remoteReads = 0; | |
| 274 | const layer = VcsStatusBroadcaster.layer.pipe( | |
| 275 | Layer.provideMerge(NodeServices.layer), | |
| 276 | Layer.provide(makeBackgroundPolicyLayer(() => true)), | |
| 277 | Layer.provide( | |
| 278 | Layer.mock(GitWorkflowService.GitWorkflowService)({ | |
| 279 | localStatus: () => Effect.succeed(baseLocalStatus), | |
| 280 | remoteStatus: () => | |
| 281 | Effect.gen(function* () { | |
| 282 | remoteReads += 1; | |
| 283 | if (remoteReads === 2) { | |
| 284 | // Hold an older empty response while the turn-end refresh queues. | |
| 285 | yield* Deferred.succeed(firstPollStarted, undefined); | |
| 286 | yield* Deferred.await(releaseFirstPoll); | |
| 287 | return baseRemoteStatus; | |
| 288 | } | |
| 289 | return remoteReads === 1 ? baseRemoteStatus : remoteStatusWithPr; | |
| 290 | }), | |
| 291 | invalidateLocalStatus: () => Effect.void, | |
| 292 | invalidateRemoteStatus: () => Effect.void, | |
| 293 | invalidateStatus: () => Effect.void, | |
| 294 | }), | |
| 295 | ), | |
| 296 | ); | |
| 297 | ||
| 298 | return Effect.gen(function* () { | |
| 299 | const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; | |
| 300 | yield* broadcaster.getStatus({ cwd: "/repo" }); | |
| 301 | ||
| 302 | const poll = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped); | |
| 303 | yield* Deferred.await(firstPollStarted); | |
| 304 | const refresh = yield* broadcaster.refreshPullRequestStatus("/repo").pipe(Effect.forkScoped); | |
| 305 | yield* Deferred.succeed(releaseFirstPoll, undefined); | |
| 306 | yield* Fiber.join(poll); | |
| 307 | const refreshed = yield* Fiber.join(refresh); | |
| 308 | ||
| 309 | assert.deepStrictEqual(refreshed, remoteStatusWithPr); | |
| 310 | const final = yield* broadcaster.getStatus({ cwd: "/repo" }); | |
| 311 | assert.deepStrictEqual(final.pr, remoteStatusWithPr.pr); | |
| 312 | }).pipe(Effect.provide(layer), Effect.scoped); | |
| 313 | }); | |
| 314 | ||
| 315 | it.effect("an initial status read cannot overwrite an explicit refresh", () => { | |
| 316 | const firstReadStarted = Deferred.makeUnsafe<void>(); | |
| 317 | const releaseFirstRead = Deferred.makeUnsafe<void>(); | |
| 318 | let remoteReads = 0; | |
| 319 | const layer = VcsStatusBroadcaster.layer.pipe( | |
| 320 | Layer.provide(FileSystem.layerNoop({ realPath: (path) => Effect.succeed(path) })), | |
| 321 | Layer.provideMerge(NodeServices.layer), | |
| 322 | Layer.provide(makeBackgroundPolicyLayer(() => true)), | |
| 323 | Layer.provide( | |
| 324 | Layer.mock(GitWorkflowService.GitWorkflowService)({ | |
| 325 | localStatus: () => Effect.succeed(baseLocalStatus), | |
| 326 | remoteStatus: () => | |
| 327 | Effect.gen(function* () { | |
| 328 | remoteReads += 1; | |
| 329 | if (remoteReads === 1) { | |
| 330 | yield* Deferred.succeed(firstReadStarted, undefined); | |
| 331 | yield* Deferred.await(releaseFirstRead); | |
| 332 | return baseRemoteStatus; | |
| 333 | } | |
| 334 | return remoteStatusWithPr; | |
| 335 | }), | |
| 336 | invalidateStatus: () => Effect.void, | |
| 337 | }), | |
| 338 | ), | |
| 339 | ); | |
| 340 | return Effect.gen(function* () { | |
| 341 | const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; | |
| 342 | const initial = yield* broadcaster.getStatus({ cwd: "/repo" }).pipe(Effect.forkScoped); | |
| 343 | yield* Deferred.await(firstReadStarted); | |
| 344 | const refresh = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped); | |
| 345 | // Run ready fibers before releasing the delayed first read. | |
| 346 | yield* TestClock.adjust(Duration.zero); | |
| 347 | yield* Deferred.succeed(releaseFirstRead, undefined); | |
| 348 | yield* Fiber.join(initial); | |
| 349 | yield* Fiber.join(refresh); | |
| 350 | assert.deepStrictEqual( | |
| 351 | (yield* broadcaster.getStatus({ cwd: "/repo" })).pr, | |
| 352 | remoteStatusWithPr.pr, | |
| 353 | ); | |
| 354 | }).pipe(Effect.provide(layer), Effect.scoped); | |
| 355 | }); | |
| 356 | ||
| 357 | it.effect("turn-end refresh skips a loaded cwd when background policy pauses it", () => { | |
| 358 | const state = { | |
| 359 | currentLocalStatus: baseLocalStatus, | |
| 360 | currentRemoteStatus: baseRemoteStatus, | |
| 361 | localStatusCalls: 0, | |
| 362 | remoteStatusCalls: 0, | |
| 363 | localInvalidationCalls: 0, | |
| 364 | remoteInvalidationCalls: 0, | |
| 365 | backgroundWorkEnabled: false, | |
| 366 | }; | |
| 367 | return Effect.gen(function* () { | |
| 368 | const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; | |
| 369 | yield* broadcaster.getStatus({ cwd: "/repo" }); | |
| 370 | yield* broadcaster.refreshPullRequestStatus("/repo"); | |
| 371 | assert.equal(state.remoteStatusCalls, 1); | |
| 372 | assert.equal(state.remoteInvalidationCalls, 0); | |
| 373 | }).pipe(Effect.provide(makeTestLayer(state))); | |
| 374 | }); | |
| 375 | ||
| 231 | 376 | it.effect("refreshes the cached snapshot after explicit invalidation", () => { |
| 232 | 377 | const state = { |
| 233 | 378 | currentLocalStatus: baseLocalStatus, |
apps/server/src/vcs/VcsStatusBroadcaster.ts+87-21| ... | ... | @@ -10,6 +10,7 @@ import * as PubSub from "effect/PubSub"; |
| 10 | 10 | import * as Ref from "effect/Ref"; |
| 11 | 11 | import * as Schedule from "effect/Schedule"; |
| 12 | 12 | import * as Scope from "effect/Scope"; |
| 13 | import * as Semaphore from "effect/Semaphore"; | |
| 13 | 14 | import * as Stream from "effect/Stream"; |
| 14 | 15 | import * as SynchronizedRef from "effect/SynchronizedRef"; |
| 15 | 16 | import type { |
| ... | ... | @@ -184,6 +185,14 @@ export class VcsStatusBroadcaster extends Context.Service< |
| 184 | 185 | cwd: string, |
| 185 | 186 | ) => Effect.Effect<VcsStatusLocalResult, GitManagerServiceError>; |
| 186 | 187 | readonly refreshStatus: (cwd: string) => Effect.Effect<VcsStatusResult, GitManagerServiceError>; |
| 188 | /** | |
| 189 | * Refresh a loaded cwd after a turn if background policy allows it. | |
| 190 | * GitManager retries missing PRs for the current branch and keeps known | |
| 191 | * PRs and failed lookup backoff cached. This does not fetch Git remotes. | |
| 192 | */ | |
| 193 | readonly refreshPullRequestStatus: ( | |
| 194 | cwd: string, | |
| 195 | ) => Effect.Effect<VcsStatusRemoteResult | null, GitManagerServiceError>; | |
| 187 | 196 | readonly streamStatus: ( |
| 188 | 197 | input: VcsStatusInput, |
| 189 | 198 | options?: StreamStatusOptions, |
| ... | ... | @@ -214,6 +223,18 @@ export const make = Effect.gen(function* () { |
| 214 | 223 | Scope.close(scope, Exit.void), |
| 215 | 224 | ); |
| 216 | 225 | const cacheRef = yield* Ref.make(new Map<string, CachedVcsStatus>()); |
| 226 | // One permit per cwd for remote reads that write the cache. Without it a | |
| 227 | // periodic poll that started before `gh pr create` can finish after the | |
| 228 | // turn-end refresh and overwrite the fresh PR with its stale `pr: null`. | |
| 229 | const remoteWriteLocks = new Map<string, Semaphore.Semaphore>(); | |
| 230 | const withRemoteWriteLock = <A, E, R>(cwd: string, effect: Effect.Effect<A, E, R>) => { | |
| 231 | let lock = remoteWriteLocks.get(cwd); | |
| 232 | if (lock === undefined) { | |
| 233 | lock = Semaphore.makeUnsafe(1); | |
| 234 | remoteWriteLocks.set(cwd, lock); | |
| 235 | } | |
| 236 | return lock.withPermits(1)(effect); | |
| 237 | }; | |
| 217 | 238 | const pollersRef = yield* SynchronizedRef.make(new Map<string, ActiveRemotePoller>()); |
| 218 | 239 | |
| 219 | 240 | const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* ( |
| ... | ... | @@ -351,14 +372,20 @@ export const make = Effect.gen(function* () { |
| 351 | 372 | if (cached?.local && cached.remote) { |
| 352 | 373 | return mergeGitStatusParts(cached.local.value, cached.remote.value); |
| 353 | 374 | } |
| 354 | const [local, remote] = yield* Effect.all( | |
| 355 | [ | |
| 356 | cached?.local ? Effect.succeed(cached.local.value) : workflow.localStatus({ cwd }), | |
| 357 | cached?.remote ? Effect.succeed(cached.remote.value) : workflow.remoteStatus({ cwd }), | |
| 358 | ], | |
| 359 | { concurrency: "unbounded" }, | |
| 375 | return yield* withRemoteWriteLock( | |
| 376 | cwd, | |
| 377 | Effect.gen(function* () { | |
| 378 | const latest = yield* getCachedStatus(cwd); | |
| 379 | const [local, remote] = yield* Effect.all( | |
| 380 | [ | |
| 381 | latest?.local ? Effect.succeed(latest.local.value) : workflow.localStatus({ cwd }), | |
| 382 | latest?.remote ? Effect.succeed(latest.remote.value) : workflow.remoteStatus({ cwd }), | |
| 383 | ], | |
| 384 | { concurrency: "unbounded" }, | |
| 385 | ); | |
| 386 | return yield* updateCachedStatus(cwd, local, remote); | |
| 387 | }), | |
| 360 | 388 | ); |
| 361 | return yield* updateCachedStatus(cwd, local, remote); | |
| 362 | 389 | }); |
| 363 | 390 | |
| 364 | 391 | const refreshLocalStatusCore = Effect.fn("VcsStatusBroadcaster.refreshLocalStatusCore")( |
| ... | ... | @@ -421,13 +448,18 @@ export const make = Effect.gen(function* () { |
| 421 | 448 | readonly policyCwds?: ReadonlyArray<string>; |
| 422 | 449 | }, |
| 423 | 450 | ) { |
| 424 | if (options?.refreshUpstream !== false) { | |
| 425 | yield* workflow.invalidateRemoteStatus(cwd); | |
| 426 | } | |
| 427 | const remote = yield* workflow.remoteStatus({ cwd }, options); | |
| 428 | const pulled = yield* maybeAutoPull(cwd, remote, options?.policyCwds ?? [cwd]); | |
| 429 | if (pulled !== null) return pulled.remote; | |
| 430 | return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); | |
| 451 | return yield* withRemoteWriteLock( | |
| 452 | cwd, | |
| 453 | Effect.gen(function* () { | |
| 454 | if (options?.refreshUpstream !== false) { | |
| 455 | yield* workflow.invalidateRemoteStatus(cwd); | |
| 456 | } | |
| 457 | const remote = yield* workflow.remoteStatus({ cwd }, options); | |
| 458 | const pulled = yield* maybeAutoPull(cwd, remote, options?.policyCwds ?? [cwd]); | |
| 459 | if (pulled !== null) return pulled.remote; | |
| 460 | return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); | |
| 461 | }), | |
| 462 | ); | |
| 431 | 463 | }); |
| 432 | 464 | |
| 433 | 465 | const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn( |
| ... | ... | @@ -436,16 +468,49 @@ export const make = Effect.gen(function* () { |
| 436 | 468 | const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); |
| 437 | 469 | // invalidateStatus (not the two partial invalidations) so an explicit |
| 438 | 470 | // refresh also bypasses GitManager's slow PR-lookup cache. |
| 439 | yield* workflow.invalidateStatus(cwd); | |
| 440 | const [local, remote] = yield* Effect.all( | |
| 441 | [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd })], | |
| 442 | { concurrency: "unbounded" }, | |
| 471 | return yield* withRemoteWriteLock( | |
| 472 | cwd, | |
| 473 | Effect.gen(function* () { | |
| 474 | yield* workflow.invalidateStatus(cwd); | |
| 475 | const [local, remote] = yield* Effect.all( | |
| 476 | [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd })], | |
| 477 | { concurrency: "unbounded" }, | |
| 478 | ); | |
| 479 | const pulled = yield* maybeAutoPull(cwd, remote, [rawCwd]); | |
| 480 | if (pulled !== null) return mergeGitStatusParts(pulled.local, pulled.remote); | |
| 481 | return yield* updateCachedStatus(cwd, local, remote, { publish: true }); | |
| 482 | }), | |
| 443 | 483 | ); |
| 444 | const pulled = yield* maybeAutoPull(cwd, remote, [rawCwd]); | |
| 445 | if (pulled !== null) return mergeGitStatusParts(pulled.local, pulled.remote); | |
| 446 | return yield* updateCachedStatus(cwd, local, remote, { publish: true }); | |
| 447 | 484 | }); |
| 448 | 485 | |
| 486 | const refreshPullRequestStatus: VcsStatusBroadcaster["Service"]["refreshPullRequestStatus"] = | |
| 487 | Effect.fn("VcsStatusBroadcaster.refreshPullRequestStatus")(function* (rawCwd) { | |
| 488 | const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); | |
| 489 | return yield* withRemoteWriteLock( | |
| 490 | cwd, | |
| 491 | Effect.gen(function* () { | |
| 492 | const cached = yield* getCachedStatus(cwd); | |
| 493 | if (cached?.remote?.value == null) return null; | |
| 494 | const poller = (yield* SynchronizedRef.get(pollersRef)).get(cwd); | |
| 495 | const demandCwds = poller ? [...(yield* Ref.get(poller.demandCwds)).keys()] : [rawCwd]; | |
| 496 | const shouldRefresh = (yield* Effect.forEach( | |
| 497 | demandCwds, | |
| 498 | (demandCwd) => | |
| 499 | backgroundPolicy.shouldRunScopeWork({ type: "vcs-status", cwd: demandCwd }), | |
| 500 | { concurrency: "unbounded" }, | |
| 501 | )).some(Boolean); | |
| 502 | if (!shouldRefresh) return null; | |
| 503 | // Resolve the checked-out branch again. A cached PR can belong to | |
| 504 | // the previous branch after an agent checks out another branch. | |
| 505 | const remote = yield* workflow.remoteStatus( | |
| 506 | { cwd }, | |
| 507 | { refreshUpstream: false, refreshMissingPullRequest: true }, | |
| 508 | ); | |
| 509 | return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); | |
| 510 | }), | |
| 511 | ); | |
| 512 | }); | |
| 513 | ||
| 449 | 514 | const makeRemoteRefreshLoop = ( |
| 450 | 515 | cwd: string, |
| 451 | 516 | demandCwdsRef: Ref.Ref<ReadonlyMap<string, number>>, |
| ... | ... | @@ -657,6 +722,7 @@ export const make = Effect.gen(function* () { |
| 657 | 722 | getStatus, |
| 658 | 723 | refreshLocalStatus, |
| 659 | 724 | refreshStatus, |
| 725 | refreshPullRequestStatus, | |
| 660 | 726 | streamStatus, |
| 661 | 727 | }); |
| 662 | 728 | }); |
docs/internals/overview.md+11-4| ... | ... | @@ -99,10 +99,17 @@ once per minute, including when no client is connected. It dispatches the guarde |
| 99 | 99 | `thread.auto-settle` command, which uses the existing settlement event lifecycle. Automatic |
| 100 | 100 | settlement excludes live background work and requires a comparable PR timestamp for immediate PR |
| 101 | 101 | settlement. The command carries the latest activity timestamp and rejects any later event for its |
| 102 | thread after the reactor's snapshot. | |
| 103 | Clients render the persisted settlement state and do not derive settlement from PR or inactivity | |
| 104 | state. A committed `thread.settled` event also lets `ProviderCommandReactor` stop an idle provider | |
| 105 | session. | |
| 102 | thread after the reactor's snapshot. The sweep looks a branch up from the thread's worktree when it | |
| 103 | still exists, so it shares the per-cwd PR cache the sidebar polls instead of spending a second host | |
| 104 | request. Clients render the persisted settlement state and do not derive settlement from PR or | |
| 105 | inactivity state. A committed `thread.settled` event also lets `ProviderCommandReactor` stop an idle | |
| 106 | provider session. | |
| 107 | ||
| 108 | At turn completion, `CheckpointReactor` refreshes PR discovery when the checkout matches the | |
| 109 | thread's non-default branch. `VcsStatusBroadcaster` requires loaded remote status and permission | |
| 110 | from background policy. `GitManager` retries only a successful "no PR" cache entry for the current | |
| 111 | branch, preserving known PRs and failure backoff without fetching remotes. Remote status reads | |
| 112 | that write the broadcaster cache share a lock per cwd, including the initial status load. | |
| 106 | 113 | |
| 107 | 114 | ## Drainable workers |
| 108 | 115 |
docs/user/source-control.md+3| ... | ... | @@ -40,6 +40,9 @@ T3 Code works with the platforms your team already uses: |
| 40 | 40 | **Stay on top of open reviews** |
| 41 | 41 | |
| 42 | 42 | - See if your current branch already has an open PR/MR |
| 43 | - When an agent finishes a turn on your thread's branch, T3 Code checks for a newly opened | |
| 44 | PR/MR if background activity is enabled for that repository. Known reviews keep their normal | |
| 45 | refresh schedule. | |
| 43 | 46 | - Open several reviews from the **Pull requests** page as tabs in the right panel |
| 44 | 47 | - Your authored reviews stay at the top and use the selected sort within their group. By default, |
| 45 | 48 | see passing and approved reviews first, passing reviews awaiting approval next, and conflicting |