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..b5cade0f84d46 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( @@ -133,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/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/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/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..a9df07044943a 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"; @@ -77,6 +78,9 @@ function layerTestFor(state: { automaticFetchInterval?: Effect.Effect; beforeRemoteStatus?: Effect.Effect; beforeLocalStatus?: Effect.Effect; + beforeCurrentBranch?: Effect.Effect; + beforeInvalidate?: Effect.Effect; + branchRead?: () => Effect.Effect; fileSystem?: FileSystem.FileSystem; pullCurrentBranch?: GitWorkflowService.GitWorkflowService["Service"]["pullCurrentBranch"]; }) { @@ -102,7 +106,12 @@ function layerTestFor(state: { Effect.as(state.currentLocalStatus), ); }), - currentBranch: () => Effect.sync(() => state.currentLocalStatus.refName), + currentBranch: () => + 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( @@ -121,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; @@ -207,6 +216,135 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); }); + it.effect( + "streams fresh branch-only reads, detached HEAD, non-repositories, 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 }, + ]); + 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", + 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.refreshLocalStatus("/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, + }; + 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 detailsFiber = yield* Stream.runHead( + broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }), + ).pipe(Effect.forkScoped); + state.currentLocalStatus = { + ...baseLocalStatus, + workingTree: { files: [], insertions: 8, deletions: 2 }, + }; + yield* Deferred.succeed(resume, undefined); + yield* Fiber.join(refresh); + assert.deepEqual( + yield* Fiber.join(detailsFiber), + Option.some({ _tag: "snapshot" as const, local: state.currentLocalStatus, remote: null }), + ); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }); + it.effect("publishes terminal branch switches and validates cached status reads", () => { const state: Parameters[0] = { currentLocalStatus: baseLocalStatus, @@ -419,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" }); @@ -1112,3 +1251,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 3921077ebbb42..be1c86940e64e 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; @@ -250,12 +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< + 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 === 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 @@ -274,21 +301,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 && local.isRepo) { - yield* PubSub.publish(branchChangesPubSub, { cwd, refName: local.refName }); - } + yield* publishBranch(cwd, local.refName); if (options?.publish && remoteCleared) { yield* PubSub.publish(changesPubSub, { cwd, @@ -332,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); @@ -378,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, @@ -556,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, @@ -738,6 +764,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,8 +806,66 @@ export const make = Effect.gen(function* () { }), ); + // Serialize branch publication and cache invalidation with detailed reads and refreshes. + const refreshBranchStatusCore = Effect.fn("VcsStatusBroadcaster.refreshBranchStatusCore")( + (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, revision: branches.get(cwd)?.revision ?? 0 }; + }), + ), + ); + + const refreshBranchStatus: VcsStatusBroadcaster["Service"]["refreshBranchStatus"] = Effect.fn( + "VcsStatusBroadcaster.refreshBranchStatus", + )(function* (input) { + const { refName } = yield* refreshBranchStatusCore( + yield* withFileSystem(normalizeCwd(input.cwd)), + ); + return { 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* refreshBranchStatusCore(cwd); + return Stream.concat( + Stream.make({ refName: branch.refName }), + Stream.fromSubscription(subscription).pipe( + Stream.filter((event) => event.cwd === cwd && event.revision > branch.revision), + Stream.map(({ refName }) => ({ refName })), + ), + ).pipe(Stream.changesWith((left, right) => left.refName === right.refName)); + }), + ); + return VcsStatusBroadcaster.of({ - streamBranchChanges: Stream.fromPubSub(branchChangesPubSub), + streamBranchStatus, + refreshBranchStatus, + streamBranchChanges: Stream.fromPubSub(branchChangesPubSub).pipe( + Stream.map(({ cwd, refName }) => ({ cwd, refName })), + ), getStatus, refreshLocalStatus, refreshStatus, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 1fa81ddc7026e..8d272dd584eeb 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -3112,6 +3112,9 @@ 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).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/apps/web/src/components/sidebar/SidebarThreadRow.tsx b/apps/web/src/components/sidebar/SidebarThreadRow.tsx index 41cb733daff15..39cf531b2a3c3 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"; @@ -687,14 +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.status({ - environmentId: thread.environmentId, - input: { cwd: gitCwd, includeRemote: false }, - }) - : 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, @@ -1640,14 +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.status({ - environmentId: thread.environmentId, - input: { cwd: gitCwd, includeRemote: false }, - }) - : 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/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.test.ts b/packages/client-runtime/src/rpc/client.test.ts index e9779c4f86606..9207ff1aafd58 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,71 @@ 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 inputs: unknown[] = []; + const client = { + [supported ? WS_METHODS.vcsRefreshBranch : WS_METHODS.vcsRefreshStatus]: ( + input: unknown, + ) => { + inputs.push(input); + return Effect.void; + }, + } 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), + ), + ).toBeUndefined(); + expect(inputs).toEqual(supported ? [{ cwd: "/repo" }] : []); + }).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 executed = yield* Deferred.make(); + const original = { + [WS_METHODS.vcsRefreshBranch]: () => + Deferred.succeed(executed, undefined).pipe(Effect.asVoid), + } 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)).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 e74abde6a2c59..d3a49c2aa58b3 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 @@ -186,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 ( @@ -208,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, @@ -300,7 +308,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; @@ -436,6 +447,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 session = yield* currentSession(); + const config = yield* session.initialConfig.pipe( + Effect.mapError( + (error) => + new EnvironmentRpcUnavailableError({ + environmentId: supervisor.target.environmentId, + message: error.message, + }), + ), + ); + if (config.environment.capabilities.vcsBranchStatus === true) { + return yield* requestInSession(session, WS_METHODS.vcsRefreshBranch, input); + } + // Older hosts retain their existing status stream without adding per-row wake scans. +}); + 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 ef508f2126c7f..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 { @@ -294,6 +300,19 @@ 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) => + subscribeVcsBranch(input), + }), + refreshBranch: createEnvironmentRpcCommand(runtime, { + label: "environment-data:vcs:refresh-branch", + tag: WS_METHODS.vcsRefreshBranch, + execute: refreshVcsBranch, + scheduler: vcsCommandScheduler, + concurrency: vcsCommandConcurrency, + }), status: createEnvironmentSubscriptionAtomFamily(runtime, { label: "environment-data:vcs:status", idleTtlMs: VCS_STATUS_IDLE_TTL_MS, 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), 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..088a57cc7882d 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: Schema.Void, + 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,