diff --git a/apps/server/src/git/GitManager.test.ts b/apps/server/src/git/GitManager.test.ts index 1ce25b4106be2b2e818acae07bb4a7e5e2e2dde2..9f5f368d2cf248d15644794322c6f254507edc5f 100644 --- a/apps/server/src/git/GitManager.test.ts +++ b/apps/server/src/git/GitManager.test.ts @@ -1772,6 +1772,58 @@ it.layer(layerGitManagerTest)("GitManager", (it) => { }), ); + it.effect("branch PR lookup announces a pull request when it reads it merged", () => + Effect.scoped( + Effect.gen(function* () { + const repoDir = yield* makeTempDir("t3code-git-manager-"); + yield* initRepo(repoDir); + const remoteDir = yield* createBareRemote(); + yield* runGit(repoDir, ["remote", "add", "origin", remoteDir]); + for (const branch of ["feature/merges-later", "feature/already-merged"]) { + yield* runGit(repoDir, ["checkout", "-b", branch, "main"]); + yield* runGit(repoDir, ["push", "-u", "origin", branch]); + } + const pullRequest = (number: number, headRefName: string, state: string) => + encodeCliJson([ + { + number, + title: headRefName, + url: `https://github.com/pingdotgg/codething-mvp/pull/${number}`, + baseRefName: "main", + headRefName, + state, + updatedAt: "2026-04-07T15:00:00Z", + }, + ]); + const { manager } = yield* makeManager({ + ghScenario: { + prListSequenceByHeadSelector: { + "feature/merges-later": [ + pullRequest(401, "feature/merges-later", "OPEN"), + pullRequest(401, "feature/merges-later", "MERGED"), + ], + "feature/already-merged": [pullRequest(402, "feature/already-merged", "MERGED")], + }, + }, + }); + const changes = yield* manager.subscribePullRequestStateChanges; + const lookup = (branch: string) => manager.branchPullRequest({ cwd: repoDir, branch }); + + yield* lookup("feature/merges-later"); + yield* lookup("feature/already-merged"); + // Open answers are re-read after a minute. + yield* TestClock.adjust("61 seconds"); + expect((yield* lookup("feature/merges-later"))?.state).toBe("merged"); + + const announced = yield* Stream.runCollect(Stream.take(changes, 2)); + expect(announced).toEqual([ + { host: "github.com", repository: "pingdotgg/codething-mvp", number: 402 }, + { host: "github.com", repository: "pingdotgg/codething-mvp", number: 401 }, + ]); + }), + ), + ); + it.effect("branch PR lookup propagates provider failures", () => Effect.gen(function* () { const repoDir = yield* makeTempDir("t3code-git-manager-"); diff --git a/apps/server/src/git/GitManager.ts b/apps/server/src/git/GitManager.ts index e9b244dfbe86d060b03cff6ce7c255bbfd0f860d..593910ba9a4abd6dd1c11fc7b9762bb03129b1c2 100644 --- a/apps/server/src/git/GitManager.ts +++ b/apps/server/src/git/GitManager.ts @@ -12,8 +12,11 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Order from "effect/Order"; import * as Path from "effect/Path"; +import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; import { GitActionProgressEvent, GitActionProgressPhase, @@ -35,6 +38,7 @@ import { type SourceControlProviderKind, type SourceControlWritingStyleSettings, type ThreadId, + type ThreadPullRequestKey, type VcsCreateWorktreeInput, type VcsCreateWorktreeResult, } from "@t3tools/contracts"; @@ -55,6 +59,11 @@ import { isSshRemoteUrl, type ChangeRequestTerminology, } from "@t3tools/shared/sourceControl"; +import { parseChangeRequestUrl } from "@t3tools/shared/changeRequestUrl"; +import { + normalizeThreadPullRequestKey, + threadPullRequestKeyOf, +} from "@t3tools/shared/threadPullRequests"; import { GitManagerError, GitPullRequestMaterializationError } from "@t3tools/contracts"; import * as TextGeneration from "../textGeneration/TextGeneration.ts"; @@ -139,6 +148,15 @@ export class GitManager extends Context.Service< input: GitRunStackedActionInput, options?: GitRunStackedActionOptions, ) => Effect.Effect; + /** + * Pull requests a branch PR lookup saw in a new state (merged, closed, reopened), so thread + * links can catch up before the next sync sweep. Best effort. + */ + readonly subscribePullRequestStateChanges: Effect.Effect< + Stream.Stream, + never, + Scope.Scope + >; } >()("t3/git/GitManager") {} @@ -1122,6 +1140,26 @@ export const make = Effect.gen(function* () { prLookupFailureStreakByKey.set(key, streak); return prLookupFailureTtl(streak); }; + // The last state a branch PR lookup read per pull request. The thread details panel shows + // this lookup, so a merge it sees must reach the thread's link and settlement right away. + const pullRequestStateChanges = yield* PubSub.sliding(64); + const lookupStates = new Map(); + const noteLookupState = (pr: PullRequestInfo) => + Effect.suspend(() => { + const parsed = parseChangeRequestUrl(pr.url); + if (parsed === null || parsed.number !== pr.number) return Effect.void; + const key = normalizeThreadPullRequestKey(parsed); + const id = threadPullRequestKeyOf(key); + // An unseen pull request counts as open, so only a terminal first read announces. + const previous = lookupStates.get(id) ?? "open"; + lookupStates.delete(id); + if (lookupStates.size >= PR_LOOKUP_CACHE_CAPACITY) { + const oldest = lookupStates.keys().next().value; + if (oldest !== undefined) lookupStates.delete(oldest); + } + lookupStates.set(id, pr.state); + return previous === pr.state ? Effect.void : PubSub.publish(pullRequestStateChanges, key); + }); const prLookupCache = yield* Cache.makeWith( (key: string) => { const [ @@ -1154,6 +1192,7 @@ export const make = Effect.gen(function* () { return { latest: null, headContext }; } const latest = yield* findLatestPrForHeadContext(cwd, headContext); + if (latest !== null) yield* noteLookupState(latest); return { latest, headContext }; }); }, @@ -2896,6 +2935,9 @@ export const make = Effect.gen(function* () { resolvePullRequest, preparePullRequestThread, runStackedAction, + subscribePullRequestStateChanges: PubSub.subscribe(pullRequestStateChanges).pipe( + Effect.map((subscription) => Stream.fromSubscription(subscription)), + ), }); }); diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 55c1866ee9991b9862c80f5ac3901e24815a4006..36f4bdb158913f84d12a8c6f6e5ccc8775dc11c9 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -32,6 +32,7 @@ import * as Ref from "effect/Ref"; import { TestClock } from "effect/testing"; import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts"; +import * as GitManager from "../git/GitManager.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import * as ServerActivation from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; @@ -222,6 +223,9 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: invalidate: options.invalidate ?? (() => Effect.void), subscribeStateChanges: Effect.succeed(Stream.fromQueue(stateChanges)), }), + Layer.mock(GitManager.GitManager)({ + subscribePullRequestStateChanges: Effect.succeed(Stream.empty), + }), Layer.mock(ProjectionStore.ProjectionStoreV2)({ // Mirrors the store's filter: active threads that have at least one link. getThreadsWithPullRequests: (threadId) => diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index d171e4b301c117caf3b8c807e47f3063a77ba4ed..5f1fcd8269acd4c42c8f1b0903ec075ea17ff421 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -28,6 +28,7 @@ import type * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; +import * as GitManager from "../git/GitManager.ts"; import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import { forkParked } from "../serverActivation.ts"; @@ -155,6 +156,7 @@ export const make = Effect.gen(function* () { const engine = yield* Orchestrator.OrchestratorV2; const projections = yield* ProjectionStore.ProjectionStoreV2; const pullRequests = yield* PullRequestService.PullRequestService; + const git = yield* GitManager.GitManager; const crypto = yield* Crypto.Crypto; const lastSyncedAt = new Map(); @@ -424,8 +426,12 @@ export const make = Effect.gen(function* () { "PullRequestSyncReactor.start", )(function* () { const events = engine.streamDomainEvents; - // A client reading a pull request can see it merge or close before the next sweep does. - const stateChanges = yield* pullRequests.subscribeStateChanges; + // A client reading a pull request, or its branch status, can see it merge or close before + // the next sweep does. + const stateChanges = Stream.merge( + yield* pullRequests.subscribeStateChanges, + yield* git.subscribePullRequestStateChanges, + ); yield* forkParked( Stream.runForEach(stateChanges, requestSync).pipe( Effect.catchCause(logSkipped("pull request state change stream failed", {})),