| author |  Theo Browne <me@t3.gg> 2026-10-08 02:38:54-07:00 |
| committer |  GitHub <noreply@github.com> 2026-10-08 02:38:54-07:00 |
| log | 0ce6b9deb0c3f6b4dd9df7ed809f7acb9d237ae0 |
| tree | e4f5280c841b7f5adc92d54c8ef777c335ed16c9 |
| parent | 9381533771ef50d58304f79fac2b74ebd8932be7 |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(server): threads settle as soon as branch status sees their PR merge (#17148)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>4 files changed, 106 insertions(+), 2 deletions(-)
apps
server
src
orchestration-v2
apps/server/src/git/GitManager.test.ts+52
| ... | @@ -1772,6 +1772,58 @@ it.layer(layerGitManagerTest)("GitManager", (it) => { | ... | @@ -1772,6 +1772,58 @@ it.layer(layerGitManagerTest)("GitManager", (it) => { |
| 1772 | }), | 1772 | }), |
| 1773 | ); | 1773 | ); |
| 1774 | 1774 | ||
| 1775 | it.effect("branch PR lookup announces a pull request when it reads it merged", () => | ||
| 1776 | Effect.scoped( | ||
| 1777 | Effect.gen(function* () { | ||
| 1778 | const repoDir = yield* makeTempDir("t3code-git-manager-"); | ||
| 1779 | yield* initRepo(repoDir); | ||
| 1780 | const remoteDir = yield* createBareRemote(); | ||
| 1781 | yield* runGit(repoDir, ["remote", "add", "origin", remoteDir]); | ||
| 1782 | for (const branch of ["feature/merges-later", "feature/already-merged"]) { | ||
| 1783 | yield* runGit(repoDir, ["checkout", "-b", branch, "main"]); | ||
| 1784 | yield* runGit(repoDir, ["push", "-u", "origin", branch]); | ||
| 1785 | } | ||
| 1786 | const pullRequest = (number: number, headRefName: string, state: string) => | ||
| 1787 | encodeCliJson([ | ||
| 1788 | { | ||
| 1789 | number, | ||
| 1790 | title: headRefName, | ||
| 1791 | url: `https://github.com/pingdotgg/codething-mvp/pull/${number}`, | ||
| 1792 | baseRefName: "main", | ||
| 1793 | headRefName, | ||
| 1794 | state, | ||
| 1795 | updatedAt: "2026-04-07T15:00:00Z", | ||
| 1796 | }, | ||
| 1797 | ]); | ||
| 1798 | const { manager } = yield* makeManager({ | ||
| 1799 | ghScenario: { | ||
| 1800 | prListSequenceByHeadSelector: { | ||
| 1801 | "feature/merges-later": [ | ||
| 1802 | pullRequest(401, "feature/merges-later", "OPEN"), | ||
| 1803 | pullRequest(401, "feature/merges-later", "MERGED"), | ||
| 1804 | ], | ||
| 1805 | "feature/already-merged": [pullRequest(402, "feature/already-merged", "MERGED")], | ||
| 1806 | }, | ||
| 1807 | }, | ||
| 1808 | }); | ||
| 1809 | const changes = yield* manager.subscribePullRequestStateChanges; | ||
| 1810 | const lookup = (branch: string) => manager.branchPullRequest({ cwd: repoDir, branch }); | ||
| 1811 | |||
| 1812 | yield* lookup("feature/merges-later"); | ||
| 1813 | yield* lookup("feature/already-merged"); | ||
| 1814 | // Open answers are re-read after a minute. | ||
| 1815 | yield* TestClock.adjust("61 seconds"); | ||
| 1816 | expect((yield* lookup("feature/merges-later"))?.state).toBe("merged"); | ||
| 1817 | |||
| 1818 | const announced = yield* Stream.runCollect(Stream.take(changes, 2)); | ||
| 1819 | expect(announced).toEqual([ | ||
| 1820 | { host: "github.com", repository: "pingdotgg/codething-mvp", number: 402 }, | ||
| 1821 | { host: "github.com", repository: "pingdotgg/codething-mvp", number: 401 }, | ||
| 1822 | ]); | ||
| 1823 | }), | ||
| 1824 | ), | ||
| 1825 | ); | ||
| 1826 | |||
| 1775 | it.effect("branch PR lookup propagates provider failures", () => | 1827 | it.effect("branch PR lookup propagates provider failures", () => |
| 1776 | Effect.gen(function* () { | 1828 | Effect.gen(function* () { |
| 1777 | const repoDir = yield* makeTempDir("t3code-git-manager-"); | 1829 | const repoDir = yield* makeTempDir("t3code-git-manager-"); |
apps/server/src/git/GitManager.ts+42
| ... | @@ -12,8 +12,11 @@ import * as Layer from "effect/Layer"; | ... | @@ -12,8 +12,11 @@ import * as Layer from "effect/Layer"; |
| 12 | import * as Option from "effect/Option"; | 12 | import * as Option from "effect/Option"; |
| 13 | import * as Order from "effect/Order"; | 13 | import * as Order from "effect/Order"; |
| 14 | import * as Path from "effect/Path"; | 14 | import * as Path from "effect/Path"; |
| 15 | import * as PubSub from "effect/PubSub"; | ||
| 15 | import * as Ref from "effect/Ref"; | 16 | import * as Ref from "effect/Ref"; |
| 16 | import * as Schema from "effect/Schema"; | 17 | import * as Schema from "effect/Schema"; |
| 18 | import * as Scope from "effect/Scope"; | ||
| 19 | import * as Stream from "effect/Stream"; | ||
| 17 | import { | 20 | import { |
| 18 | GitActionProgressEvent, | 21 | GitActionProgressEvent, |
| 19 | GitActionProgressPhase, | 22 | GitActionProgressPhase, |
| ... | @@ -35,6 +38,7 @@ import { | ... | @@ -35,6 +38,7 @@ import { |
| 35 | type SourceControlProviderKind, | 38 | type SourceControlProviderKind, |
| 36 | type SourceControlWritingStyleSettings, | 39 | type SourceControlWritingStyleSettings, |
| 37 | type ThreadId, | 40 | type ThreadId, |
| 41 | type ThreadPullRequestKey, | ||
| 38 | type VcsCreateWorktreeInput, | 42 | type VcsCreateWorktreeInput, |
| 39 | type VcsCreateWorktreeResult, | 43 | type VcsCreateWorktreeResult, |
| 40 | } from "@t3tools/contracts"; | 44 | } from "@t3tools/contracts"; |
| ... | @@ -55,6 +59,11 @@ import { | ... | @@ -55,6 +59,11 @@ import { |
| 55 | isSshRemoteUrl, | 59 | isSshRemoteUrl, |
| 56 | type ChangeRequestTerminology, | 60 | type ChangeRequestTerminology, |
| 57 | } from "@t3tools/shared/sourceControl"; | 61 | } from "@t3tools/shared/sourceControl"; |
| 62 | import { parseChangeRequestUrl } from "@t3tools/shared/changeRequestUrl"; | ||
| 63 | import { | ||
| 64 | normalizeThreadPullRequestKey, | ||
| 65 | threadPullRequestKeyOf, | ||
| 66 | } from "@t3tools/shared/threadPullRequests"; | ||
| 58 | 67 | ||
| 59 | import { GitManagerError, GitPullRequestMaterializationError } from "@t3tools/contracts"; | 68 | import { GitManagerError, GitPullRequestMaterializationError } from "@t3tools/contracts"; |
| 60 | import * as TextGeneration from "../textGeneration/TextGeneration.ts"; | 69 | import * as TextGeneration from "../textGeneration/TextGeneration.ts"; |
| ... | @@ -139,6 +148,15 @@ export class GitManager extends Context.Service< | ... | @@ -139,6 +148,15 @@ export class GitManager extends Context.Service< |
| 139 | input: GitRunStackedActionInput, | 148 | input: GitRunStackedActionInput, |
| 140 | options?: GitRunStackedActionOptions, | 149 | options?: GitRunStackedActionOptions, |
| 141 | ) => Effect.Effect<GitRunStackedActionResult, GitManagerServiceError>; | 150 | ) => Effect.Effect<GitRunStackedActionResult, GitManagerServiceError>; |
| 151 | /** | ||
| 152 | * Pull requests a branch PR lookup saw in a new state (merged, closed, reopened), so thread | ||
| 153 | * links can catch up before the next sync sweep. Best effort. | ||
| 154 | */ | ||
| 155 | readonly subscribePullRequestStateChanges: Effect.Effect< | ||
| 156 | Stream.Stream<ThreadPullRequestKey>, | ||
| 157 | never, | ||
| 158 | Scope.Scope | ||
| 159 | >; | ||
| 142 | } | 160 | } |
| 143 | >()("t3/git/GitManager") {} | 161 | >()("t3/git/GitManager") {} |
| 144 | 162 | ||
| ... | @@ -1122,6 +1140,26 @@ export const make = Effect.gen(function* () { | ... | @@ -1122,6 +1140,26 @@ export const make = Effect.gen(function* () { |
| 1122 | prLookupFailureStreakByKey.set(key, streak); | 1140 | prLookupFailureStreakByKey.set(key, streak); |
| 1123 | return prLookupFailureTtl(streak); | 1141 | return prLookupFailureTtl(streak); |
| 1124 | }; | 1142 | }; |
| 1143 | // The last state a branch PR lookup read per pull request. The thread details panel shows | ||
| 1144 | // this lookup, so a merge it sees must reach the thread's link and settlement right away. | ||
| 1145 | const pullRequestStateChanges = yield* PubSub.sliding<ThreadPullRequestKey>(64); | ||
| 1146 | const lookupStates = new Map<string, PullRequestInfo["state"]>(); | ||
| 1147 | const noteLookupState = (pr: PullRequestInfo) => | ||
| 1148 | Effect.suspend(() => { | ||
| 1149 | const parsed = parseChangeRequestUrl(pr.url); | ||
| 1150 | if (parsed === null || parsed.number !== pr.number) return Effect.void; | ||
| 1151 | const key = normalizeThreadPullRequestKey(parsed); | ||
| 1152 | const id = threadPullRequestKeyOf(key); | ||
| 1153 | // An unseen pull request counts as open, so only a terminal first read announces. | ||
| 1154 | const previous = lookupStates.get(id) ?? "open"; | ||
| 1155 | lookupStates.delete(id); | ||
| 1156 | if (lookupStates.size >= PR_LOOKUP_CACHE_CAPACITY) { | ||
| 1157 | const oldest = lookupStates.keys().next().value; | ||
| 1158 | if (oldest !== undefined) lookupStates.delete(oldest); | ||
| 1159 | } | ||
| 1160 | lookupStates.set(id, pr.state); | ||
| 1161 | return previous === pr.state ? Effect.void : PubSub.publish(pullRequestStateChanges, key); | ||
| 1162 | }); | ||
| 1125 | const prLookupCache = yield* Cache.makeWith( | 1163 | const prLookupCache = yield* Cache.makeWith( |
| 1126 | (key: string) => { | 1164 | (key: string) => { |
| 1127 | const [ | 1165 | const [ |
| ... | @@ -1154,6 +1192,7 @@ export const make = Effect.gen(function* () { | ... | @@ -1154,6 +1192,7 @@ export const make = Effect.gen(function* () { |
| 1154 | return { latest: null, headContext }; | 1192 | return { latest: null, headContext }; |
| 1155 | } | 1193 | } |
| 1156 | const latest = yield* findLatestPrForHeadContext(cwd, headContext); | 1194 | const latest = yield* findLatestPrForHeadContext(cwd, headContext); |
| 1195 | if (latest !== null) yield* noteLookupState(latest); | ||
| 1157 | return { latest, headContext }; | 1196 | return { latest, headContext }; |
| 1158 | }); | 1197 | }); |
| 1159 | }, | 1198 | }, |
| ... | @@ -2896,6 +2935,9 @@ export const make = Effect.gen(function* () { | ... | @@ -2896,6 +2935,9 @@ export const make = Effect.gen(function* () { |
| 2896 | resolvePullRequest, | 2935 | resolvePullRequest, |
| 2897 | preparePullRequestThread, | 2936 | preparePullRequestThread, |
| 2898 | runStackedAction, | 2937 | runStackedAction, |
| 2938 | subscribePullRequestStateChanges: PubSub.subscribe(pullRequestStateChanges).pipe( | ||
| 2939 | Effect.map((subscription) => Stream.fromSubscription(subscription)), | ||
| 2940 | ), | ||
| 2899 | }); | 2941 | }); |
| 2900 | }); | 2942 | }); |
| 2901 | 2943 |
apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts+4
| ... | @@ -32,6 +32,7 @@ import * as Ref from "effect/Ref"; | ... | @@ -32,6 +32,7 @@ import * as Ref from "effect/Ref"; |
| 32 | import { TestClock } from "effect/testing"; | 32 | import { TestClock } from "effect/testing"; |
| 33 | 33 | ||
| 34 | import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts"; | 34 | import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts"; |
| 35 | import * as GitManager from "../git/GitManager.ts"; | ||
| 35 | import * as PullRequestService from "../pullRequest/PullRequestService.ts"; | 36 | import * as PullRequestService from "../pullRequest/PullRequestService.ts"; |
| 36 | import * as ServerActivation from "../serverActivation.ts"; | 37 | import * as ServerActivation from "../serverActivation.ts"; |
| 37 | import * as Orchestrator from "./Orchestrator.ts"; | 38 | import * as Orchestrator from "./Orchestrator.ts"; |
| ... | @@ -222,6 +223,9 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: | ... | @@ -222,6 +223,9 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: |
| 222 | invalidate: options.invalidate ?? (() => Effect.void), | 223 | invalidate: options.invalidate ?? (() => Effect.void), |
| 223 | subscribeStateChanges: Effect.succeed(Stream.fromQueue(stateChanges)), | 224 | subscribeStateChanges: Effect.succeed(Stream.fromQueue(stateChanges)), |
| 224 | }), | 225 | }), |
| 226 | Layer.mock(GitManager.GitManager)({ | ||
| 227 | subscribePullRequestStateChanges: Effect.succeed(Stream.empty), | ||
| 228 | }), | ||
| 225 | Layer.mock(ProjectionStore.ProjectionStoreV2)({ | 229 | Layer.mock(ProjectionStore.ProjectionStoreV2)({ |
| 226 | // Mirrors the store's filter: active threads that have at least one link. | 230 | // Mirrors the store's filter: active threads that have at least one link. |
| 227 | getThreadsWithPullRequests: (threadId) => | 231 | getThreadsWithPullRequests: (threadId) => |
apps/server/src/orchestration-v2/PullRequestSyncReactor.ts+8-2
| ... | @@ -28,6 +28,7 @@ import type * as Scope from "effect/Scope"; | ... | @@ -28,6 +28,7 @@ import type * as Scope from "effect/Scope"; |
| 28 | import * as Semaphore from "effect/Semaphore"; | 28 | import * as Semaphore from "effect/Semaphore"; |
| 29 | import * as Stream from "effect/Stream"; | 29 | import * as Stream from "effect/Stream"; |
| 30 | 30 | ||
| 31 | import * as GitManager from "../git/GitManager.ts"; | ||
| 31 | import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts"; | 32 | import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts"; |
| 32 | import * as PullRequestService from "../pullRequest/PullRequestService.ts"; | 33 | import * as PullRequestService from "../pullRequest/PullRequestService.ts"; |
| 33 | import { forkParked } from "../serverActivation.ts"; | 34 | import { forkParked } from "../serverActivation.ts"; |
| ... | @@ -155,6 +156,7 @@ export const make = Effect.gen(function* () { | ... | @@ -155,6 +156,7 @@ export const make = Effect.gen(function* () { |
| 155 | const engine = yield* Orchestrator.OrchestratorV2; | 156 | const engine = yield* Orchestrator.OrchestratorV2; |
| 156 | const projections = yield* ProjectionStore.ProjectionStoreV2; | 157 | const projections = yield* ProjectionStore.ProjectionStoreV2; |
| 157 | const pullRequests = yield* PullRequestService.PullRequestService; | 158 | const pullRequests = yield* PullRequestService.PullRequestService; |
| 159 | const git = yield* GitManager.GitManager; | ||
| 158 | const crypto = yield* Crypto.Crypto; | 160 | const crypto = yield* Crypto.Crypto; |
| 159 | 161 | ||
| 160 | const lastSyncedAt = new Map<string, number>(); | 162 | const lastSyncedAt = new Map<string, number>(); |
| ... | @@ -424,8 +426,12 @@ export const make = Effect.gen(function* () { | ... | @@ -424,8 +426,12 @@ export const make = Effect.gen(function* () { |
| 424 | "PullRequestSyncReactor.start", | 426 | "PullRequestSyncReactor.start", |
| 425 | )(function* () { | 427 | )(function* () { |
| 426 | const events = engine.streamDomainEvents; | 428 | const events = engine.streamDomainEvents; |
| 427 | // A client reading a pull request can see it merge or close before the next sweep does. | 429 | // A client reading a pull request, or its branch status, can see it merge or close before |
| 428 | const stateChanges = yield* pullRequests.subscribeStateChanges; | 430 | // the next sweep does. |
| 431 | const stateChanges = Stream.merge( | ||
| 432 | yield* pullRequests.subscribeStateChanges, | ||
| 433 | yield* git.subscribePullRequestStateChanges, | ||
| 434 | ); | ||
| 429 | yield* forkParked( | 435 | yield* forkParked( |
| 430 | Stream.runForEach(stateChanges, requestSync).pipe( | 436 | Stream.runForEach(stateChanges, requestSync).pipe( |
| 431 | Effect.catchCause(logSkipped("pull request state change stream failed", {})), | 437 | Effect.catchCause(logSkipped("pull request state change stream failed", {})), |