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.test.ts b/apps/server/src/ws.test.ts index 2be9e61c7c916..2c34c0b0688e4 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,130 @@ it.effect("a publish whose push fails still refreshes the new remote", () => ), ); +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])), + ), + ), + ); + 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( + 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", diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 035f8b1688d44..2e52a479c08af 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, @@ -3178,9 +3188,19 @@ 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).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),