# Files changed · PR \#23422 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/23422/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [fix\(server\): update branch labels after terminal checkouts](https://git.cafe/pingdotgg/t3code/pull/23422.md) Version: 19 Head commit: 42eb1f0890aebbd5d0fb8031c0889e333a0368df Base commit: 3b95e7831b11b17627dff16843bc087bfca397ec Comparison base: 3b95e7831b11b17627dff16843bc087bfca397ec Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/git/GitWorkflowService.test.ts Status: modified; +45; −21 ```diff @@ -12,6 +12,23 @@ import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; +const testLayer = GitWorkflowService.layer.pipe( + Layer.provide( + Layer.mock(VcsDriverRegistry.VcsDriverRegistry)({ + resolve: () => Effect.succeed({ kind: "git" } as VcsDriverRegistry.VcsDriverHandle), + }), + ), + Layer.provide(Layer.mock(GitManager.GitManager)({})), + Layer.provideMerge(GitVcsDriver.layer), + Layer.provideMerge(VcsProcess.layer), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-ref-named-" }).pipe( + Layer.provide(NodeServices.layer), + ), + ), + Layer.provideMerge(NodeServices.layer), +); + it.layer(NodeServices.layer)("GitWorkflowService", (it) => { it.effect("finds a name among the refs of every namespace", () => Effect.gen(function* () { @@ -33,26 +50,33 @@ (refName) => workflow.hasRefNamed({ cwd, refName }), ); assert.deepEqual(found, [true, true, true, true, false, false]); - }).pipe( - Effect.scoped, - Effect.provide( - GitWorkflowService.layer.pipe( - Layer.provide( - Layer.mock(VcsDriverRegistry.VcsDriverRegistry)({ - resolve: () => Effect.succeed({ kind: "git" } as VcsDriverRegistry.VcsDriverHandle), - }), - ), - Layer.provide(Layer.mock(GitManager.GitManager)({})), - Layer.provideMerge(GitVcsDriver.layer), - Layer.provideMerge(VcsProcess.layer), - Layer.provideMerge( - ServerConfig.layerTest(process.cwd(), { prefix: "t3-ref-named-" }).pipe( - Layer.provide(NodeServices.layer), - ), + }).pipe(Effect.scoped, Effect.provide(testLayer)), + ); + + it.effect("reports the checked-out branch by the name status uses", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const cwd = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-checked-out-" }); + const workflow = yield* GitWorkflowService.GitWorkflowService; + const driver = yield* GitVcsDriver.GitVcsDriver; + const git = (args: ReadonlyArray) => + driver.execute({ operation: "test.setupRepo", cwd, args: [...args] }); + yield* git(["init", "-b", "main"]); + yield* git([ + "-c", + "user.email=t@t", + "-c", + "user.name=t", - ), - Layer.provideMerge(NodeServices.layer), - ), - ), - ), + "commit", + "--allow-empty", + "-m", + "init", + ]); + yield* git(["tag", "main"]); + const onBranch = yield* workflow.checkedOutBranch(cwd); + yield* git(["checkout", "--detach"]); + const detached = yield* workflow.checkedOutBranch(cwd); + assert.deepEqual([onBranch, detached], ["main", null]); + }).pipe(Effect.scoped, Effect.provide(testLayer)), ); }); ``` ## apps/server/src/git/GitWorkflowService.ts Status: modified; +29; −0 ```diff @@ -59,6 +59,10 @@ readonly invalidateLocalStatus: (cwd: string) => Effect.Effect; readonly invalidateRemoteStatus: (cwd: string) => Effect.Effect; readonly invalidateStatus: (cwd: string) => Effect.Effect; + /** The checkout's own HEAD file, inside a linked worktree's admin directory when there is one. */ + readonly headPath: (cwd: string) => Effect.Effect; + /** The checked-out branch without reading status; null when detached or not a repository. */ + readonly checkedOutBranch: (cwd: string) => Effect.Effect; readonly pullCurrentBranch: (cwd: string) => Effect.Effect; readonly runStackedAction: ( input: GitRunStackedActionInput, @@ -366,6 +370,31 @@ invalidateLocalStatus: gitManager.invalidateLocalStatus, invalidateRemoteStatus: gitManager.invalidateRemoteStatus, invalidateStatus: gitManager.invalidateStatus, + headPath: (cwd) => + git + .execute({ + operation: "GitWorkflowService.headPath", + cwd, + args: ["rev-parse", "--path-format=absolute", "--git-path", "HEAD"], + allowNonZeroExit: true, + }) + .pipe( + Effect.map((result) => (result.exitCode === 0 ? result.stdout.trim() || null : null)), + Effect.orElseSucceed(() => null), + ), + checkedOutBranch: (cwd) => + git + .execute({ + operation: "GitWorkflowService.checkedOutBranch", + cwd, + // symbolic-ref --short disambiguates against same-named tags (heads/main); + // status porcelain reports the bare branch name, which this must match. + args: ["branch", "--show-current"], + allowNonZeroExit: true, + }) + .pipe( + Effect.map((result) => (result.exitCode === 0 ? result.stdout.trim() || null : null)), + ), pullCurrentBranch: (cwd) => ensureGitCommand("GitWorkflowService.pullCurrentBranch", cwd).pipe( Effect.andThen(git.pullCurrentBranch(cwd)), ``` ## apps/server/src/vcs/VcsStatusBroadcaster.test.ts Status: modified; +423; −11 ```diff @@ -22,6 +22,7 @@ VcsStatusStreamEvent, } from "@t3tools/contracts"; import { GitManagerError } from "@t3tools/contracts"; +import { applyGitStatusStreamEvent } from "@t3tools/shared/git"; import * as VcsStatusBroadcaster from "./VcsStatusBroadcaster.ts"; import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; @@ -79,6 +80,10 @@ backgroundWorkEnabled?: boolean; automaticFetchInterval?: Effect.Effect; beforeRemoteStatus?: Effect.Effect; + headPath?: string; + headPathCalls?: number; + failLocalStatus?: boolean; + onLocalStatusFailure?: Effect.Effect; }) { return VcsStatusBroadcaster.layer.pipe( Layer.provideMerge(NodeServices.layer), @@ -92,20 +97,38 @@ Layer.provide( Layer.mock(GitWorkflowService.GitWorkflowService)({ localStatus: () => + Effect.suspend(() => { - Effect.sync(() => { state.localStatusCalls += 1; + return state.failLocalStatus + ? (state.onLocalStatusFailure ?? Effect.void).pipe( + Effect.andThen( + Effect.fail( + new GitManagerError({ + operation: "VcsStatusBroadcaster.test", - return state.currentLocalStatus; + cwd: "/repo", + detail: "index.lock is held", + }), + ), + ), + ) + : Effect.succeed(state.currentLocalStatus); }), + // Captured before the hook, like a read that resolved the branch and then waits on + // its PR lookup. remoteStatus: (_input, options) => + Effect.suspend(() => { + const remote = state.currentRemoteStatus; + return (state.beforeRemoteStatus ?? Effect.void).pipe( + Effect.andThen( + Effect.sync(() => { - Effect.suspend(() => state.beforeRemoteStatus ?? Effect.void).pipe( - Effect.andThen( - Effect.sync(() => { + state.remoteStatusCalls += 1; - state.remoteStatusCalls += 1; - state.remoteStatusRefreshUpstreamValues?.push(options?.refreshUpstream); - return state.currentRemoteStatus; + state.remoteStatusRefreshUpstreamValues?.push(options?.refreshUpstream); + return remote; + }), + ), + ); + }), + checkedOutBranch: () => Effect.sync(() => state.currentLocalStatus.refName), - }), - ), - ), invalidateLocalStatus: () => Effect.sync(() => { state.localInvalidationCalls += 1; @@ -119,6 +142,11 @@ state.localInvalidationCalls += 1; state.remoteInvalidationCalls += 1; }), + headPath: () => + Effect.sync(() => { + state.headPathCalls = (state.headPathCalls ?? 0) + 1; + return state.headPath ?? null; + }), }), ), ); @@ -503,8 +531,391 @@ const status = yield* Fiber.join(refresh); assert.deepStrictEqual(status.branchChanges, fetchedLocal.branchChanges); }).pipe(Effect.provide(layerTestFor(state))); + }); + + it.effect("a background poll publishes a branch checked out outside T3 Code", () => { + const state = { + currentLocalStatus: { ...baseLocalStatus, isDefaultRef: true, refName: "main" }, + currentRemoteStatus: remoteStatusWithPr as VcsStatusRemoteResult | null, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + // Seeding the cache leaves the 30s poll as the only refresh. + yield* broadcaster.getStatus({ cwd: "/repo" }); + const clientStates: Array = []; + const subscribed = yield* Deferred.make(); + const polled = yield* Deferred.make(); + yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => + Effect.gen(function* () { + clientStates.push(applyGitStatusStreamEvent(clientStates.at(-1) ?? null, event)); + yield* Deferred.succeed(clientStates.length === 1 ? subscribed : polled, undefined); + }), + ).pipe(Effect.forkScoped); + yield* Deferred.await(subscribed); + + // `git switch -c repro` in the terminal: nothing calls a refresh RPC. + state.currentLocalStatus = { ...baseLocalStatus, isDefaultRef: false, refName: "repro" }; + state.currentRemoteStatus = { ...baseRemoteStatus, hasUpstream: false }; + yield* TestClock.adjust(Duration.seconds(30)); + yield* Deferred.await(polled); + + // One event, so the new branch is never shown with the previous branch's PR. + assert.strictEqual(clientStates.length, 2); + assert.strictEqual(clientStates[1]?.refName, "repro"); + assert.strictEqual(clientStates[1]?.pr, null); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }); + + it.effect("a checkout during a poll's remote read cannot pair branches", () => { + const state: Parameters[0] = { + currentLocalStatus: { ...baseLocalStatus, isDefaultRef: true, refName: "main" }, + currentRemoteStatus: { ...remoteStatusWithPr, aheadCount: 1 }, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const clientStates: Array = []; + const subscribed = yield* Deferred.make(); + const checkedOut = yield* Deferred.make(); + yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => + Effect.gen(function* () { + const next = applyGitStatusStreamEvent(clientStates.at(-1) ?? null, event); + clientStates.push(next); + yield* Deferred.succeed(subscribed, undefined); + if (next.refName === "repro") yield* Deferred.succeed(checkedOut, undefined); + }), + ).pipe(Effect.forkScoped); + yield* Deferred.await(subscribed); + + const lookupStarted = yield* Deferred.make(); + const finishLookup = yield* Deferred.make(); + state.beforeRemoteStatus = Deferred.succeed(lookupStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishLookup)), + ); + // The poll reads main's remote and is still looking up its PR. + state.currentRemoteStatus = { ...remoteStatusWithPr, aheadCount: 2 }; + yield* TestClock.adjust(Duration.seconds(30)); + yield* Deferred.await(lookupStarted); + delete state.beforeRemoteStatus; + // `git switch -c repro` in the terminal lands before the poll reads local status. + state.currentLocalStatus = { ...baseLocalStatus, isDefaultRef: false, refName: "repro" }; + state.currentRemoteStatus = { ...baseRemoteStatus, hasUpstream: false }; + yield* Deferred.succeed(finishLookup, undefined); + yield* Deferred.await(checkedOut); + + assert.deepStrictEqual( + clientStates.map((status) => [ + status.refName, + status.pr?.number ?? null, + status.aheadCount, + ]), + [ + ["main", remoteStatusWithPr.pr?.number, 1], + ["repro", null, 0], + ], + ); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }); + + it.effect("a poll whose local read fails publishes nothing and the next poll recovers", () => { + const state: Parameters[0] = { + currentLocalStatus: { ...baseLocalStatus, isDefaultRef: true, refName: "main" }, + currentRemoteStatus: remoteStatusWithPr, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const clientStates: Array = []; + const received = [yield* Deferred.make()]; + yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => + Effect.gen(function* () { + clientStates.push(applyGitStatusStreamEvent(clientStates.at(-1) ?? null, event)); + yield* Deferred.succeed(received[clientStates.length - 1]!, undefined); + }), + ).pipe(Effect.forkScoped); + yield* Deferred.await(received[0]!); + + // An editor commit holds index.lock while main moves ahead. + state.failLocalStatus = true; + state.currentRemoteStatus = { ...remoteStatusWithPr, aheadCount: 1 }; + let failedLocalCalls = state.localStatusCalls; + received.push(yield* Deferred.make()); + yield* TestClock.adjust(Duration.seconds(30)); + assert.strictEqual(state.localStatusCalls, failedLocalCalls + 1); + assert.strictEqual(clientStates.length, 1); + + // The lock clears on the same branch: the counts still differ from the cache, so the + // next poll reads local status again instead of keeping the old totals. + state.failLocalStatus = false; + state.currentLocalStatus = { + ...state.currentLocalStatus, + hasWorkingTreeChanges: true, + workingTree: { files: [], insertions: 3, deletions: 0 }, + }; + yield* TestClock.adjust(Duration.seconds(30)); + yield* Deferred.await(received[1]!); + assert.deepStrictEqual( + [ + clientStates[1]?.refName, + clientStates[1]?.aheadCount, + clientStates[1]?.pr, + clientStates[1]?.hasWorkingTreeChanges, + ], + ["main", 1, remoteStatusWithPr.pr, true], + ); + + // Still locked after `git switch -c repro`: repro's remote cannot go out under main. + state.currentLocalStatus = { ...baseLocalStatus, isDefaultRef: false, refName: "repro" }; + state.currentRemoteStatus = { ...baseRemoteStatus, hasUpstream: false }; + state.failLocalStatus = true; + failedLocalCalls = state.localStatusCalls; + received.push(yield* Deferred.make()); + yield* TestClock.adjust(Duration.seconds(30)); + assert.strictEqual(state.localStatusCalls, failedLocalCalls + 1); + assert.strictEqual(clientStates.length, 2); + + state.failLocalStatus = false; + yield* TestClock.adjust(Duration.seconds(30)); + yield* Deferred.await(received[2]!); + assert.strictEqual(clientStates.length, 3); + assert.deepStrictEqual( + [clientStates[2]?.refName, clientStates[2]?.hasUpstream, clientStates[2]?.pr], + ["repro", false, null], + ); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); }); + it.effect("a terminal checkout publishes the new branch without waiting for a poll", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const gitDir = yield* fs.makeTempDirectoryScoped({ prefix: "vcs-head-watch-" }); + const headPath = path.join(gitDir, "HEAD"); + yield* fs.writeFileString(headPath, "ref: refs/heads/main\n"); + const state = { + currentLocalStatus: { ...baseLocalStatus, isDefaultRef: true, refName: "main" }, + currentRemoteStatus: remoteStatusWithPr as VcsStatusRemoteResult | null, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + remoteStatusRefreshUpstreamValues: [] as Array, + headPath, + }; + + yield* Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const clientStates: Array = []; + const subscribed = yield* Deferred.make(); + const checkedOut = yield* Deferred.make(); + yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => + Effect.gen(function* () { + const next = applyGitStatusStreamEvent(clientStates.at(-1) ?? null, event); + clientStates.push(next); + yield* Deferred.succeed(subscribed, undefined); + if (next.refName === "repro") yield* Deferred.succeed(checkedOut, undefined); + }), + ).pipe(Effect.forkScoped); + yield* Deferred.await(subscribed); + + // What `git switch -c repro` does to HEAD, with the test clock never advanced. + state.currentLocalStatus = { ...baseLocalStatus, isDefaultRef: false, refName: "repro" }; + state.currentRemoteStatus = { ...baseRemoteStatus, hasUpstream: false }; + const lockPath = path.join(gitDir, "HEAD.lock"); + yield* fs.writeFileString(lockPath, "ref: refs/heads/repro\n"); + yield* fs.rename(lockPath, headPath); + yield* Deferred.await(checkedOut); + + assert.deepStrictEqual( + clientStates.filter((status) => status.refName === "repro").map((status) => status.pr), + [null], + ); + // A checkout must not trigger a network fetch. + assert.deepStrictEqual(state.remoteStatusRefreshUpstreamValues.slice(1), [false]); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + + it.effect("a checkout read that fails while index.lock is held retries until it succeeds", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const gitDir = yield* fs.makeTempDirectoryScoped({ prefix: "vcs-head-watch-" }); + const headPath = path.join(gitDir, "HEAD"); + yield* fs.writeFileString(headPath, "ref: refs/heads/main\n"); + const failed = yield* Deferred.make(); + const state = { + currentLocalStatus: { ...baseLocalStatus, isDefaultRef: true, refName: "main" }, + currentRemoteStatus: null, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + headPath, + failLocalStatus: false, + onLocalStatusFailure: Deferred.succeed(failed, undefined).pipe(Effect.asVoid), + }; + + yield* Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const subscribed = yield* Deferred.make(); + const checkedOut = yield* Deferred.make(); + let clientState: VcsStatusResult | null = null; + yield* Stream.runForEach( + broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }), + (event) => + Effect.gen(function* () { + clientState = applyGitStatusStreamEvent(clientState, event); + yield* Deferred.succeed(subscribed, undefined); + if (clientState.refName === "repro") yield* Deferred.succeed(checkedOut, undefined); + }), + ).pipe(Effect.forkScoped); + yield* Deferred.await(subscribed); + + // A terminal commit holds index.lock across the checkout, so the HEAD event's read fails. + state.failLocalStatus = true; + state.currentLocalStatus = { ...baseLocalStatus, isDefaultRef: false, refName: "repro" }; + const lockPath = path.join(gitDir, "HEAD.lock"); + yield* fs.writeFileString(lockPath, "ref: refs/heads/repro\n"); + yield* fs.rename(lockPath, headPath); + yield* Deferred.await(failed); + const failedCalls = state.localStatusCalls; + + // Removing index.lock moves no HEAD, and no remote poll runs for this subscriber. + state.failLocalStatus = false; + yield* TestClock.adjust(Duration.seconds(30)); + yield* Deferred.await(checkedOut); + assert.strictEqual(state.localStatusCalls, failedCalls + 1); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + + it.effect("subscribers share one HEAD watch that ends with the last of them", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const gitDir = yield* fs.makeTempDirectoryScoped({ prefix: "vcs-head-watch-" }); + const headPath = path.join(gitDir, "HEAD"); + yield* fs.writeFileString(headPath, "ref: refs/heads/main\n"); + const state = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: baseRemoteStatus as VcsStatusRemoteResult | null, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + headPath, + headPathCalls: 0, + }; + + yield* Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const subscribe = Effect.gen(function* () { + const received = yield* Deferred.make(); + const fiber = yield* Stream.runForEach( + broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }), + () => Deferred.succeed(received, undefined), + ).pipe(Effect.forkScoped); + yield* Deferred.await(received); + return fiber; + }); + + const first = yield* subscribe; + const second = yield* subscribe; + assert.strictEqual(state.headPathCalls, 1); + + yield* Fiber.interrupt(first); + yield* Fiber.interrupt(second); + yield* subscribe; + assert.strictEqual(state.headPathCalls, 2); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + + it.effect("a HEAD watch that ended restarts for the next subscriber and keeps earlier ones", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const gitDir = yield* fs.makeTempDirectoryScoped({ prefix: "vcs-head-watch-" }); + const headPath = path.join(gitDir, "HEAD"); + yield* fs.writeFileString(headPath, "ref: refs/heads/main\n"); + const state: Parameters[0] = { + currentLocalStatus: { ...baseLocalStatus, isDefaultRef: true, refName: "main" }, + currentRemoteStatus: null, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + headPathCalls: 0, + }; + + yield* Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const subscribe = Effect.gen(function* () { + const subscribed = yield* Deferred.make(); + const refs = new Map>(); + const sawRef = (refName: string) => { + const existing = refs.get(refName); + if (existing) return existing; + const created = Deferred.makeUnsafe(); + refs.set(refName, created); + return created; + }; + let clientState: VcsStatusResult | null = null; + const fiber = yield* Stream.runForEach( + broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }), + (event) => + Effect.gen(function* () { + clientState = applyGitStatusStreamEvent(clientState, event); + yield* Deferred.succeed(subscribed, undefined); + yield* Deferred.succeed(sawRef(clientState.refName ?? ""), undefined); + }), + ).pipe(Effect.forkScoped); + yield* Deferred.await(subscribed); + return { fiber, sawRef }; + }); + const checkout = (refName: string) => + Effect.gen(function* () { + state.currentLocalStatus = { ...baseLocalStatus, isDefaultRef: false, refName }; + const lockPath = path.join(gitDir, "HEAD.lock"); + yield* fs.writeFileString(lockPath, `ref: refs/heads/${refName}\n`); + yield* fs.rename(lockPath, headPath); + }); + + // The repository has no HEAD yet, so the first watch ends at once. + const first = yield* subscribe; + state.headPath = headPath; + const second = yield* subscribe; + assert.strictEqual(state.headPathCalls, 2); + + yield* checkout("one"); + yield* Deferred.await(first.sawRef("one")); + yield* Deferred.await(second.sawRef("one")); + + // The replacement belongs to both subscribers, so the second leaving keeps it. + yield* Fiber.interrupt(second.fiber); + yield* checkout("two"); + yield* Deferred.await(first.sawRef("two")); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + it.effect("releases remote demand while passive observers remain", () => { const state = { currentLocalStatus: baseLocalStatus, @@ -542,6 +953,7 @@ : Effect.void, ), ), + checkedOutBranch: () => Effect.succeed(baseLocalStatus.refName), invalidateLocalStatus: () => Effect.sync(() => { state.localInvalidationCalls += 1; ``` ## apps/server/src/vcs/VcsStatusBroadcaster.ts Status: modified; +155; −14 ```diff @@ -1,11 +1,13 @@ import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; 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 Path from "effect/Path"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schedule from "effect/Schedule"; @@ -35,6 +37,13 @@ const VCS_STATUS_REFRESH_FAILURE_BASE_DELAY = Duration.seconds(30); const VCS_STATUS_REFRESH_FAILURE_MAX_DELAY = Duration.minutes(15); const MAX_FAILURE_DIAGNOSTIC_VALUES = 8; +const CHECKOUT_RACE_ATTEMPTS = 3; +// A held index.lock fails the read, and its removal moves no HEAD to trigger another. +const CHECKOUT_REFRESH_RETRY = Schedule.exponential(Duration.seconds(1)).pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), +); const MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH = 128; function boundedDiagnosticValue(value: string): string { @@ -135,6 +144,12 @@ readonly remote: CachedValue | null; } +interface ActiveHeadWatcher { + readonly fiber: Fiber.Fiber; + readonly ready: Deferred.Deferred; + readonly subscriberCount: number; +} + interface ActiveRemotePoller { readonly fiber: Fiber.Fiber; readonly subscriberCount: number; @@ -234,6 +249,7 @@ const workflow = yield* GitWorkflowService.GitWorkflowService; const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; const changesPubSub = yield* Effect.acquireRelease( PubSub.unbounded(), (pubsub) => PubSub.shutdown(pubsub), @@ -249,6 +265,7 @@ const withRemoteWriteLock = (cwd: string, effect: Effect.Effect) => remoteWriteLocks.withLock(cwd, effect); const pollersRef = yield* SynchronizedRef.make(new Map()); + const headWatchersRef = yield* SynchronizedRef.make(new Map()); const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* ( cwd: string, @@ -484,23 +501,50 @@ if (options?.refreshUpstream !== false) { yield* workflow.invalidateRemoteStatus(cwd); } + // A checkout outside T3 Code can land between these reads, and the remote result does + // not name its branch. Reading the branch around it keeps one branch's PR and counts + // from reaching clients beside another branch's name. + for (let attempt = 1; attempt <= CHECKOUT_RACE_ATTEMPTS; attempt += 1) { + // The first read may hold the previous branch's result in the remote cache. + if (attempt > 1) yield* workflow.invalidateRemoteStatus(cwd); + const branch = yield* workflow.checkedOutBranch(cwd); + const remote = yield* workflow.remoteStatus( + { cwd }, + attempt === 1 ? options : { ...options, refreshUpstream: false }, + ); + if ((yield* workflow.checkedOutBranch(cwd)) !== branch) continue; + const pulled = + options?.refreshUpstream === false + ? null + : yield* maybeAutoPull(cwd, remote, options?.policyCwds ?? [cwd]); + if (pulled !== null) return pulled.remote; + if (remote === null) { + return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); - const previousRemote = (yield* getCachedStatus(cwd))?.remote?.value; + } + const cached = yield* getCachedStatus(cwd); + const previousRemote = cached?.remote?.value; + const branchShown = cached?.local?.value.refName === branch; + const countsMoved = + !previousRemote || - const remote = yield* workflow.remoteStatus({ cwd }, options); - const pulled = - options?.refreshUpstream === false - ? null - : yield* maybeAutoPull(cwd, remote, options?.policyCwds ?? [cwd]); - if (pulled !== null) return pulled.remote; - if ( - remote && - (!previousRemote || previousRemote.aheadCount !== remote.aheadCount || previousRemote.behindCount !== remote.behindCount || + previousRemote.aheadOfDefaultCount !== remote.aheadOfDefaultCount; + if (branchShown && !countsMoved) { + return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); + } - previousRemote.aheadOfDefaultCount !== remote.aheadOfDefaultCount) - ) { - yield* refreshLocalStatusCore(cwd); + yield* workflow.invalidateLocalStatus(cwd); + // A failed read (an editor commit can hold index.lock) publishes nothing, so the + // counts still differ from the cache and the next poll reads local status again. + const local = yield* workflow.localStatus({ cwd }); + if (local.refName !== branch) continue; + if (cached?.local?.fingerprint === fingerprintStatusPart(local)) { + return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); + } + yield* updateCachedStatus(cwd, local, remote, { publish: true }); + return remote; } + // HEAD kept moving. The HEAD watch refreshes again once it settles. - return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); + return (yield* getCachedStatus(cwd))?.remote?.value ?? null; }), ); }); @@ -722,11 +766,108 @@ } }); + // A checkout outside T3 Code, such as `git switch` in the terminal, calls no refresh RPC. + // Rereading remote too, without a fetch, sends the new branch with its own PR and counts. + const refreshCheckedOutRef = Effect.fn("VcsStatusBroadcaster.refreshCheckedOutRef")(function* ( + cwd: string, + ) { + if ((yield* getCachedStatus(cwd))?.remote) { + const remote = yield* refreshRemoteStatus(cwd, { refreshUpstream: false }); + if (remote) return; + } + yield* refreshLocalStatusCore(cwd); + }); + + // Runs for the subscription lifetime, so tracing would leave a span open indefinitely. + const watchHead = Effect.fnUntraced(function* (cwd: string, ready: Deferred.Deferred) { + const headPath = yield* workflow.headPath(cwd); + if (headPath === null) return; + // Git replaces HEAD by renaming HEAD.lock, which a watch on the file itself would lose. + yield* fs + .watch(path.dirname(headPath), { + onReady: () => Deferred.doneUnsafe(ready, Exit.void), + }) + .pipe( + Stream.filter((event) => path.basename(event.path) === "HEAD"), + // A rebase moves HEAD many times; refresh once for whatever arrived during a refresh. + Stream.buffer({ capacity: 1, strategy: "sliding" }), + Stream.runForEach(() => + refreshCheckedOutRef(cwd).pipe( + Effect.retry(CHECKOUT_REFRESH_RETRY), + Effect.catchCause((cause) => + Effect.logWarning("VCS checked-out ref refresh failed", { + cwdLength: cwd.length, + ...remoteRefreshFailureDiagnostics(cause), + }), + ), + ), + ), + ); + }); + + const retainHeadWatcher = Effect.fn("VcsStatusBroadcaster.retainHeadWatcher")(function* ( + cwd: string, + ) { + const ready = yield* SynchronizedRef.modifyEffect(headWatchersRef, (activeWatchers) => { + const existing = activeWatchers.get(cwd); + if (existing && existing.fiber.pollUnsafe() === undefined) { + const nextWatchers = new Map(activeWatchers).set(cwd, { + ...existing, + subscriberCount: existing.subscriberCount + 1, + }); + return Effect.succeed([existing.ready, nextWatchers] as const); + } + // A watch that ended, say before the repository existed, restarts for everyone holding it. + return Effect.gen(function* () { + const ready = yield* Deferred.make(); + const fiber = yield* watchHead(cwd, ready).pipe( + Effect.catchCause((cause) => + Effect.logWarning("VCS HEAD watch failed", { + cwdLength: cwd.length, + ...remoteRefreshFailureDiagnostics(cause), + }), + ), + // Subscribers wait for the watch, so one that cannot start must not hold them. + Effect.ensuring(Deferred.succeed(ready, undefined)), + Effect.forkIn(broadcasterScope), + ); + const nextWatchers = new Map(activeWatchers).set(cwd, { + fiber, + ready, + subscriberCount: (existing?.subscriberCount ?? 0) + 1, + }); + return [ready, nextWatchers] as const; + }); + }); + yield* Deferred.await(ready); + }); + + const releaseHeadWatcher = Effect.fn("VcsStatusBroadcaster.releaseHeadWatcher")(function* ( + cwd: string, + ) { + const watcherToInterrupt = yield* SynchronizedRef.modify(headWatchersRef, (activeWatchers) => { + const existing = activeWatchers.get(cwd); + if (!existing) return [null, activeWatchers] as const; + const nextWatchers = new Map(activeWatchers); + if (existing.subscriberCount > 1) { + nextWatchers.set(cwd, { ...existing, subscriberCount: existing.subscriberCount - 1 }); + return [null, nextWatchers] as const; + } + nextWatchers.delete(cwd); + return [existing.fiber, nextWatchers] as const; + }); + if (watcherToInterrupt) { + yield* Fiber.interrupt(watcherToInterrupt).pipe(Effect.ignore); + } + }); + const streamStatus: VcsStatusBroadcaster["Service"]["streamStatus"] = (input) => Stream.unwrap( Effect.gen(function* () { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); const subscription = yield* PubSub.subscribe(changesPubSub); + // Watch before reading so a checkout in between still publishes. + yield* Effect.acquireRelease(retainHeadWatcher(cwd), () => releaseHeadWatcher(cwd)); const initialLocal = yield* getOrLoadLocalStatus(cwd); const cachedStatus = yield* getCachedStatus(cwd); const initialRemote = cachedStatus?.remote?.value ?? null; ``` ## scripts/fixtures/issue-7409/data.json Status: added; +28; −0 ```diff @@ -1,0 +1,28 @@ +{ + "timestamp": "2026-01-01T12:00:00Z", + "author": "Synthetic Evidence", + "email": "fixture@example.invalid", + "featureBranch": "repro/7409-long-feature-name-for-responsive-branch-label", + "files": { + "README.md": "# Synthetic checkout fixture\n\nThis repository contains only invented evidence data.\n", + "notes.txt": "A stable file for the dirty-worktree control.\n" + }, + "featureFile": "Feature branch content for changed-file totals.\n", + "projects": [ + { + "id": "74090000-0000-4000-8000-000000000001", + "title": "Checkout fixture", + "directory": "regular" + }, + { + "id": "74090000-0000-4000-8000-000000000002", + "title": "Linked checkout fixture", + "directory": "linked" + }, + { + "id": "74090000-0000-4000-8000-000000000003", + "title": "Non-repository control", + "directory": "nonrepo" + } + ] +} ``` ## scripts/fixtures/issue-7409/seed.py Status: added; +100; −0 ```diff @@ -1,0 +1,100 @@ +#!/usr/bin/env python3 +"""Create synthetic Git repositories and optional clean T3 projection fixtures. + +Run with a NEW --root. For --state, stop the isolated server first. The schema +is read from --schema-from; only migration bookkeeping is copied. No user data +or credentials are copied. The original database must never be the destination. +Example: python3 seed.py --root /tmp/checkout-proof --schema-from /tmp/schema.sqlite \ + --state /tmp/isolated/userdata/statev2.sqlite +Reset captures with git -C /regular switch main (and similarly linked). +""" +import argparse +import json +import os +from pathlib import Path +import sqlite3 +import subprocess + +parser = argparse.ArgumentParser(description=__doc__) +parser.add_argument('--root', type=Path, required=True) +parser.add_argument('--schema-from', type=Path) +parser.add_argument('--state', type=Path) +args = parser.parse_args() +data = json.loads(Path(__file__).with_name('data.json').read_text()) +root = args.root.resolve() +root.mkdir(parents=True, exist_ok=False) +env = dict(os.environ, GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL='/dev/null', + GIT_AUTHOR_NAME=data['author'], GIT_COMMITTER_NAME=data['author'], + GIT_AUTHOR_EMAIL=data['email'], GIT_COMMITTER_EMAIL=data['email'], + GIT_AUTHOR_DATE=data['timestamp'], GIT_COMMITTER_DATE=data['timestamp']) +for key in ('GIT_DIR', 'GIT_WORK_TREE', 'GIT_COMMON_DIR', 'GIT_INDEX_FILE'): + env.pop(key, None) + +def git(directory, *arguments): + return subprocess.check_output(['git', '-C', str(directory), *arguments], env=env, stderr=subprocess.PIPE, text=True).strip() + +def initialize(name): + directory = root / name + directory.mkdir() + git(directory, 'init', '--initial-branch=main') + for filename, contents in data['files'].items(): + (directory / filename).write_text(contents) + git(directory, 'add', '.') + git(directory, 'commit', '-m', 'Synthetic initial state') + return directory + +regular = initialize('regular') +git(regular, 'branch', data['featureBranch']) +remote_root = initialize('remote-root') +git(remote_root, 'branch', 'untracked') +git(remote_root, 'switch', '-c', 'tracked') +(remote_root / 'feature.txt').write_text(data['featureFile']) +git(remote_root, 'add', '.') +git(remote_root, 'commit', '-m', 'Synthetic tracked feature') +git(remote_root, 'switch', 'main') +git(root, 'clone', '--bare', str(remote_root), str(root / 'origin.git')) +git(remote_root, 'remote', 'add', 'origin', str(root / 'origin.git')) +git(remote_root, 'fetch', 'origin') +git(remote_root, 'branch', '--set-upstream-to=origin/tracked', 'tracked') +git(remote_root, 'worktree', 'add', str(root / 'linked'), 'untracked') +(root / 'nonrepo').mkdir() +manifest = {'root': str(root), 'data': data, 'commits': { + 'main': git(regular, 'rev-parse', 'main'), + 'tracked': git(remote_root, 'rev-parse', 'tracked')}, + 'headPaths': {name: git(root / name, 'rev-parse', '--path-format=absolute', '--git-path', 'HEAD') + for name in ['regular', 'linked']}} +if args.state: + if not args.schema_from or args.state.resolve() == args.schema_from.resolve(): + parser.error('--state requires a distinct --schema-from') + if args.state.exists(): + parser.error('--state must not already exist; move the stopped isolated database aside first') + source = sqlite3.connect(f'file:{args.schema_from.resolve()}?mode=ro', uri=True) + args.state.parent.mkdir(parents=True, exist_ok=True) + destination = sqlite3.connect(args.state) + for sql, in source.execute("SELECT sql FROM sqlite_master WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%' ORDER BY CASE type WHEN 'table' THEN 0 WHEN 'index' THEN 1 ELSE 2 END"): + destination.execute(sql) + migrations = source.execute('SELECT * FROM effect_sql_migrations').fetchall() + for row in migrations: + destination.execute('INSERT INTO effect_sql_migrations VALUES (' + ','.join('?' for _ in row) + ')', row) + timestamp = data['timestamp'] + for index, project in enumerate(data['projects']): + destination.execute('INSERT INTO projection_projects (project_id,title,workspace_root,scripts_json,created_at,updated_at) VALUES (?,?,?,?,?,?)', + (project['id'], project['title'], str(root / project['directory']), '[]', timestamp, timestamp)) + thread_id = f'74090000-0000-4000-8000-00000000001{index}' + payload = dict(createdBy='user', creationSource='web', id=thread_id, projectId=project['id'], + title='Stopped synthetic thread', providerInstanceId='codex', + modelSelection=dict(instanceId='codex', model='gpt-5.4'), runtimeMode='full-access', + interactionMode='default', branch='main' if index == 0 else 'untracked' if index == 1 else None, + worktreePath=None, activeProviderThreadId=None, linkOrigin=None, + lineage=dict(parentThreadId=None, relationshipToParent=None, rootThreadId=thread_id), + forkedFrom=None, createdAt=timestamp, updatedAt=timestamp, archivedAt=None, + settledOverride=None, settledAt=None, unsettledAt=None, snoozedUntil=None, + snoozedAt=None, pinnedAt=None, pinOrderKey=None, activeOrderKey=None, + lastVisitedAt=None, deletedAt=None) + destination.execute('INSERT INTO orchestration_v2_projection_threads (thread_id,project_id,title,default_provider,runtime_mode,interaction_mode,created_at,updated_at,payload_json,provider_instance_id) VALUES (?,?,?,?,?,?,?,?,?,?)', + (thread_id, project['id'], payload['title'], 'codex', 'full-access', 'default', timestamp, timestamp, json.dumps(payload), 'codex')) + destination.commit() + destination.close() + source.close() +(root / 'manifest.json').write_text(json.dumps(manifest, indent=2) + '\n') +print(json.dumps(manifest, indent=2)) ```