From 97b59761021f98b4a5de2c46ec23d01f22c027f5 Mon Sep 17 00:00:00 2001 From: Wout Stiens <71498452+StiensWout@users.noreply.github.com> Date: Sun, 8 Nov 2026 12:37:06 +0100 Subject: [PATCH 1/5] perf(vcs): read branches without computing sidebar change totals Co-authored-by: GPT 6.1 Sol --- .../git/GitBranchesSheet.occupancy.test.tsx | 2 +- .../threads/git/GitBranchesSheet.test.tsx | 2 +- .../features/threads/git/GitBranchesSheet.tsx | 2 +- apps/server/src/auth/RpcAuthorization.test.ts | 41 ++++++ apps/server/src/auth/RpcAuthorization.ts | 2 + .../src/observability/RpcInstrumentation.ts | 2 + .../src/vcs/VcsStatusBroadcaster.test.ts | 129 +++++++++++++++++- apps/server/src/vcs/VcsStatusBroadcaster.ts | 72 +++++++++- apps/server/src/ws.ts | 2 + .../components/sidebar/SidebarThreadRow.tsx | 25 +++- apps/web/src/state/vcs.ts | 6 + packages/client-runtime/src/rpc/client.ts | 1 + packages/client-runtime/src/state/vcs.test.ts | 58 ++++++++ packages/client-runtime/src/state/vcs.ts | 12 ++ packages/contracts/src/git.ts | 3 + packages/contracts/src/rpc.ts | 18 +++ 16 files changed, 366 insertions(+), 11 deletions(-) diff --git a/apps/mobile/src/features/threads/git/GitBranchesSheet.occupancy.test.tsx b/apps/mobile/src/features/threads/git/GitBranchesSheet.occupancy.test.tsx index 26337e8be75a4..b2d29db8c8c72 100644 --- a/apps/mobile/src/features/threads/git/GitBranchesSheet.occupancy.test.tsx +++ b/apps/mobile/src/features/threads/git/GitBranchesSheet.occupancy.test.tsx @@ -51,7 +51,7 @@ vi.mock("../../../state/use-selected-thread-git-actions", () => ({ useSelectedThreadGitActions: () => ({ canChangeThreadBranch: true }), })); vi.mock("../../../state/query", () => ({ useEnvironmentQuery: () => ({ data: null }) })); -vi.mock("../../../state/vcs", () => ({ vcsEnvironment: { status: () => null } })); +vi.mock("../../../state/vcs", () => ({ vcsEnvironment: { branch: () => null } })); import { GitBranchesSheet } from "./GitBranchesSheet"; diff --git a/apps/mobile/src/features/threads/git/GitBranchesSheet.test.tsx b/apps/mobile/src/features/threads/git/GitBranchesSheet.test.tsx index 509dd51240219..6bd008116babb 100644 --- a/apps/mobile/src/features/threads/git/GitBranchesSheet.test.tsx +++ b/apps/mobile/src/features/threads/git/GitBranchesSheet.test.tsx @@ -71,7 +71,7 @@ vi.mock("../../../state/use-selected-thread-git-actions", () => ({ }, }), })); -vi.mock("../../../state/vcs", () => ({ vcsEnvironment: { status: () => null } })); +vi.mock("../../../state/vcs", () => ({ vcsEnvironment: { branch: () => null } })); vi.mock("./gitSheetComponents", () => ({ SheetActionButton: (props: ActionProps) => { state.actions.set(props.label, props); diff --git a/apps/mobile/src/features/threads/git/GitBranchesSheet.tsx b/apps/mobile/src/features/threads/git/GitBranchesSheet.tsx index f8ddd8d7498bb..67d0818ee6c4d 100644 --- a/apps/mobile/src/features/threads/git/GitBranchesSheet.tsx +++ b/apps/mobile/src/features/threads/git/GitBranchesSheet.tsx @@ -33,7 +33,7 @@ export function GitBranchesSheet(_props: GitBranchesSheetProps) { const gitStatus = useEnvironmentQuery( selectedThread !== null && selectedThreadCwd !== null - ? vcsEnvironment.status({ + ? vcsEnvironment.branch({ environmentId: selectedThread.environmentId, input: { cwd: selectedThreadCwd }, }) diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index e59c428dca0aa..b030f60b30e54 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -407,3 +407,44 @@ it.effect("denies manual cleanup before the handler without settings permission" expect(handled).toBe(false); }).pipe(Effect.scoped), ); + +describe("branch observation authorization", () => { + const tested = [WS_METHODS.subscribeVcsBranch, WS_METHODS.vcsRefreshBranch] as const; + const group = WsRpcGroup.omit( + ...[...WsRpcGroup.requests.keys()].filter( + (tag): tag is Exclude => + !(tested as ReadonlyArray).includes(tag), + ), + ); + it.effect.each([false, true])("requires thread read permission (granted: %s)", (granted) => + Effect.gen(function* () { + const branch = { refName: "main" }; + const client = yield* RpcTest.makeClient(group).pipe( + Effect.provide( + Layer.mergeAll( + group.toLayerHandler(WS_METHODS.subscribeVcsBranch, () => Stream.make(branch)), + group.toLayerHandler(WS_METHODS.vcsRefreshBranch, () => Effect.succeed(branch)), + RpcAuthorization.layer(granted ? [AuthOrchestrationReadScope] : []), + ), + ), + ); + const refreshed = yield* client[WS_METHODS.vcsRefreshBranch]({ cwd: "/repo" }).pipe( + Effect.result, + ); + const subscribed = yield* Stream.runCollect( + client[WS_METHODS.subscribeVcsBranch]({ cwd: "/repo" }), + ).pipe(Effect.result); + if (granted) { + expect(refreshed).toMatchObject({ success: branch }); + expect(subscribed).toMatchObject({ success: [branch] }); + } else { + expect(refreshed).toMatchObject({ + failure: { requiredPermission: AuthOrchestrationReadScope }, + }); + expect(subscribed).toMatchObject({ + failure: { requiredPermission: AuthOrchestrationReadScope }, + }); + } + }).pipe(Effect.scoped), + ); +}); diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 174a5e8b29cee..70b8db54f6775 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -169,6 +169,8 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.mcpAppsReadResource]: AuthOrchestrationReadScope, [WS_METHODS.mcpAppsUpdateModelContext]: AuthOrchestrationOperateScope, [WS_METHODS.subscribeVcsStatus]: AuthOrchestrationReadScope, + [WS_METHODS.subscribeVcsBranch]: AuthOrchestrationReadScope, + [WS_METHODS.vcsRefreshBranch]: AuthOrchestrationReadScope, [WS_METHODS.subscribeWorktreeSetup]: AuthOrchestrationReadScope, [WS_METHODS.worktreeSetupCancel]: AuthOrchestrationOperateScope, [WS_METHODS.subscribeResourceTelemetry]: AuthDiagnosticsReadScope, diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index d19f83a932d1f..c6b60310d4939 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -173,6 +173,8 @@ const RPC_AGGREGATES = { [WS_METHODS.mcpAppsUpdateModelContext]: "provider", [WS_METHODS.mcpAppsReadResource]: "provider", [WS_METHODS.subscribeVcsStatus]: "vcs", + [WS_METHODS.subscribeVcsBranch]: "vcs", + [WS_METHODS.vcsRefreshBranch]: "vcs", [WS_METHODS.subscribeWorktreeSetup]: "vcs", [WS_METHODS.worktreeSetupCancel]: "vcs", [WS_METHODS.subscribeResourceTelemetry]: "server", diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 60f39b592d6de..5563dd91a226b 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -77,6 +77,7 @@ function layerTestFor(state: { automaticFetchInterval?: Effect.Effect; beforeRemoteStatus?: Effect.Effect; beforeLocalStatus?: Effect.Effect; + beforeCurrentBranch?: Effect.Effect; fileSystem?: FileSystem.FileSystem; pullCurrentBranch?: GitWorkflowService.GitWorkflowService["Service"]["pullCurrentBranch"]; }) { @@ -102,7 +103,10 @@ function layerTestFor(state: { Effect.as(state.currentLocalStatus), ); }), - currentBranch: () => Effect.sync(() => state.currentLocalStatus.refName), + currentBranch: () => + (state.beforeCurrentBranch ?? Effect.void).pipe( + Effect.andThen(Effect.sync(() => state.currentLocalStatus.refName)), + ), remoteStatus: (_input, options) => Effect.suspend(() => { return (state.beforeRemoteStatus ?? Effect.void).pipe( @@ -207,6 +211,129 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); }); + it.effect( + "streams fresh branch-only reads, detached HEAD, and detailed demand without mixing results", + () => { + const state: Parameters[0] = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: null, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + fileSystem: FileSystem.makeNoop({ + realPath: (cwd) => Effect.succeed(cwd === "/alias" ? "/repo" : cwd), + }), + }; + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const branches = yield* Stream.toPull(broadcaster.streamBranchStatus({ cwd: "/repo" })); + assert.deepEqual(yield* branches, [{ refName: baseLocalStatus.refName }]); + assert.equal(state.localStatusCalls, 0); + state.currentLocalStatus = { ...baseLocalStatus, refName: "external" }; + yield* broadcaster.refreshBranchStatus({ cwd: "/repo" }); + assert.deepEqual(yield* branches, [{ refName: "external" }]); + state.currentLocalStatus = { ...baseLocalStatus, refName: null }; + yield* broadcaster.refreshBranchStatus({ cwd: "/repo" }); + assert.deepEqual(yield* branches, [{ refName: null }]); + assert.equal(state.localStatusCalls, 0); + const detailScope = yield* Scope.make(); + const details = yield* Stream.toPull( + broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }), + ).pipe(Scope.provide(detailScope)); + assert.deepEqual(yield* details, [ + { _tag: "snapshot", local: state.currentLocalStatus, remote: null }, + ]); + state.currentLocalStatus = { + ...baseLocalStatus, + refName: "changed", + workingTree: { files: [], insertions: 12, deletions: 3 }, + }; + yield* broadcaster.refreshBranchStatus({ cwd: "/repo" }); + assert.deepEqual(yield* details, [ + { _tag: "localUpdated", local: state.currentLocalStatus }, + ]); + assert.deepEqual(yield* branches, [{ refName: "changed" }]); + const secondScope = yield* Scope.make(); + const secondDetails = yield* Stream.toPull( + broadcaster.streamStatus({ cwd: "/alias", includeRemote: false }), + ).pipe(Scope.provide(secondScope)); + yield* secondDetails; + yield* Scope.close(detailScope, Exit.void); + state.currentLocalStatus = { + ...state.currentLocalStatus, + workingTree: { files: [], insertions: 20, deletions: 4 }, + }; + yield* broadcaster.refreshBranchStatus({ cwd: "/repo" }); + assert.deepEqual(yield* secondDetails, [ + { _tag: "localUpdated", local: state.currentLocalStatus }, + ]); + yield* Scope.close(secondScope, Exit.void); + const callsAfterRelease = state.localStatusCalls; + state.currentLocalStatus = { + ...state.currentLocalStatus, + workingTree: { files: [], insertions: 30, deletions: 5 }, + }; + yield* broadcaster.refreshBranchStatus({ cwd: "/repo" }); + assert.equal(state.localStatusCalls, callsAfterRelease); + const reopened = yield* Stream.toPull( + broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }), + ); + assert.deepEqual(yield* reopened, [ + { _tag: "snapshot", local: state.currentLocalStatus, remote: null }, + ]); + state.currentLocalStatus = { + ...baseLocalStatus, + isRepo: false, + refName: null, + }; + const missing = yield* Stream.toPull(broadcaster.streamBranchStatus({ cwd: "/missing" })); + assert.deepEqual(yield* missing, [{ refName: null }]); + yield* broadcaster.refreshBranchStatus({ cwd: "/repo" }); + assert.deepEqual(yield* reopened, [ + { _tag: "localUpdated", local: state.currentLocalStatus }, + ]); + assert.deepEqual(yield* branches, [{ refName: null }]); + assert.equal(state.remoteStatusCalls, 0); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }, + ); + + it.effect("upgrades an in-flight branch refresh when detailed demand arrives", () => { + const state: Parameters[0] = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: null, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const reading = yield* Deferred.make(); + const resume = yield* Deferred.make(); + state.beforeCurrentBranch = Deferred.succeed(reading, undefined).pipe( + Effect.andThen(Deferred.await(resume)), + ); + const refresh = yield* broadcaster + .refreshBranchStatus({ cwd: "/repo" }) + .pipe(Effect.forkScoped); + yield* Deferred.await(reading); + const details = yield* Stream.toPull( + broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }), + ); + yield* details; + state.currentLocalStatus = { + ...baseLocalStatus, + workingTree: { files: [], insertions: 8, deletions: 2 }, + }; + yield* Deferred.succeed(resume, undefined); + yield* Fiber.join(refresh); + assert.equal(state.localStatusCalls, 2); + assert.deepEqual(yield* details, [{ _tag: "localUpdated", local: state.currentLocalStatus }]); + }).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..a5f9f23b18884 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -15,6 +15,7 @@ import * as SynchronizedRef from "effect/SynchronizedRef"; import type { GitManagerServiceError, VcsStatusInput, + VcsBranchStatus, VcsStatusLocalResult, VcsStatusRemoteResult, VcsStatusResult, @@ -205,6 +206,12 @@ function remoteRefreshFailureDelay( export class VcsStatusBroadcaster extends Context.Service< VcsStatusBroadcaster, { + readonly streamBranchStatus: ( + input: VcsStatusInput, + ) => Stream.Stream; + readonly refreshBranchStatus: ( + input: VcsStatusInput, + ) => Effect.Effect; readonly streamBranchChanges: Stream.Stream<{ readonly cwd: string; readonly refName: string | null; @@ -256,6 +263,16 @@ export const make = Effect.gen(function* () { const broadcasterScope = yield* Effect.acquireRelease(Scope.make(), (scope) => Scope.close(scope, Exit.void), ); + const detailedSubscribers = new Map(); + const branches = new Map(); + const publishBranch = Effect.fn("VcsStatusBroadcaster.publishBranch")(function* ( + cwd: string, + refName: string | null, + ) { + if (branches.has(cwd) && branches.get(cwd) === refName) return; + branches.set(cwd, refName); + yield* PubSub.publish(branchChangesPubSub, { cwd, refName }); + }); const cacheRef = yield* Ref.make(new Map()); // One permit per cwd for remote reads that write the cache. Without it a // periodic poll that started before an agent creates a PR can finish after the @@ -286,8 +303,8 @@ export const make = Effect.gen(function* () { new Map(cache).set(cwd, { local: nextLocal, remote }), ] as const; }); - if (branchChanged && local.isRepo) { - yield* PubSub.publish(branchChangesPubSub, { cwd, refName: local.refName }); + if (branchChanged) { + yield* publishBranch(cwd, local.refName); } if (options?.publish && remoteCleared) { yield* PubSub.publish(changesPubSub, { @@ -738,6 +755,18 @@ export const make = Effect.gen(function* () { Stream.unwrap( Effect.gen(function* () { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); + yield* Effect.acquireRelease( + Effect.sync(() => { + const previous = detailedSubscribers.get(cwd) ?? 0; + detailedSubscribers.set(cwd, previous + 1); + }), + () => + Effect.sync(() => { + const remaining = (detailedSubscribers.get(cwd) ?? 1) - 1; + if (remaining === 0) detailedSubscribers.delete(cwd); + else detailedSubscribers.set(cwd, remaining); + }), + ); const subscription = yield* PubSub.subscribe(changesPubSub); const initialLocal = yield* getOrLoadLocalStatus(cwd); const cachedStatus = yield* getCachedStatus(cwd); @@ -768,7 +797,46 @@ export const make = Effect.gen(function* () { }), ); + const refreshBranchStatus = Effect.fn("VcsStatusBroadcaster.refreshBranchStatus")(function* ( + input: VcsStatusInput, + ) { + const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); + if (!detailedSubscribers.has(cwd)) { + const refName = yield* workflow.currentBranch(cwd); + // Branch reads cannot certify cached Changes totals, even if HEAD did not move. + yield* workflow.invalidateLocalStatus(cwd); + if (!detailedSubscribers.has(cwd)) { + yield* Ref.update(cacheRef, (cache) => { + const previous = cache.get(cwd); + return previous ? new Map(cache).set(cwd, { ...previous, local: null }) : cache; + }); + yield* publishBranch(cwd, refName); + return { refName }; + } + } + const local = yield* refreshLocalStatusCore(cwd); + return { refName: local.refName }; + }); + + const streamBranchStatus: VcsStatusBroadcaster["Service"]["streamBranchStatus"] = (input) => + Stream.unwrap( + Effect.gen(function* () { + const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); + const subscription = yield* PubSub.subscribe(branchChangesPubSub); + const branch = yield* refreshBranchStatus({ cwd }); + return Stream.concat( + Stream.make(branch), + Stream.fromSubscription(subscription).pipe( + Stream.filter((event) => event.cwd === cwd), + Stream.map(({ refName }) => ({ refName })), + ), + ).pipe(Stream.changesWith((left, right) => left.refName === right.refName)); + }), + ); + return VcsStatusBroadcaster.of({ + streamBranchStatus, + refreshBranchStatus, streamBranchChanges: Stream.fromPubSub(branchChangesPubSub), getStatus, refreshLocalStatus, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 1fa81ddc7026e..4249d5e096dfc 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -3112,6 +3112,8 @@ export const layerWsRpc = ( }), [WS_METHODS.assetsPersistChatAttachments]: (input) => persistChatAttachments(input).pipe(Effect.map((attachments) => ({ attachments }))), + [WS_METHODS.subscribeVcsBranch]: (input) => vcsStatusBroadcaster.streamBranchStatus(input), + [WS_METHODS.vcsRefreshBranch]: (input) => vcsStatusBroadcaster.refreshBranchStatus(input), [WS_METHODS.subscribeVcsStatus]: (input) => vcsStatusBroadcaster.streamStatus(input), [WS_METHODS.subscribeWorktreeSetup]: (input) => worktreeSetupTracker.stream(input.threadId), [WS_METHODS.worktreeSetupCancel]: (input) => diff --git a/apps/web/src/components/sidebar/SidebarThreadRow.tsx b/apps/web/src/components/sidebar/SidebarThreadRow.tsx index 41cb733daff15..58af96e635244 100644 --- a/apps/web/src/components/sidebar/SidebarThreadRow.tsx +++ b/apps/web/src/components/sidebar/SidebarThreadRow.tsx @@ -1,3 +1,4 @@ +import { useAtomMount } from "@effect/atom-react"; import { ThreadHoverCard, ThreadHoverCardPopup } from "../ThreadHoverCard"; import { ThreadTerminalStatusButton } from "../ThreadStatusIndicators"; import { discardComposerDraft } from "../../lib/discardComposerDraft"; @@ -68,7 +69,7 @@ import { useUiStateStore } from "../../uiStateStore"; import { useThreadSelectionStore } from "../../threadSelectionStore"; import { useNowMinute } from "../../hooks/useNowMinute"; import { useNowSecond } from "../../hooks/useNowSecond"; -import { vcsEnvironment } from "../../state/vcs"; +import { vcsEnvironment, vcsBranchRevalidation } from "../../state/vcs"; import { useEnvironmentQuery } from "../../state/query"; import { readEnvironmentScope, useEnvironmentScope } from "../../state/session"; import { formatRelativeTimeLabel, parseTimestampDate } from "../../timestampFormat"; @@ -689,12 +690,19 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP ); const gitStatus = useEnvironmentQuery( leaseLiveStatus && (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null - ? vcsEnvironment.status({ + ? vcsEnvironment.branch({ environmentId: thread.environmentId, - input: { cwd: gitCwd, includeRemote: false }, + input: { cwd: gitCwd }, }) : null, ); + useAtomMount( + vcsBranchRevalidation( + leaseLiveStatus && (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null + ? { environmentId: thread.environmentId, cwd: gitCwd } + : null, + ), + ); const visibleGitStatus = useRetainedValue( JSON.stringify([thread.environmentId, gitCwd]), gitStatus.data, @@ -1642,12 +1650,19 @@ export const SidebarSearchResultRow = memo(function SidebarSearchResultRow(props const gitCwd = thread.worktreePath ?? props.project?.workspaceRoot ?? null; const gitStatus = useEnvironmentQuery( leaseLiveStatus && (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null - ? vcsEnvironment.status({ + ? vcsEnvironment.branch({ environmentId: thread.environmentId, - input: { cwd: gitCwd, includeRemote: false }, + input: { cwd: gitCwd }, }) : null, ); + useAtomMount( + vcsBranchRevalidation( + leaseLiveStatus && (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null + ? { environmentId: thread.environmentId, cwd: gitCwd } + : null, + ), + ); const visibleGitStatus = useRetainedValue( JSON.stringify([thread.environmentId, gitCwd]), gitStatus.data, diff --git a/apps/web/src/state/vcs.ts b/apps/web/src/state/vcs.ts index fd4f34caaa68d..7604ff3b23773 100644 --- a/apps/web/src/state/vcs.ts +++ b/apps/web/src/state/vcs.ts @@ -19,3 +19,9 @@ export const vcsStatusRevalidation = createVcsStatusRevalidationAtomFamily({ subscribeWake: subscribeDocumentReturn, wakeDebounceMs: 250, }); + +export const vcsBranchRevalidation = createVcsStatusRevalidationAtomFamily({ + refreshStatus: vcsEnvironment.refreshBranch, + subscribeWake: subscribeDocumentReturn, + wakeDebounceMs: 250, +}); diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index e74abde6a2c59..21c4c146ac25b 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -68,6 +68,7 @@ export type EnvironmentSubscriptionRpcTag = | typeof WS_METHODS.subscribeResourceTelemetry | typeof WS_METHODS.pullRequestsSubscribeRefreshes | typeof WS_METHODS.subscribeVcsStatus + | typeof WS_METHODS.subscribeVcsBranch | typeof WS_METHODS.subscribeWorktreeSetup | typeof WS_METHODS.subscribeProjectClones | typeof WS_METHODS.terminalAttach diff --git a/packages/client-runtime/src/state/vcs.test.ts b/packages/client-runtime/src/state/vcs.test.ts index cdbda23422237..b11d483328cb9 100644 --- a/packages/client-runtime/src/state/vcs.test.ts +++ b/packages/client-runtime/src/state/vcs.test.ts @@ -124,6 +124,64 @@ function cacheWithRefs( }); } +describe("branch subscriptions", () => { + it.effect("shares checkout demand, reconnects, and releases the old stream", () => + Effect.gen(function* () { + const released = yield* Queue.unbounded(); + let subscriptions = 0; + const clientFor = (refName: string) => + ({ + [WS_METHODS.subscribeVcsBranch]: () => { + subscriptions += 1; + return Stream.concat(Stream.make({ refName }), Stream.never).pipe( + Stream.ensuring(Queue.offer(released, refName)), + ); + }, + }) as unknown as WsRpcProtocolClient; + const activeSession = yield* SubscriptionRef.make(Option.some(session(clientFor("main")))); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE), + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + }); + const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({ + followStream: (_environmentId, stream) => + Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + } as EnvironmentRegistry.EnvironmentRegistry["Service"]); + const runtime = Atom.runtime( + Layer.merge( + Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry), + Layer.succeed(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())), + ), + ); + const atoms = createVcsEnvironmentAtoms(runtime); + const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (value) => + Effect.sync(() => value.dispose()), + ); + const target = { environmentId: TARGET.environmentId, input: { cwd: "/repo" } }; + const first = yield* Stream.toPull( + AtomRegistry.toStreamResult(registry, atoms.branch(target)), + ); + expect(yield* first).toEqual([{ refName: "main" }]); + const second = yield* Stream.toPull( + AtomRegistry.toStreamResult(registry, atoms.branch(target)), + ); + expect(yield* second).toEqual([{ refName: "main" }]); + expect(subscriptions).toBe(1); + yield* SubscriptionRef.set(activeSession, Option.some(session(clientFor("external")))); + expect(yield* Queue.take(released)).toBe("main"); + expect(yield* first).toEqual([{ refName: "external" }]); + expect(yield* second).toEqual([{ refName: "external" }]); + registry.dispose(); + expect(yield* Queue.take(released)).toBe("external"); + }).pipe(Effect.scoped), + ); +}); + describe("cached VCS refs", () => { it("invalidates all ref streams in the mutated environment", () => { const registry = AtomRegistry.make(); diff --git a/packages/client-runtime/src/state/vcs.ts b/packages/client-runtime/src/state/vcs.ts index ef508f2126c7f..a5679c6d59d14 100644 --- a/packages/client-runtime/src/state/vcs.ts +++ b/packages/client-runtime/src/state/vcs.ts @@ -294,6 +294,18 @@ export function createVcsEnvironmentAtoms( label: "environment-data:vcs:read-refs", tag: WS_METHODS.vcsListRefs, }), + branch: createEnvironmentSubscriptionAtomFamily(runtime, { + label: "environment-data:vcs:branch", + idleTtlMs: VCS_STATUS_IDLE_TTL_MS, + subscribe: (input: EnvironmentRpcInput) => + subscribe(WS_METHODS.subscribeVcsBranch, input), + }), + refreshBranch: createEnvironmentRpcCommand(runtime, { + label: "environment-data:vcs:refresh-branch", + tag: WS_METHODS.vcsRefreshBranch, + scheduler: vcsCommandScheduler, + concurrency: vcsCommandConcurrency, + }), status: createEnvironmentSubscriptionAtomFamily(runtime, { label: "environment-data:vcs:status", idleTtlMs: VCS_STATUS_IDLE_TTL_MS, diff --git a/packages/contracts/src/git.ts b/packages/contracts/src/git.ts index 0dc958f59c490..a3b5e34076219 100644 --- a/packages/contracts/src/git.ts +++ b/packages/contracts/src/git.ts @@ -111,6 +111,9 @@ export const VcsStatusInput = Schema.Struct({ }); export type VcsStatusInput = typeof VcsStatusInput.Type; +export const VcsBranchStatus = Schema.Struct({ refName: Schema.NullOr(Schema.String) }); +export type VcsBranchStatus = typeof VcsBranchStatus.Type; + export const VcsStatusSubscriptionInput = Schema.Struct({ ...VcsStatusInput.fields, /** Passive observers receive cached remote status without retaining its refresh loop. */ diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index a9868d64a8110..e401903c1e84e 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -145,6 +145,7 @@ import { GitResolvePullRequestResult, GitRunStackedActionInput, VcsStatusInput, + VcsBranchStatus, VcsStatusSubscriptionInput, VcsStatusResult, VcsStatusStreamEvent, @@ -605,6 +606,8 @@ export const WS_METHODS = { subscribeProjectClones: "subscribeProjectClones", subscribeVcsStatus: "subscribeVcsStatus", + subscribeVcsBranch: "subscribeVcsBranch", + vcsRefreshBranch: "vcs.refreshBranch", subscribeWorktreeSetup: "subscribeWorktreeSetup", worktreeSetupCancel: "worktreeSetup.cancel", subscribeTerminalMetadata: "subscribeTerminalMetadata", @@ -1454,6 +1457,19 @@ const WsProviderUploadFeedbackRpc = Rpc.make(WS_METHODS.providerUploadFeedback, error: Schema.Union([ProviderUploadFeedbackError, EnvironmentAuthorizationError]), }); +const WsSubscribeVcsBranchRpc = Rpc.make(WS_METHODS.subscribeVcsBranch, { + payload: VcsStatusInput, + success: VcsBranchStatus, + error: Schema.Union([GitManagerServiceError, EnvironmentAuthorizationError]), + stream: true, +}); + +const WsVcsRefreshBranchRpc = Rpc.make(WS_METHODS.vcsRefreshBranch, { + payload: VcsStatusInput, + success: VcsBranchStatus, + error: Schema.Union([GitManagerServiceError, EnvironmentAuthorizationError]), +}); + const WsSubscribeVcsStatusRpc = Rpc.make(WS_METHODS.subscribeVcsStatus, { payload: VcsStatusSubscriptionInput, success: VcsStatusStreamEvent, @@ -2149,6 +2165,8 @@ export const WsRpcGroup = RpcGroup.make( WsMcpAppsUpdateModelContextRpc, WsProviderUploadFeedbackRpc, WsSubscribeVcsStatusRpc, + WsSubscribeVcsBranchRpc, + WsVcsRefreshBranchRpc, WsSubscribeWorktreeSetupRpc, WsWorktreeSetupCancelRpc, WsVcsPullRpc, From bd6d89322c0176b0349525b78a96e1d91339afb7 Mon Sep 17 00:00:00 2001 From: Wout Stiens <71498452+StiensWout@users.noreply.github.com> Date: Sun, 8 Nov 2026 12:41:51 +0100 Subject: [PATCH 2/5] refactor(vcs): trim sidebar branch subscription Share one branch target per sidebar row, normalize the checkout once per branch stream, keep one branch dedup owner, and drop tests that repeat generic subscription and middleware coverage. Co-authored-by: GPT 6.1 Sol Co-Authored-By: Claude Opus 5.5 --- apps/server/src/auth/RpcAuthorization.test.ts | 47 ++------------- .../src/vcs/VcsStatusBroadcaster.test.ts | 4 +- apps/server/src/vcs/VcsStatusBroadcaster.ts | 57 +++++++++--------- .../components/sidebar/SidebarThreadRow.tsx | 48 +++++++-------- packages/client-runtime/src/state/vcs.test.ts | 58 ------------------- 5 files changed, 58 insertions(+), 156 deletions(-) diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index b030f60b30e54..de94aca577021 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -68,6 +68,12 @@ describe("RPC authorization scopes", () => { } }); + it("lets thread readers observe and refresh checkout branches", () => { + for (const method of [WS_METHODS.subscribeVcsBranch, WS_METHODS.vcsRefreshBranch]) { + expect(requiredScopeForRpcMethod(method)).toBe(AuthOrchestrationReadScope); + } + }); + it("rejects unknown RPC method names", () => { for (const method of ["server.notRegistered", "toString", "constructor"]) { expect(() => requiredScopeForRpcMethod(method)).toThrow( @@ -407,44 +413,3 @@ it.effect("denies manual cleanup before the handler without settings permission" expect(handled).toBe(false); }).pipe(Effect.scoped), ); - -describe("branch observation authorization", () => { - const tested = [WS_METHODS.subscribeVcsBranch, WS_METHODS.vcsRefreshBranch] as const; - const group = WsRpcGroup.omit( - ...[...WsRpcGroup.requests.keys()].filter( - (tag): tag is Exclude => - !(tested as ReadonlyArray).includes(tag), - ), - ); - it.effect.each([false, true])("requires thread read permission (granted: %s)", (granted) => - Effect.gen(function* () { - const branch = { refName: "main" }; - const client = yield* RpcTest.makeClient(group).pipe( - Effect.provide( - Layer.mergeAll( - group.toLayerHandler(WS_METHODS.subscribeVcsBranch, () => Stream.make(branch)), - group.toLayerHandler(WS_METHODS.vcsRefreshBranch, () => Effect.succeed(branch)), - RpcAuthorization.layer(granted ? [AuthOrchestrationReadScope] : []), - ), - ), - ); - const refreshed = yield* client[WS_METHODS.vcsRefreshBranch]({ cwd: "/repo" }).pipe( - Effect.result, - ); - const subscribed = yield* Stream.runCollect( - client[WS_METHODS.subscribeVcsBranch]({ cwd: "/repo" }), - ).pipe(Effect.result); - if (granted) { - expect(refreshed).toMatchObject({ success: branch }); - expect(subscribed).toMatchObject({ success: [branch] }); - } else { - expect(refreshed).toMatchObject({ - failure: { requiredPermission: AuthOrchestrationReadScope }, - }); - expect(subscribed).toMatchObject({ - failure: { requiredPermission: AuthOrchestrationReadScope }, - }); - } - }).pipe(Effect.scoped), - ); -}); diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 5563dd91a226b..a1ed249421b48 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -212,7 +212,7 @@ describe("VcsStatusBroadcaster", () => { }); it.effect( - "streams fresh branch-only reads, detached HEAD, and detailed demand without mixing results", + "streams fresh branch-only reads, detached HEAD, non-repositories, and detailed demand without mixing results", () => { const state: Parameters[0] = { currentLocalStatus: baseLocalStatus, @@ -287,8 +287,6 @@ describe("VcsStatusBroadcaster", () => { isRepo: false, refName: null, }; - const missing = yield* Stream.toPull(broadcaster.streamBranchStatus({ cwd: "/missing" })); - assert.deepEqual(yield* missing, [{ refName: null }]); yield* broadcaster.refreshBranchStatus({ cwd: "/repo" }); assert.deepEqual(yield* reopened, [ { _tag: "localUpdated", local: state.currentLocalStatus }, diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index a5f9f23b18884..9b012737762c6 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -269,7 +269,7 @@ export const make = Effect.gen(function* () { cwd: string, refName: string | null, ) { - if (branches.has(cwd) && branches.get(cwd) === refName) return; + if (branches.get(cwd) === refName) return; branches.set(cwd, refName); yield* PubSub.publish(branchChangesPubSub, { cwd, refName }); }); @@ -291,21 +291,15 @@ export const make = Effect.gen(function* () { const updateCachedLocalStatus = Effect.fn("VcsStatusBroadcaster.updateCachedLocalStatus")( function* (cwd: string, local: VcsStatusLocalResult, options?: { publish?: boolean }) { const nextLocal = { fingerprint: fingerprintStatusPart(local), value: local }; - const [localChanged, branchChanged, remoteCleared] = yield* Ref.modify(cacheRef, (cache) => { + const [localChanged, remoteCleared] = yield* Ref.modify(cacheRef, (cache) => { const previous = cache.get(cwd) ?? { local: null, remote: null }; const remote = previous.remote?.branch === local.refName ? previous.remote : null; return [ - [ - previous.local?.fingerprint !== nextLocal.fingerprint, - previous.local?.value.refName !== local.refName, - previous.remote !== remote, - ], + [previous.local?.fingerprint !== nextLocal.fingerprint, previous.remote !== remote], new Map(cache).set(cwd, { local: nextLocal, remote }), ] as const; }); - if (branchChanged) { - yield* publishBranch(cwd, local.refName); - } + yield* publishBranch(cwd, local.refName); if (options?.publish && remoteCleared) { yield* PubSub.publish(changesPubSub, { cwd, @@ -797,25 +791,32 @@ export const make = Effect.gen(function* () { }), ); - const refreshBranchStatus = Effect.fn("VcsStatusBroadcaster.refreshBranchStatus")(function* ( - input: VcsStatusInput, - ) { - const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); - if (!detailedSubscribers.has(cwd)) { - const refName = yield* workflow.currentBranch(cwd); - // Branch reads cannot certify cached Changes totals, even if HEAD did not move. - yield* workflow.invalidateLocalStatus(cwd); + // Without detailed subscribers only HEAD is read; otherwise the full local refresh runs and + // publishes the branch with it. Demand is rechecked so a joining subscriber gets fresh totals. + const refreshBranchStatusCore = Effect.fn("VcsStatusBroadcaster.refreshBranchStatusCore")( + function* (cwd: string) { if (!detailedSubscribers.has(cwd)) { - yield* Ref.update(cacheRef, (cache) => { - const previous = cache.get(cwd); - return previous ? new Map(cache).set(cwd, { ...previous, local: null }) : cache; - }); - yield* publishBranch(cwd, refName); - return { refName }; + const refName = yield* workflow.currentBranch(cwd); + if (!detailedSubscribers.has(cwd)) { + // Branch reads cannot certify cached Changes totals, even if HEAD did not move. + yield* workflow.invalidateLocalStatus(cwd); + yield* Ref.update(cacheRef, (cache) => { + const previous = cache.get(cwd); + return previous ? new Map(cache).set(cwd, { ...previous, local: null }) : cache; + }); + yield* publishBranch(cwd, refName); + return { refName }; + } } - } - const local = yield* refreshLocalStatusCore(cwd); - return { refName: local.refName }; + const local = yield* refreshLocalStatusCore(cwd); + return { refName: local.refName }; + }, + ); + + const refreshBranchStatus: VcsStatusBroadcaster["Service"]["refreshBranchStatus"] = Effect.fn( + "VcsStatusBroadcaster.refreshBranchStatus", + )(function* (input) { + return yield* refreshBranchStatusCore(yield* withFileSystem(normalizeCwd(input.cwd))); }); const streamBranchStatus: VcsStatusBroadcaster["Service"]["streamBranchStatus"] = (input) => @@ -823,7 +824,7 @@ export const make = Effect.gen(function* () { Effect.gen(function* () { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); const subscription = yield* PubSub.subscribe(branchChangesPubSub); - const branch = yield* refreshBranchStatus({ cwd }); + const branch = yield* refreshBranchStatusCore(cwd); return Stream.concat( Stream.make(branch), Stream.fromSubscription(subscription).pipe( diff --git a/apps/web/src/components/sidebar/SidebarThreadRow.tsx b/apps/web/src/components/sidebar/SidebarThreadRow.tsx index 58af96e635244..39cf531b2a3c3 100644 --- a/apps/web/src/components/sidebar/SidebarThreadRow.tsx +++ b/apps/web/src/components/sidebar/SidebarThreadRow.tsx @@ -688,21 +688,19 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP thread.pullRequests, thread.branchPullRequest, ); - const gitStatus = useEnvironmentQuery( + const branchTarget = leaseLiveStatus && (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null - ? vcsEnvironment.branch({ - environmentId: thread.environmentId, - input: { cwd: gitCwd }, - }) - : null, - ); - useAtomMount( - vcsBranchRevalidation( - leaseLiveStatus && (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null - ? { environmentId: thread.environmentId, cwd: gitCwd } - : null, - ), + ? { environmentId: thread.environmentId, cwd: gitCwd } + : null; + const gitStatus = useEnvironmentQuery( + branchTarget === null + ? null + : vcsEnvironment.branch({ + environmentId: branchTarget.environmentId, + input: { cwd: branchTarget.cwd }, + }), ); + useAtomMount(vcsBranchRevalidation(branchTarget)); const visibleGitStatus = useRetainedValue( JSON.stringify([thread.environmentId, gitCwd]), gitStatus.data, @@ -1648,21 +1646,19 @@ export const SidebarSearchResultRow = memo(function SidebarSearchResultRow(props props.isHighlighted || props.isRouteActive, ); const gitCwd = thread.worktreePath ?? props.project?.workspaceRoot ?? null; - const gitStatus = useEnvironmentQuery( + const branchTarget = leaseLiveStatus && (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null - ? vcsEnvironment.branch({ - environmentId: thread.environmentId, - input: { cwd: gitCwd }, - }) - : null, - ); - useAtomMount( - vcsBranchRevalidation( - leaseLiveStatus && (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null - ? { environmentId: thread.environmentId, cwd: gitCwd } - : null, - ), + ? { environmentId: thread.environmentId, cwd: gitCwd } + : null; + const gitStatus = useEnvironmentQuery( + branchTarget === null + ? null + : vcsEnvironment.branch({ + environmentId: branchTarget.environmentId, + input: { cwd: branchTarget.cwd }, + }), ); + useAtomMount(vcsBranchRevalidation(branchTarget)); const visibleGitStatus = useRetainedValue( JSON.stringify([thread.environmentId, gitCwd]), gitStatus.data, diff --git a/packages/client-runtime/src/state/vcs.test.ts b/packages/client-runtime/src/state/vcs.test.ts index b11d483328cb9..cdbda23422237 100644 --- a/packages/client-runtime/src/state/vcs.test.ts +++ b/packages/client-runtime/src/state/vcs.test.ts @@ -124,64 +124,6 @@ function cacheWithRefs( }); } -describe("branch subscriptions", () => { - it.effect("shares checkout demand, reconnects, and releases the old stream", () => - Effect.gen(function* () { - const released = yield* Queue.unbounded(); - let subscriptions = 0; - const clientFor = (refName: string) => - ({ - [WS_METHODS.subscribeVcsBranch]: () => { - subscriptions += 1; - return Stream.concat(Stream.make({ refName }), Stream.never).pipe( - Stream.ensuring(Queue.offer(released, refName)), - ); - }, - }) as unknown as WsRpcProtocolClient; - const activeSession = yield* SubscriptionRef.make(Option.some(session(clientFor("main")))); - const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ - target: TARGET, - state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE), - session: activeSession, - prepared: yield* SubscriptionRef.make(Option.none()), - connect: Effect.void, - disconnect: Effect.void, - retryNow: Effect.void, - }); - const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({ - followStream: (_environmentId, stream) => - Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), - } as EnvironmentRegistry.EnvironmentRegistry["Service"]); - const runtime = Atom.runtime( - Layer.merge( - Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry), - Layer.succeed(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())), - ), - ); - const atoms = createVcsEnvironmentAtoms(runtime); - const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (value) => - Effect.sync(() => value.dispose()), - ); - const target = { environmentId: TARGET.environmentId, input: { cwd: "/repo" } }; - const first = yield* Stream.toPull( - AtomRegistry.toStreamResult(registry, atoms.branch(target)), - ); - expect(yield* first).toEqual([{ refName: "main" }]); - const second = yield* Stream.toPull( - AtomRegistry.toStreamResult(registry, atoms.branch(target)), - ); - expect(yield* second).toEqual([{ refName: "main" }]); - expect(subscriptions).toBe(1); - yield* SubscriptionRef.set(activeSession, Option.some(session(clientFor("external")))); - expect(yield* Queue.take(released)).toBe("main"); - expect(yield* first).toEqual([{ refName: "external" }]); - expect(yield* second).toEqual([{ refName: "external" }]); - registry.dispose(); - expect(yield* Queue.take(released)).toBe("external"); - }).pipe(Effect.scoped), - ); -}); - describe("cached VCS refs", () => { it("invalidates all ref streams in the mutated environment", () => { const registry = AtomRegistry.make(); From 29c4b0120a4b2be3ce361a89d2d619d2eb0e75dd Mon Sep 17 00:00:00 2001 From: Wout Stiens <71498452+StiensWout@users.noreply.github.com> Date: Sun, 8 Nov 2026 13:09:53 +0100 Subject: [PATCH 3/5] fix(vcs): serialize branch reads and negotiate server support Co-authored-by: GPT 6.1 Sol Co-authored-by: Claude Opus 5.5 Co-authored-by: Grok 4.7 --- apps/server/src/auth/RpcAuthorization.test.ts | 55 ++++++ .../src/environment/ServerEnvironment.ts | 1 + .../src/vcs/VcsStatusBroadcaster.test.ts | 162 ++++++++++++++++-- apps/server/src/vcs/VcsStatusBroadcaster.ts | 97 +++++++---- .../client-runtime/src/rpc/client.test.ts | 29 +++- packages/client-runtime/src/rpc/client.ts | 58 ++++++- packages/client-runtime/src/state/vcs.test.ts | 78 +++++++++ packages/client-runtime/src/state/vcs.ts | 11 +- packages/contracts/src/environment.ts | 1 + 9 files changed, 443 insertions(+), 49 deletions(-) diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index de94aca577021..b5cade0f84d46 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -139,6 +139,61 @@ describe("RPC scope middleware", () => { ); }); +it.effect.each([WS_METHODS.subscribeVcsBranch, WS_METHODS.subscribeVcsStatus] as const)( + "enforces read permission on stream %s", + (method) => + Effect.gen(function* () { + const group = WsRpcGroup.omit( + ...[...WsRpcGroup.requests.keys()].filter( + ( + tag, + ): tag is Exclude< + keyof typeof RPC_REQUIRED_SCOPES, + typeof WS_METHODS.subscribeVcsBranch | typeof WS_METHODS.subscribeVcsStatus + > => tag !== WS_METHODS.subscribeVcsBranch && tag !== WS_METHODS.subscribeVcsStatus, + ), + ); + let handled = 0; + const handlers = Layer.mergeAll( + group.toLayerHandler(WS_METHODS.subscribeVcsBranch, () => + Stream.fromEffect( + Effect.sync(() => { + handled++; + return { refName: "main" }; + }), + ), + ), + group.toLayerHandler(WS_METHODS.subscribeVcsStatus, () => + Stream.fromEffect( + Effect.sync(() => { + handled++; + }), + ).pipe(Stream.drain), + ), + ); + const denied = yield* RpcTest.makeClient(group).pipe( + Effect.provide(Layer.merge(handlers, RpcAuthorization.layer([]))), + ); + expect( + yield* Stream.runDrain( + method === WS_METHODS.subscribeVcsBranch + ? denied[method]({ cwd: "/repo" }).pipe(Stream.drain) + : denied[method]({ cwd: "/repo" }).pipe(Stream.drain), + ).pipe(Effect.flip), + ).toMatchObject({ requiredPermission: AuthOrchestrationReadScope }); + expect(handled).toBe(0); + const granted = yield* RpcTest.makeClient(group).pipe( + Effect.provide(Layer.merge(handlers, RpcAuthorization.layer([AuthOrchestrationReadScope]))), + ); + yield* Stream.runDrain( + method === WS_METHODS.subscribeVcsBranch + ? granted[method]({ cwd: "/repo" }).pipe(Stream.drain) + : granted[method]({ cwd: "/repo" }).pipe(Stream.drain), + ); + expect(handled).toBe(1); + }).pipe(Effect.scoped), +); + describe("settings mutation authorization", () => { const group = WsRpcGroup.omit( ...[...WsRpcGroup.requests.keys()].filter( diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 6d2c30db16130..ddce94a44e363 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -227,6 +227,7 @@ export const make = Effect.gen(function* () { pullRequests: true, pullRequestChecks: true, sideChats: true, + vcsBranchStatus: true, inlineMessageContext: true, requiredWorktreeBootstrap: true, threadSettlement: true, diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index a1ed249421b48..807500b954127 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -11,6 +11,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; import * as Scope from "effect/Scope"; import * as Scheduler from "effect/Scheduler"; import * as Stream from "effect/Stream"; @@ -78,6 +79,8 @@ function layerTestFor(state: { beforeRemoteStatus?: Effect.Effect; beforeLocalStatus?: Effect.Effect; beforeCurrentBranch?: Effect.Effect; + beforeInvalidate?: Effect.Effect; + branchRead?: () => Effect.Effect; fileSystem?: FileSystem.FileSystem; pullCurrentBranch?: GitWorkflowService.GitWorkflowService["Service"]["pullCurrentBranch"]; }) { @@ -104,9 +107,11 @@ function layerTestFor(state: { ); }), currentBranch: () => - (state.beforeCurrentBranch ?? Effect.void).pipe( - Effect.andThen(Effect.sync(() => state.currentLocalStatus.refName)), - ), + state.branchRead + ? state.branchRead() + : (state.beforeCurrentBranch ?? Effect.void).pipe( + Effect.andThen(Effect.sync(() => state.currentLocalStatus.refName)), + ), remoteStatus: (_input, options) => Effect.suspend(() => { return (state.beforeRemoteStatus ?? Effect.void).pipe( @@ -125,7 +130,7 @@ function layerTestFor(state: { invalidateLocalStatus: () => Effect.sync(() => { state.localInvalidationCalls += 1; - }), + }).pipe(Effect.andThen(state.beforeInvalidate ?? Effect.void)), invalidateRemoteStatus: () => Effect.sync(() => { state.remoteInvalidationCalls += 1; @@ -244,6 +249,13 @@ describe("VcsStatusBroadcaster", () => { assert.deepEqual(yield* details, [ { _tag: "snapshot", local: state.currentLocalStatus, remote: null }, ]); + const callsWithDetail = state.localStatusCalls; + const extraBranches = yield* Stream.toPull( + broadcaster.streamBranchStatus({ cwd: "/repo" }), + ); + yield* extraBranches; + yield* broadcaster.refreshBranchStatus({ cwd: "/repo" }); + assert.equal(state.localStatusCalls, callsWithDetail); state.currentLocalStatus = { ...baseLocalStatus, refName: "changed", @@ -264,7 +276,7 @@ describe("VcsStatusBroadcaster", () => { ...state.currentLocalStatus, workingTree: { files: [], insertions: 20, deletions: 4 }, }; - yield* broadcaster.refreshBranchStatus({ cwd: "/repo" }); + yield* broadcaster.refreshLocalStatus("/repo"); assert.deepEqual(yield* secondDetails, [ { _tag: "localUpdated", local: state.currentLocalStatus }, ]); @@ -317,18 +329,19 @@ describe("VcsStatusBroadcaster", () => { .refreshBranchStatus({ cwd: "/repo" }) .pipe(Effect.forkScoped); yield* Deferred.await(reading); - const details = yield* Stream.toPull( + const detailsFiber = yield* Stream.runHead( broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }), - ); - yield* details; + ).pipe(Effect.forkScoped); state.currentLocalStatus = { ...baseLocalStatus, workingTree: { files: [], insertions: 8, deletions: 2 }, }; yield* Deferred.succeed(resume, undefined); yield* Fiber.join(refresh); - assert.equal(state.localStatusCalls, 2); - assert.deepEqual(yield* details, [{ _tag: "localUpdated", local: state.currentLocalStatus }]); + assert.deepEqual( + yield* Fiber.join(detailsFiber), + Option.some({ _tag: "snapshot" as const, local: state.currentLocalStatus, remote: null }), + ); }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); }); @@ -1237,3 +1250,132 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(layerTest)); }); }); + +const newState = (): Parameters[0] => ({ + fileSystem: FileSystem.makeNoop({ realPath: (cwd) => Effect.succeed(cwd) }), + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: null, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, +}); +it.effect("detail join during invalidation must get fresh totals", () => { + const state = newState(); + return Effect.gen(function* () { + const b = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const oldScope = yield* Scope.make(); + const old = yield* Stream.toPull(b.streamStatus({ cwd: "/repo", includeRemote: false })).pipe( + Scope.provide(oldScope), + ); + yield* old; + yield* Scope.close(oldScope, Exit.void); + state.currentLocalStatus = { + ...baseLocalStatus, + workingTree: { files: [], insertions: 42, deletions: 0 }, + }; + const entered = yield* Deferred.make(); + const resume = yield* Deferred.make(); + state.beforeInvalidate = Deferred.succeed(entered, undefined).pipe( + Effect.andThen(Deferred.await(resume)), + ); + const refresh = yield* b.refreshBranchStatus({ cwd: "/repo" }).pipe(Effect.forkScoped); + yield* Deferred.await(entered); + const snapshotFiber = yield* Stream.runHead( + b.streamStatus({ cwd: "/repo", includeRemote: false }), + ).pipe(Effect.forkScoped({ startImmediately: true })); + yield* Deferred.succeed(resume, undefined); + yield* Fiber.join(refresh); + assert.deepEqual( + yield* Fiber.join(snapshotFiber), + Option.some({ _tag: "snapshot" as const, local: state.currentLocalStatus, remote: null }), + ); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); +}); +it.effect("serializes branch reads with detailed refresh publication", () => { + const state = newState(); + return Effect.gen(function* () { + const b = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const entered = yield* Deferred.make(); + const resume = yield* Deferred.make(); + state.branchRead = () => + Deferred.succeed(entered, undefined).pipe( + Effect.andThen(Deferred.await(resume)), + Effect.as("old"), + ); + const older = yield* b.refreshBranchStatus({ cwd: "/repo" }).pipe(Effect.forkScoped); + yield* Deferred.await(entered); + state.currentLocalStatus = { ...baseLocalStatus, refName: "new" }; + const newer = yield* b + .refreshLocalStatus("/repo") + .pipe(Effect.forkScoped({ startImmediately: true })); + yield* Deferred.succeed(resume, undefined); + yield* Fiber.join(older); + yield* Fiber.join(newer); + state.branchRead = () => Effect.succeed("new"); + const result = yield* b.getStatus({ cwd: "/repo" }); + assert.equal(result.refName, "new"); + assert.equal(state.localStatusCalls, 1); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); +}); + +it.effect("publishes overlapping branch reads in checkout order", () => { + const state = newState(); + return Effect.gen(function* () { + const b = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const branches = yield* Stream.toPull(b.streamBranchStatus({ cwd: "/repo" })); + yield* branches; + const entered = yield* Deferred.make(); + const resume = yield* Deferred.make(); + let reads = 0; + state.branchRead = () => + ++reads === 1 + ? Deferred.succeed(entered, undefined).pipe( + Effect.andThen(Deferred.await(resume)), + Effect.as("old"), + ) + : Effect.succeed("new"); + const older = yield* b.refreshBranchStatus({ cwd: "/repo" }).pipe(Effect.forkScoped); + yield* Deferred.await(entered); + const newer = yield* b + .refreshBranchStatus({ cwd: "/repo" }) + .pipe(Effect.forkScoped({ startImmediately: true })); + yield* Deferred.succeed(resume, undefined); + yield* Fiber.join(older); + yield* Fiber.join(newer); + assert.deepEqual(yield* branches, [{ refName: "old" }, { refName: "new" }]); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); +}); + +it.effect("does not replay older queued branches after a new subscriber snapshot", () => { + const state = newState(); + return Effect.gen(function* () { + const b = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const entered = yield* Deferred.make(); + const resume = yield* Deferred.make(); + const values = yield* Queue.unbounded<{ readonly refName: string | null }>(); + let reads = 0; + state.branchRead = () => + ++reads === 1 + ? Deferred.succeed(entered, undefined).pipe( + Effect.andThen(Deferred.await(resume)), + Effect.as("old"), + ) + : Effect.succeed("new"); + const older = yield* b.refreshBranchStatus({ cwd: "/repo" }).pipe(Effect.forkScoped); + yield* Deferred.await(entered); + const subscriber = yield* b.streamBranchStatus({ cwd: "/repo" }).pipe( + Stream.tap((value) => Queue.offer(values, value)), + Stream.take(2), + Stream.runDrain, + Effect.forkScoped({ startImmediately: true }), + ); + yield* Deferred.succeed(resume, undefined); + yield* Fiber.join(older); + assert.deepEqual(yield* Queue.take(values), { refName: "new" }); + state.branchRead = () => Effect.succeed("latest"); + yield* b.refreshBranchStatus({ cwd: "/repo" }); + assert.deepEqual(yield* Queue.take(values), { refName: "latest" }); + yield* Fiber.join(subscriber); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); +}); diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 9b012737762c6..c59468826fb29 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -257,22 +257,32 @@ export const make = Effect.gen(function* () { (pubsub) => PubSub.shutdown(pubsub), ); const branchChangesPubSub = yield* Effect.acquireRelease( - PubSub.unbounded<{ readonly cwd: string; readonly refName: string | null }>(), + PubSub.unbounded<{ + readonly cwd: string; + readonly refName: string | null; + readonly revision: number; + }>(), (pubsub) => PubSub.shutdown(pubsub), ); const broadcasterScope = yield* Effect.acquireRelease(Scope.make(), (scope) => Scope.close(scope, Exit.void), ); const detailedSubscribers = new Map(); - const branches = new Map(); + const branches = new Map< + string, + { readonly refName: string | null; readonly revision: number } + >(); + let branchRevision = 0; const publishBranch = Effect.fn("VcsStatusBroadcaster.publishBranch")(function* ( cwd: string, refName: string | null, ) { - if (branches.get(cwd) === refName) return; - branches.set(cwd, refName); - yield* PubSub.publish(branchChangesPubSub, { cwd, refName }); + if (branches.get(cwd)?.refName === refName) return; + const revision = ++branchRevision; + branches.set(cwd, { refName, revision }); + yield* PubSub.publish(branchChangesPubSub, { cwd, refName, revision }); }); + const localRefreshLocks = yield* KeyedLock.make(); const cacheRef = yield* Ref.make(new Map()); // One permit per cwd for remote reads that write the cache. Without it a // periodic poll that started before an agent creates a PR can finish after the @@ -343,24 +353,30 @@ export const make = Effect.gen(function* () { return yield* updateCachedLocalStatus(cwd, local); }); - const getOrLoadLocalStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")(function* ( - cwd: string, - ) { - const cached = yield* getCachedStatus(cwd); - if (cached?.local) { - const branch = yield* workflow.currentBranch(cwd); - if (branch === cached.local.value.refName) return cached.local.value; - return yield* refreshLocalStatusCore(cwd); - } - return yield* loadLocalStatus(cwd); - }); + const getOrLoadLocalStatusUnlocked = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")( + function* (cwd: string) { + const cached = yield* getCachedStatus(cwd); + if (cached?.local) { + const branch = yield* workflow.currentBranch(cwd); + if (branch === cached.local.value.refName) return cached.local.value; + yield* workflow.invalidateLocalStatus(cwd); + return yield* updateCachedLocalStatus(cwd, yield* workflow.localStatus({ cwd }), { + publish: true, + }); + } + return yield* loadLocalStatus(cwd); + }, + ); + + const getOrLoadLocalStatus = (cwd: string) => + localRefreshLocks.withLock(cwd, getOrLoadLocalStatusUnlocked(cwd)); const getStatusParts = Effect.fn("VcsStatusBroadcaster.getStatusParts")(function* (cwd: string) { const cached = yield* getCachedStatus(cwd); if (cached?.local) { return { local: cached.local.value, remote: cached.remote?.value ?? null }; } - return { local: yield* loadLocalStatus(cwd), remote: null }; + return { local: yield* getOrLoadLocalStatus(cwd), remote: null }; }); const withFileSystem = Effect.provideService(FileSystem.FileSystem, fs); @@ -389,7 +405,6 @@ export const make = Effect.gen(function* () { // A refresh that started after a request saw every change that request could mean, so queued // callers share it instead of each running git status again. - const localRefreshLocks = yield* KeyedLock.make(); const localRefreshRequests = new Map(); const localRefreshResults = new Map< string, @@ -791,32 +806,42 @@ export const make = Effect.gen(function* () { }), ); - // Without detailed subscribers only HEAD is read; otherwise the full local refresh runs and - // publishes the branch with it. Demand is rechecked so a joining subscriber gets fresh totals. + // Serialize branch publication and cache invalidation with detailed reads and refreshes. const refreshBranchStatusCore = Effect.fn("VcsStatusBroadcaster.refreshBranchStatusCore")( - function* (cwd: string) { - if (!detailedSubscribers.has(cwd)) { - const refName = yield* workflow.currentBranch(cwd); - if (!detailedSubscribers.has(cwd)) { - // Branch reads cannot certify cached Changes totals, even if HEAD did not move. + (cwd: string) => + localRefreshLocks.withLock( + cwd, + Effect.gen(function* () { + if (detailedSubscribers.has(cwd)) { + const local = yield* getOrLoadLocalStatusUnlocked(cwd); + return { refName: local.refName, revision: branches.get(cwd)?.revision ?? 0 }; + } + const refName = yield* workflow.currentBranch(cwd); + // HEAD cannot certify working-tree totals, even when it has not moved. + localRefreshResults.delete(cwd); yield* workflow.invalidateLocalStatus(cwd); yield* Ref.update(cacheRef, (cache) => { const previous = cache.get(cwd); return previous ? new Map(cache).set(cwd, { ...previous, local: null }) : cache; }); + if (detailedSubscribers.has(cwd)) { + const local = yield* workflow.localStatus({ cwd }); + yield* updateCachedLocalStatus(cwd, local, { publish: true }); + return { refName: local.refName, revision: branches.get(cwd)?.revision ?? 0 }; + } yield* publishBranch(cwd, refName); - return { refName }; - } - } - const local = yield* refreshLocalStatusCore(cwd); - return { refName: local.refName }; - }, + return { refName, revision: branches.get(cwd)?.revision ?? 0 }; + }), + ), ); const refreshBranchStatus: VcsStatusBroadcaster["Service"]["refreshBranchStatus"] = Effect.fn( "VcsStatusBroadcaster.refreshBranchStatus", )(function* (input) { - return yield* refreshBranchStatusCore(yield* withFileSystem(normalizeCwd(input.cwd))); + const { refName } = yield* refreshBranchStatusCore( + yield* withFileSystem(normalizeCwd(input.cwd)), + ); + return { refName }; }); const streamBranchStatus: VcsStatusBroadcaster["Service"]["streamBranchStatus"] = (input) => @@ -826,9 +851,9 @@ export const make = Effect.gen(function* () { const subscription = yield* PubSub.subscribe(branchChangesPubSub); const branch = yield* refreshBranchStatusCore(cwd); return Stream.concat( - Stream.make(branch), + Stream.make({ refName: branch.refName }), Stream.fromSubscription(subscription).pipe( - Stream.filter((event) => event.cwd === cwd), + Stream.filter((event) => event.cwd === cwd && event.revision > branch.revision), Stream.map(({ refName }) => ({ refName })), ), ).pipe(Stream.changesWith((left, right) => left.refName === right.refName)); @@ -838,7 +863,9 @@ export const make = Effect.gen(function* () { return VcsStatusBroadcaster.of({ streamBranchStatus, refreshBranchStatus, - streamBranchChanges: Stream.fromPubSub(branchChangesPubSub), + streamBranchChanges: Stream.fromPubSub(branchChangesPubSub).pipe( + Stream.map(({ cwd, refName }) => ({ cwd, refName })), + ), getStatus, refreshLocalStatus, refreshStatus, diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index e9779c4f86606..17def752a1568 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -1,6 +1,7 @@ import { EnvironmentId, type RelayClientInstallProgressEvent, + type ServerConfig, type ServerLifecycleStreamEvent, WS_METHODS, } from "@t3tools/contracts"; @@ -26,7 +27,7 @@ import { import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as RpcSession from "../rpc/session.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; -import { runStream, subscribe, subscribeDynamicWithSession } from "./client.ts"; +import { runStream, subscribe, subscribeDynamicWithSession, refreshVcsBranch } from "./client.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), @@ -414,3 +415,29 @@ describe("environment RPC", () => { }), ); }); + +it.effect.each([true, false])( + "refreshes branches using the connected server capability (%s)", + (supported) => + Effect.gen(function* () { + const { activeSession, supervisor } = yield* makeHarness(); + const client = { + [supported ? WS_METHODS.vcsRefreshBranch : WS_METHODS.vcsRefreshStatus]: () => + Effect.succeed({ refName: "external" }), + } as unknown as WsRpcProtocolClient; + yield* SubscriptionRef.set( + activeSession, + Option.some({ + ...session(client), + initialConfig: Effect.succeed({ + environment: { capabilities: supported ? { vcsBranchStatus: true } : {} }, + } as ServerConfig), + }), + ); + expect( + yield* refreshVcsBranch({ cwd: "/repo" }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + ), + ).toEqual({ refName: "external" }); + }).pipe(Effect.scoped), +); diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 21c4c146ac25b..78d950fb7279a 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -301,7 +301,10 @@ function subscribeDynamicMapped( method: tag, input, }); - const stream = mapStream(session, method(input)).pipe( + const stream = mapStream( + session, + Stream.suspend(() => method(input)), + ).pipe( Stream.onFirst(() => Effect.sync(() => { expectedFailureRetries = 0; @@ -437,6 +440,59 @@ export function subscribe( return subscribeDynamic(tag, () => Effect.succeed(input), options); } +export const subscribeVcsBranch = ( + input: EnvironmentRpcInput, +) => + subscribeDynamicMapped( + WS_METHODS.subscribeVcsBranch, + () => Effect.succeed(input), + (session, stream) => + Stream.unwrap( + session.initialConfig.pipe( + Effect.mapError( + (error) => + new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: error.message, + cause: error, + }), + }), + ), + Effect.map((config) => + config.environment.capabilities.vcsBranchStatus === true + ? stream + : session.client[WS_METHODS.subscribeVcsStatus]({ + ...input, + includeRemote: false, + }).pipe( + Stream.filter((event) => event._tag !== "remoteUpdated"), + Stream.map((event) => ({ refName: event.local.refName })), + ), + ), + ), + ), + ); + +export const refreshVcsBranch = Effect.fn("EnvironmentRpc.refreshVcsBranch")(function* ( + input: EnvironmentRpcInput, +) { + const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; + const config = yield* getInitialServerConfig().pipe( + Effect.mapError( + (error) => + new EnvironmentRpcUnavailableError({ + environmentId: supervisor.target.environmentId, + message: error.message, + }), + ), + ); + if (config.environment.capabilities.vcsBranchStatus === true) { + return yield* request(WS_METHODS.vcsRefreshBranch, input); + } + const status = yield* request(WS_METHODS.vcsRefreshStatus, input); + return { refName: status.refName }; +}); + export const request = >( tag: TTag, input: EnvironmentRpcInput, diff --git a/packages/client-runtime/src/state/vcs.test.ts b/packages/client-runtime/src/state/vcs.test.ts index cdbda23422237..7327c8cc203ee 100644 --- a/packages/client-runtime/src/state/vcs.test.ts +++ b/packages/client-runtime/src/state/vcs.test.ts @@ -7,6 +7,7 @@ import { type VcsListRefsInput, type VcsListRefsResult, type VcsStatusResult, + type ServerConfig, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Deferred from "effect/Deferred"; @@ -124,6 +125,83 @@ function cacheWithRefs( }); } +describe("branch subscriptions", () => { + it.effect.each([true, false])( + "shares and releases branch streams across session replacement (capability %s)", + (supported) => + Effect.gen(function* () { + const released = yield* Queue.unbounded(); + let subscriptions = 0; + const clientFor = (refName: string, branchSupported: boolean) => + ({ + [branchSupported ? WS_METHODS.subscribeVcsBranch : WS_METHODS.subscribeVcsStatus]: + () => { + subscriptions += 1; + return Stream.concat( + Stream.make( + branchSupported + ? { refName } + : { _tag: "snapshot", local: status({ refName }), remote: null }, + ), + Stream.never, + ).pipe(Stream.ensuring(Queue.offer(released, refName))); + }, + }) as unknown as WsRpcProtocolClient; + const branchSession = (refName: string, branchSupported: boolean): RpcSession => ({ + ...session(clientFor(refName, branchSupported)), + initialConfig: Effect.succeed({ + environment: { capabilities: branchSupported ? { vcsBranchStatus: true } : {} }, + } as ServerConfig), + }); + const activeSession = yield* SubscriptionRef.make( + Option.some(branchSession("main", supported)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(CONNECTED_CONNECTION_STATE), + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + }); + const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({ + followStream: (_environmentId, stream) => + Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + } as EnvironmentRegistry.EnvironmentRegistry["Service"]); + const runtime = Atom.runtime( + Layer.merge( + Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry), + Layer.succeed(Persistence.EnvironmentCacheStore, cacheWithRefs(Option.none())), + ), + ); + const atoms = createVcsEnvironmentAtoms(runtime); + const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (value) => + Effect.sync(() => value.dispose()), + ); + const target = { environmentId: TARGET.environmentId, input: { cwd: "/repo" } }; + const first = yield* Stream.toPull( + AtomRegistry.toStreamResult(registry, atoms.branch(target)), + ); + expect(yield* first).toEqual([{ refName: "main" }]); + const second = yield* Stream.toPull( + AtomRegistry.toStreamResult(registry, atoms.branch(target)), + ); + expect(yield* second).toEqual([{ refName: "main" }]); + expect(subscriptions).toBe(1); + yield* SubscriptionRef.set( + activeSession, + Option.some(branchSession("external", !supported)), + ); + expect(yield* Queue.take(released)).toBe("main"); + expect(yield* first).toEqual([{ refName: "external" }]); + expect(yield* second).toEqual([{ refName: "external" }]); + registry.dispose(); + expect(yield* Queue.take(released)).toBe("external"); + }).pipe(Effect.scoped), + ); +}); + describe("cached VCS refs", () => { it("invalidates all ref streams in the mutated environment", () => { const registry = AtomRegistry.make(); diff --git a/packages/client-runtime/src/state/vcs.ts b/packages/client-runtime/src/state/vcs.ts index a5679c6d59d14..12d55a7477058 100644 --- a/packages/client-runtime/src/state/vcs.ts +++ b/packages/client-runtime/src/state/vcs.ts @@ -24,7 +24,13 @@ import type { EnvironmentRegistry } from "../connection/registry.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import * as Persistence from "../platform/persistence.ts"; -import { request, subscribe, type EnvironmentRpcInput } from "../rpc/client.ts"; +import { + request, + subscribe, + subscribeVcsBranch, + refreshVcsBranch, + type EnvironmentRpcInput, +} from "../rpc/client.ts"; import { followStreamInEnvironment } from "./runtime.ts"; import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts"; import { @@ -298,11 +304,12 @@ export function createVcsEnvironmentAtoms( label: "environment-data:vcs:branch", idleTtlMs: VCS_STATUS_IDLE_TTL_MS, subscribe: (input: EnvironmentRpcInput) => - subscribe(WS_METHODS.subscribeVcsBranch, input), + subscribeVcsBranch(input), }), refreshBranch: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:refresh-branch", tag: WS_METHODS.vcsRefreshBranch, + execute: refreshVcsBranch, scheduler: vcsCommandScheduler, concurrency: vcsCommandConcurrency, }), diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 96b4537478ecc..cd5e6833468e8 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -88,6 +88,7 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ pullRequestChecks: Schema.optionalKey(Schema.Boolean), /** Side chats: hidden `lineage.sideChat` forks. Older servers ignore the fork flag. */ sideChats: Schema.optionalKey(Schema.Boolean), + vcsBranchStatus: Schema.optionalKey(Schema.Boolean), /** Every protocol-2 server sends this; older mobile builds still read it. */ inlineMessageContext: Schema.optionalKey(Schema.Boolean), requiredWorktreeBootstrap: Schema.optionalKey(Schema.Boolean), From aed66c959f90081f4bb85e59cd9ede91b70e64fa Mon Sep 17 00:00:00 2001 From: Wout Stiens <71498452+StiensWout@users.noreply.github.com> Date: Sun, 8 Nov 2026 13:27:50 +0100 Subject: [PATCH 4/5] fix(vcs): preserve refresh behavior across branch-only reads Co-authored-by: GPT 6.1 Sol Co-authored-by: Claude Opus 5.5 Co-authored-by: Grok 4.7 --- .../src/vcs/VcsStatusBroadcaster.test.ts | 1 + apps/server/src/vcs/VcsStatusBroadcaster.ts | 2 +- .../client-runtime/src/rpc/client.test.ts | 43 ++++++++++++++++++- packages/client-runtime/src/rpc/client.ts | 23 +++++++--- 4 files changed, 60 insertions(+), 9 deletions(-) diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 807500b954127..a9df07044943a 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -557,6 +557,7 @@ describe("VcsStatusBroadcaster", () => { state.backgroundWorkEnabled = false; state.currentRemoteStatus = { ...remoteStatusWithPr, aheadCount: 7 }; + yield* broadcaster.refreshBranchStatus({ cwd: "/repo" }); const refreshed = yield* broadcaster.refreshPullRequestStatus("/repo"); assert.deepStrictEqual(refreshed, state.currentRemoteStatus); const merged = yield* broadcaster.getStatus({ cwd: "/repo" }); diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index c59468826fb29..be1c86940e64e 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -582,7 +582,7 @@ export const make = Effect.gen(function* () { cwd, Effect.gen(function* () { const cached = yield* getCachedStatus(cwd); - if (!cached?.local || cached.remote?.value === null) return null; + if (!cached || cached.remote?.value === null) return null; // Turn completion must update divergence even when background polling is paused. const remoteRead = yield* readRemoteStatus(cwd, { refreshUpstream: false, diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index 17def752a1568..f704730a89024 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -421,9 +421,14 @@ it.effect.each([true, false])( (supported) => Effect.gen(function* () { const { activeSession, supervisor } = yield* makeHarness(); + const inputs: unknown[] = []; const client = { - [supported ? WS_METHODS.vcsRefreshBranch : WS_METHODS.vcsRefreshStatus]: () => - Effect.succeed({ refName: "external" }), + [supported ? WS_METHODS.vcsRefreshBranch : WS_METHODS.vcsRefreshStatus]: ( + input: unknown, + ) => { + inputs.push(input); + return Effect.succeed({ refName: "external" }); + }, } as unknown as WsRpcProtocolClient; yield* SubscriptionRef.set( activeSession, @@ -439,5 +444,39 @@ it.effect.each([true, false])( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), ), ).toEqual({ refName: "external" }); + expect(inputs).toEqual([supported ? { cwd: "/repo" } : { cwd: "/repo", localOnly: true }]); }).pipe(Effect.scoped), ); + +it.effect("binds branch capability selection and execution to one session", () => + Effect.gen(function* () { + const { activeSession, supervisor } = yield* makeHarness(); + const configRequested = yield* Deferred.make(); + const config = yield* Deferred.make(); + const original = { + [WS_METHODS.vcsRefreshBranch]: () => Effect.succeed({ refName: "original" }), + } as unknown as WsRpcProtocolClient; + const replacement = { + [WS_METHODS.vcsRefreshStatus]: () => Effect.succeed({ refName: "replacement" }), + } as unknown as WsRpcProtocolClient; + yield* SubscriptionRef.set( + activeSession, + Option.some({ + ...session(original), + initialConfig: Deferred.succeed(configRequested, undefined).pipe( + Effect.andThen(Deferred.await(config)), + ), + }), + ); + const refresh = yield* refreshVcsBranch({ cwd: "/repo" }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.forkChild, + ); + yield* Deferred.await(configRequested); + yield* SubscriptionRef.set(activeSession, Option.some(session(replacement))); + yield* Deferred.succeed(config, { + environment: { capabilities: { vcsBranchStatus: true } }, + } as ServerConfig); + expect(yield* Fiber.join(refresh)).toEqual({ refName: "original" }); + }).pipe(Effect.scoped), +); diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 78d950fb7279a..82412c98e3b34 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -187,15 +187,14 @@ const authorizeRequest = Effect.fn("EnvironmentRpc.authorize")(function* ( yield* guard.authorize(supervisor.target.environmentId, method, input); }); -export const requestGuarded = Effect.fn("EnvironmentRpc.request")(function* < +const requestInSession = Effect.fn("EnvironmentRpc.request")(function* < TTag extends EnvironmentUnaryRpcTag, ->(tag: TTag, input: EnvironmentRpcInput) { +>(session: RpcSession, tag: TTag, input: EnvironmentRpcInput) { const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; yield* Effect.annotateCurrentSpan({ "environment.id": supervisor.target.environmentId, "rpc.method": tag, }); - const session = yield* currentSession(); yield* authorizeRequest(tag, input); const observer = yield* EnvironmentRpcRequestObserver; const method = session.client[tag] as ( @@ -209,6 +208,14 @@ export const requestGuarded = Effect.fn("EnvironmentRpc.request")(function* < return yield* method(input).pipe(Effect.ensuring(completeObservation)); }); +// The session-bound request owns the existing EnvironmentRpc.request span. +export const requestGuarded = Effect.fnUntraced(function* ( + tag: TTag, + input: EnvironmentRpcInput, +) { + return yield* requestInSession(yield* currentSession(), tag, input); +}); + export function runStreamGuarded( tag: TTag, input: EnvironmentRpcInput, @@ -477,7 +484,8 @@ export const refreshVcsBranch = Effect.fn("EnvironmentRpc.refreshVcsBranch")(fun input: EnvironmentRpcInput, ) { const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; - const config = yield* getInitialServerConfig().pipe( + const session = yield* currentSession(); + const config = yield* session.initialConfig.pipe( Effect.mapError( (error) => new EnvironmentRpcUnavailableError({ @@ -487,9 +495,12 @@ export const refreshVcsBranch = Effect.fn("EnvironmentRpc.refreshVcsBranch")(fun ), ); if (config.environment.capabilities.vcsBranchStatus === true) { - return yield* request(WS_METHODS.vcsRefreshBranch, input); + return yield* requestInSession(session, WS_METHODS.vcsRefreshBranch, input); } - const status = yield* request(WS_METHODS.vcsRefreshStatus, input); + const status = yield* requestInSession(session, WS_METHODS.vcsRefreshStatus, { + ...input, + localOnly: true, + }); return { refName: status.refName }; }); From 9212ddc9336048b143830082cdd0ac23a4e9f8ac Mon Sep 17 00:00:00 2001 From: Wout Stiens <71498452+StiensWout@users.noreply.github.com> Date: Sun, 8 Nov 2026 13:40:06 +0100 Subject: [PATCH 5/5] perf(client): keep sidebar wakes lightweight on older hosts Co-authored-by: GPT 6.1 Sol Co-authored-by: Claude Opus 5.5 Co-authored-by: Grok 4.7 --- apps/server/src/ws.ts | 3 ++- packages/client-runtime/src/rpc/client.test.ts | 13 ++++++++----- packages/client-runtime/src/rpc/client.ts | 6 +----- packages/contracts/src/rpc.ts | 2 +- 4 files changed, 12 insertions(+), 12 deletions(-) diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 4249d5e096dfc..8d272dd584eeb 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -3113,7 +3113,8 @@ export const layerWsRpc = ( [WS_METHODS.assetsPersistChatAttachments]: (input) => persistChatAttachments(input).pipe(Effect.map((attachments) => ({ attachments }))), [WS_METHODS.subscribeVcsBranch]: (input) => vcsStatusBroadcaster.streamBranchStatus(input), - [WS_METHODS.vcsRefreshBranch]: (input) => vcsStatusBroadcaster.refreshBranchStatus(input), + [WS_METHODS.vcsRefreshBranch]: (input) => + vcsStatusBroadcaster.refreshBranchStatus(input).pipe(Effect.asVoid), [WS_METHODS.subscribeVcsStatus]: (input) => vcsStatusBroadcaster.streamStatus(input), [WS_METHODS.subscribeWorktreeSetup]: (input) => worktreeSetupTracker.stream(input.threadId), [WS_METHODS.worktreeSetupCancel]: (input) => diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index f704730a89024..9207ff1aafd58 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -427,7 +427,7 @@ it.effect.each([true, false])( input: unknown, ) => { inputs.push(input); - return Effect.succeed({ refName: "external" }); + return Effect.void; }, } as unknown as WsRpcProtocolClient; yield* SubscriptionRef.set( @@ -443,8 +443,8 @@ it.effect.each([true, false])( yield* refreshVcsBranch({ cwd: "/repo" }).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), ), - ).toEqual({ refName: "external" }); - expect(inputs).toEqual([supported ? { cwd: "/repo" } : { cwd: "/repo", localOnly: true }]); + ).toBeUndefined(); + expect(inputs).toEqual(supported ? [{ cwd: "/repo" }] : []); }).pipe(Effect.scoped), ); @@ -453,8 +453,10 @@ it.effect("binds branch capability selection and execution to one session", () = const { activeSession, supervisor } = yield* makeHarness(); const configRequested = yield* Deferred.make(); const config = yield* Deferred.make(); + const executed = yield* Deferred.make(); const original = { - [WS_METHODS.vcsRefreshBranch]: () => Effect.succeed({ refName: "original" }), + [WS_METHODS.vcsRefreshBranch]: () => + Deferred.succeed(executed, undefined).pipe(Effect.asVoid), } as unknown as WsRpcProtocolClient; const replacement = { [WS_METHODS.vcsRefreshStatus]: () => Effect.succeed({ refName: "replacement" }), @@ -477,6 +479,7 @@ it.effect("binds branch capability selection and execution to one session", () = yield* Deferred.succeed(config, { environment: { capabilities: { vcsBranchStatus: true } }, } as ServerConfig); - expect(yield* Fiber.join(refresh)).toEqual({ refName: "original" }); + expect(yield* Fiber.join(refresh)).toBeUndefined(); + yield* Deferred.await(executed); }).pipe(Effect.scoped), ); diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 82412c98e3b34..d3a49c2aa58b3 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -497,11 +497,7 @@ export const refreshVcsBranch = Effect.fn("EnvironmentRpc.refreshVcsBranch")(fun if (config.environment.capabilities.vcsBranchStatus === true) { return yield* requestInSession(session, WS_METHODS.vcsRefreshBranch, input); } - const status = yield* requestInSession(session, WS_METHODS.vcsRefreshStatus, { - ...input, - localOnly: true, - }); - return { refName: status.refName }; + // Older hosts retain their existing status stream without adding per-row wake scans. }); export const request = >( diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index e401903c1e84e..088a57cc7882d 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -1466,7 +1466,7 @@ const WsSubscribeVcsBranchRpc = Rpc.make(WS_METHODS.subscribeVcsBranch, { const WsVcsRefreshBranchRpc = Rpc.make(WS_METHODS.vcsRefreshBranch, { payload: VcsStatusInput, - success: VcsBranchStatus, + success: Schema.Void, error: Schema.Union([GitManagerServiceError, EnvironmentAuthorizationError]), });