From c07c8b73c8b4a52ecb7ff4145f21a739ebb2c7c3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 6 Nov 2026 14:11:16 +0000 Subject: [PATCH 1/3] test(server): reproduce #32029, switchRef answers before status shows the new branch Fails on main: the vcsSwitchRef handler forks the status refresh detached, so the RPC resolves while status subscribers still see the previous branch. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/ws.test.ts | 108 +++++++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 2be9e61c7c916..60d9db37a05b4 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -24,6 +24,7 @@ import { type ServerConfig, type ServerConfigStreamEvent, type ServerProvider, + type VcsStatusLocalResult, } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; @@ -34,6 +35,7 @@ import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { Headers } from "effect/http"; @@ -1434,6 +1436,112 @@ it.effect("a publish whose push fails still refreshes the new remote", () => ), ); +it.effect("a branch switch answers only after status subscribers can see the new branch", () => + Effect.gen(function* () { + let checkedOut = "main"; + const published = yield* Ref.make("main"); + const refreshStarted = yield* Deferred.make(); + const statusAtResponse = yield* Queue.unbounded(); + const local: VcsStatusLocalResult = { + isRepo: true, + hasPrimaryRemote: false, + isDefaultRef: false, + refName: "main", + hasWorkingTreeChanges: false, + workingTree: { files: [], insertions: 0, deletions: 0 }, + }; + // A large repository's git status takes seconds after the checkout itself has finished. + const readAndPublishStatus = Effect.gen(function* () { + const refName = checkedOut; + yield* Deferred.succeed(refreshStarted, undefined); + yield* Effect.sleep("2 seconds"); + yield* Ref.set(published, refName); + return { ...local, refName }; + }); + const services = Layer.mergeAll( + layerUnusedConfigRpcServices, + Layer.mock(GitWorkflowService.GitWorkflowService)({ + switchRef: (input) => + Effect.sync(() => { + checkedOut = input.refName; + return { refName: input.refName }; + }), + }), + Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({ + refreshLocalStatus: () => readAndPublishStatus, + refreshStatus: () => + readAndPublishStatus.pipe( + Effect.map((status) => ({ + ...status, + hasUpstream: false, + aheadCount: 0, + behindCount: 0, + pr: null, + })), + ), + }), + Layer.mock(ExternalLauncher.ExternalLauncher)({}), + Layer.mock(ProviderRegistry.ProviderRegistry)({}), + Layer.mock(UsageLimitSources.UsageLimitSources)({}), + Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({}), + Layer.mock(DirectEndpoints.DirectEndpoints)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/unused" }), + Layer.mock(ScratchThreadPromotion.ScratchThreadPromotion)({}), + Layer.mock(ServerEnvironment.ServerEnvironment)({}), + Layer.mock(EnvironmentAuth.EnvironmentAuth)({}), + ); + const server = yield* RpcServer.makeNoSerialization(WsRpcGroup, { + ...WS_RPC_SERVER_OPTIONS, + // Runs as the response is produced, before any later fiber can publish status. + onFromServer: (response) => + response._tag === "Exit" + ? Ref.get(published).pipe(Effect.flatMap((ref) => Queue.offer(statusAtResponse, ref))) + : Effect.void, + }).pipe( + Effect.provide( + layerWsRpc( + { + sessionId: AuthSessionId.make("switch-ref-test"), + subject: "switch-ref-test", + method: "browser-session-cookie", + scopes: [AuthSourceControlWriteScope], + }, + {}, + yield* ServerBrowser.ServerBrowser, + () => Effect.void, + { secretReferences: true }, + ).pipe( + Layer.provide(services), + Layer.merge(RpcAuthorization.layer([AuthSourceControlWriteScope])), + ), + ), + ); + yield* server.write(0, { + _tag: "Request", + id: RpcMessage.RequestId("switch"), + tag: WS_METHODS.vcsSwitchRef, + payload: { cwd: "/repo", refName: "feature" }, + headers: Headers.empty, + }); + yield* Deferred.await(refreshStarted); + yield* TestClock.adjust("2 seconds"); + assert.equal( + yield* Queue.take(statusAtResponse), + "feature", + "vcsSwitchRef resolved while status subscribers still saw the previous branch", + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + Layer.fresh(layerConfigSources), + Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), + Layer.mock(ServerBrowser.ServerBrowser)({}), + ), + ), + Effect.scoped, + ), +); + // Pruned history cannot be replayed: a cursor below the boundary must resync from a snapshot. it.effect.each(["thread", "shell"] as const)( "%s resume from a cursor below the pruned boundary loads a snapshot", From 2a06f09424d97b61cec3f7990d1e6f4661c77aa8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 6 Nov 2026 14:17:35 +0000 Subject: [PATCH 2/3] fix(server): a branch switch answers once the new branch's status is published The vcsSwitchRef handler forked the whole status refresh detached, so the RPC answered before status subscribers saw the new branch. The web client then dropped its optimistic label and showed the stale branch for seconds. The handler now awaits the broadcaster's local refresh, which publishes the new branch, before answering. The remote and PR refresh stays detached, and skips its own up-front local read, so a checkout runs git status no more often than before. A failed local refresh is logged and does not fail the checkout. Fixes pingdotgg/t3code#32029 Co-Authored-By: Claude Opus 5.5 --- .../src/vcs/VcsStatusBroadcaster.test.ts | 25 +++++++++++++++++++ apps/server/src/vcs/VcsStatusBroadcaster.ts | 11 +++++--- apps/server/src/ws.ts | 18 ++++++++++--- 3 files changed, 48 insertions(+), 6 deletions(-) diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 60f39b592d6de..22bb26b6a8ddd 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -207,6 +207,31 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); }); + it.effect( + "a refresh after an awaited local refresh reads git status only after the fetch", + () => { + const state: Parameters[0] = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + fileSystem: FileSystem.makeNoop({ realPath: (cwd) => Effect.succeed(cwd) }), + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.refreshLocalStatus("/repo"); + assert.strictEqual(state.localStatusCalls, 1); + + yield* broadcaster.refreshStatus("/repo", { skipLocalRefresh: true }); + assert.strictEqual(state.remoteStatusCalls, 1); + assert.strictEqual(state.localStatusCalls, 2); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }, + ); + it.effect("publishes terminal branch switches and validates cached status reads", () => { const state: Parameters[0] = { currentLocalStatus: baseLocalStatus, diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 3921077ebbb42..9c0a67f46dd1a 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -217,7 +217,11 @@ export class VcsStatusBroadcaster extends Context.Service< ) => Effect.Effect; readonly refreshStatus: ( cwd: string, - options?: { readonly localOnly?: boolean }, + options?: { + readonly localOnly?: boolean; + /** For a caller that just ran `refreshLocalStatus`; local is still re-read after the fetch. */ + readonly skipLocalRefresh?: boolean; + }, ) => Effect.Effect; readonly refreshPullRequestStatus: ( cwd: string, @@ -511,9 +515,9 @@ export const make = Effect.gen(function* () { "VcsStatusBroadcaster.refreshStatus", )(function* (rawCwd, options) { const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); - const previous = options?.localOnly ? yield* getCachedStatus(cwd) : null; - const local = yield* refreshLocalStatusCore(cwd); if (options?.localOnly) { + const previous = yield* getCachedStatus(cwd); + const local = yield* refreshLocalStatusCore(cwd); return yield* withRemoteWriteLock( cwd, Effect.gen(function* () { @@ -530,6 +534,7 @@ export const make = Effect.gen(function* () { }), ); } + if (!options?.skipLocalRefresh) yield* refreshLocalStatusCore(cwd); return yield* withRemoteWriteLock( cwd, Effect.gen(function* () { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 035f8b1688d44..d3c82ed646774 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1982,11 +1982,21 @@ export const layerWsRpc = ( }; }); - const refreshGitStatus = (cwd: string) => + const refreshGitStatus = (cwd: string, options?: { readonly skipLocalRefresh?: boolean }) => vcsStatusBroadcaster - .refreshStatus(cwd) + .refreshStatus(cwd, options) .pipe(Effect.ignoreCause({ log: true }), Effect.forkDetach, Effect.asVoid); + // Answers once status subscribers can see the new ref; the network part stays detached. + const refreshGitStatusAfterCheckout = (cwd: string) => + vcsStatusBroadcaster.refreshLocalStatus(cwd).pipe( + Effect.ignoreCause({ log: true }), + Effect.andThen(refreshGitStatus(cwd, { skipLocalRefresh: true })), + // A client that disconnects mid-refresh must not leave every other client on the old ref. + Effect.forkDetach, + Effect.flatMap(Fiber.join), + ); + const getOrchestrationV2ArchivedShellSnapshot = Effect.gen(function* () { const { threads, projects, snapshotSequence } = yield* loadShellSnapshotParts({ sql, @@ -3180,7 +3190,9 @@ export const layerWsRpc = ( [WS_METHODS.vcsCreateRef]: (input) => gitWorkflow.createRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), [WS_METHODS.vcsSwitchRef]: (input) => - gitWorkflow.switchRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), + gitWorkflow + .switchRef(input) + .pipe(Effect.tap(() => refreshGitStatusAfterCheckout(input.cwd))), [WS_METHODS.vcsInit]: (input) => vcsProvisioning.initRepository(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), [WS_METHODS.reviewGetBranchCommits]: (input) => review.getBranchCommits(input), From 061f25d7709fba7c6d3e2459fd02c6524d577471 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 6 Nov 2026 14:25:54 +0000 Subject: [PATCH 3/3] fix(server): creating and switching to a branch also answers once its status is published The branch picker's create action runs vcsCreateRef with switchRef and had the same stale-label window as a plain switch. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/ws.test.ts | 214 ++++++++++++++++++++----------------- apps/server/src/ws.ts | 10 +- 2 files changed, 125 insertions(+), 99 deletions(-) diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 60d9db37a05b4..2c34c0b0688e4 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -1436,110 +1436,128 @@ it.effect("a publish whose push fails still refreshes the new remote", () => ), ); -it.effect("a branch switch answers only after status subscribers can see the new branch", () => - Effect.gen(function* () { - let checkedOut = "main"; - const published = yield* Ref.make("main"); - const refreshStarted = yield* Deferred.make(); - const statusAtResponse = yield* Queue.unbounded(); - const local: VcsStatusLocalResult = { - isRepo: true, - hasPrimaryRemote: false, - isDefaultRef: false, - refName: "main", - hasWorkingTreeChanges: false, - workingTree: { files: [], insertions: 0, deletions: 0 }, - }; - // A large repository's git status takes seconds after the checkout itself has finished. - const readAndPublishStatus = Effect.gen(function* () { - const refName = checkedOut; - yield* Deferred.succeed(refreshStarted, undefined); - yield* Effect.sleep("2 seconds"); - yield* Ref.set(published, refName); - return { ...local, refName }; - }); - const services = Layer.mergeAll( - layerUnusedConfigRpcServices, - Layer.mock(GitWorkflowService.GitWorkflowService)({ - switchRef: (input) => - Effect.sync(() => { - checkedOut = input.refName; - return { refName: input.refName }; - }), - }), - Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({ - refreshLocalStatus: () => readAndPublishStatus, - refreshStatus: () => - readAndPublishStatus.pipe( - Effect.map((status) => ({ - ...status, - hasUpstream: false, - aheadCount: 0, - behindCount: 0, - pr: null, - })), +it.effect.each([ + { + name: "switching", + tag: WS_METHODS.vcsSwitchRef, + payload: { cwd: "/repo", refName: "feature" }, + }, + { + name: "creating and switching to", + tag: WS_METHODS.vcsCreateRef, + payload: { cwd: "/repo", refName: "feature", switchRef: true }, + }, +] as const)( + "$name a branch answers only after status subscribers can see the new branch", + (request) => + Effect.gen(function* () { + let checkedOut = "main"; + const published = yield* Ref.make("main"); + const refreshStarted = yield* Deferred.make(); + const statusAtResponse = yield* Queue.unbounded(); + const local: VcsStatusLocalResult = { + isRepo: true, + hasPrimaryRemote: false, + isDefaultRef: false, + refName: "main", + hasWorkingTreeChanges: false, + workingTree: { files: [], insertions: 0, deletions: 0 }, + }; + // A large repository's git status takes seconds after the checkout itself has finished. + const readAndPublishStatus = Effect.gen(function* () { + const refName = checkedOut; + yield* Deferred.succeed(refreshStarted, undefined); + yield* Effect.sleep("2 seconds"); + yield* Ref.set(published, refName); + return { ...local, refName }; + }); + const services = Layer.mergeAll( + layerUnusedConfigRpcServices, + Layer.mock(GitWorkflowService.GitWorkflowService)({ + switchRef: (input) => + Effect.sync(() => { + checkedOut = input.refName; + return { refName: input.refName }; + }), + createRef: (input) => + Effect.sync(() => { + if (input.switchRef) checkedOut = input.refName; + return { refName: input.refName }; + }), + }), + Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({ + refreshLocalStatus: () => readAndPublishStatus, + refreshStatus: () => + readAndPublishStatus.pipe( + Effect.map((status) => ({ + ...status, + hasUpstream: false, + aheadCount: 0, + behindCount: 0, + pr: null, + })), + ), + }), + Layer.mock(ExternalLauncher.ExternalLauncher)({}), + Layer.mock(ProviderRegistry.ProviderRegistry)({}), + Layer.mock(UsageLimitSources.UsageLimitSources)({}), + Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({}), + Layer.mock(DirectEndpoints.DirectEndpoints)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/unused" }), + Layer.mock(ScratchThreadPromotion.ScratchThreadPromotion)({}), + Layer.mock(ServerEnvironment.ServerEnvironment)({}), + Layer.mock(EnvironmentAuth.EnvironmentAuth)({}), + ); + const server = yield* RpcServer.makeNoSerialization(WsRpcGroup, { + ...WS_RPC_SERVER_OPTIONS, + // Runs as the response is produced, before any later fiber can publish status. + onFromServer: (response) => + response._tag === "Exit" + ? Ref.get(published).pipe(Effect.flatMap((ref) => Queue.offer(statusAtResponse, ref))) + : Effect.void, + }).pipe( + Effect.provide( + layerWsRpc( + { + sessionId: AuthSessionId.make("switch-ref-test"), + subject: "switch-ref-test", + method: "browser-session-cookie", + scopes: [AuthSourceControlWriteScope], + }, + {}, + yield* ServerBrowser.ServerBrowser, + () => Effect.void, + { secretReferences: true }, + ).pipe( + Layer.provide(services), + Layer.merge(RpcAuthorization.layer([AuthSourceControlWriteScope])), ), - }), - Layer.mock(ExternalLauncher.ExternalLauncher)({}), - Layer.mock(ProviderRegistry.ProviderRegistry)({}), - Layer.mock(UsageLimitSources.UsageLimitSources)({}), - Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({}), - Layer.mock(DirectEndpoints.DirectEndpoints)({}), - Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/unused" }), - Layer.mock(ScratchThreadPromotion.ScratchThreadPromotion)({}), - Layer.mock(ServerEnvironment.ServerEnvironment)({}), - Layer.mock(EnvironmentAuth.EnvironmentAuth)({}), - ); - const server = yield* RpcServer.makeNoSerialization(WsRpcGroup, { - ...WS_RPC_SERVER_OPTIONS, - // Runs as the response is produced, before any later fiber can publish status. - onFromServer: (response) => - response._tag === "Exit" - ? Ref.get(published).pipe(Effect.flatMap((ref) => Queue.offer(statusAtResponse, ref))) - : Effect.void, + ), + ); + yield* server.write(0, { + _tag: "Request", + id: RpcMessage.RequestId("switch"), + tag: request.tag, + payload: request.payload, + headers: Headers.empty, + }); + yield* Deferred.await(refreshStarted); + yield* TestClock.adjust("2 seconds"); + assert.equal( + yield* Queue.take(statusAtResponse), + "feature", + `${request.tag} resolved while status subscribers still saw the previous branch`, + ); }).pipe( Effect.provide( - layerWsRpc( - { - sessionId: AuthSessionId.make("switch-ref-test"), - subject: "switch-ref-test", - method: "browser-session-cookie", - scopes: [AuthSourceControlWriteScope], - }, - {}, - yield* ServerBrowser.ServerBrowser, - () => Effect.void, - { secretReferences: true }, - ).pipe( - Layer.provide(services), - Layer.merge(RpcAuthorization.layer([AuthSourceControlWriteScope])), + Layer.mergeAll( + Layer.fresh(layerConfigSources), + Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), + Layer.mock(ServerBrowser.ServerBrowser)({}), ), ), - ); - yield* server.write(0, { - _tag: "Request", - id: RpcMessage.RequestId("switch"), - tag: WS_METHODS.vcsSwitchRef, - payload: { cwd: "/repo", refName: "feature" }, - headers: Headers.empty, - }); - yield* Deferred.await(refreshStarted); - yield* TestClock.adjust("2 seconds"); - assert.equal( - yield* Queue.take(statusAtResponse), - "feature", - "vcsSwitchRef resolved while status subscribers still saw the previous branch", - ); - }).pipe( - Effect.provide( - Layer.mergeAll( - Layer.fresh(layerConfigSources), - Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), - Layer.mock(ServerBrowser.ServerBrowser)({}), - ), + Effect.scoped, ), - Effect.scoped, - ), ); // Pruned history cannot be replayed: a cursor below the boundary must resync from a snapshot. diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index d3c82ed646774..2e52a479c08af 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -3188,7 +3188,15 @@ export const layerWsRpc = ( [WS_METHODS.vcsRemoveWorktree]: (input) => gitWorkflow.removeWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), [WS_METHODS.vcsCreateRef]: (input) => - gitWorkflow.createRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), + gitWorkflow + .createRef(input) + .pipe( + Effect.tap(() => + input.switchRef + ? refreshGitStatusAfterCheckout(input.cwd) + : refreshGitStatus(input.cwd), + ), + ), [WS_METHODS.vcsSwitchRef]: (input) => gitWorkflow .switchRef(input)