diff --git a/apps/server/src/git/GitManager.test.ts b/apps/server/src/git/GitManager.test.ts index 5d95ea5f62f9..3efe6d94ba93 100644 --- a/apps/server/src/git/GitManager.test.ts +++ b/apps/server/src/git/GitManager.test.ts @@ -949,6 +949,58 @@ it.layer(GitManagerTestLayer)("GitManager", (it) => { }), ); + it.effect("local status coherence token tracks HEAD, primary remote, and upstream identity", () => + Effect.gen(function* () { + const repoDir = yield* makeTempDir("t3code-git-manager-coherence-"); + yield* initRepo(repoDir); + const originDir = yield* createBareRemote(); + const replacementOriginDir = yield* createBareRemote(); + yield* runGit(repoDir, ["remote", "add", "origin", originDir]); + yield* runGit(repoDir, ["push", "-u", "origin", "main"]); + const { manager } = yield* makeManager(); + + const initial = yield* manager.localStatus({ cwd: repoDir }); + expect(initial.coherenceToken).toBeTruthy(); + expect(initial.remoteAssociationToken).toBeTruthy(); + + const fs = yield* FileSystem.FileSystem; + yield* fs.writeFileString(NodePath.join(repoDir, "WORKTREE-ONLY.md"), "uncommitted\n"); + yield* manager.invalidateLocalStatus(repoDir); + const afterWorkingTreeEdit = yield* manager.localStatus({ cwd: repoDir }); + expect(afterWorkingTreeEdit.coherenceToken).toBe(initial.coherenceToken); + expect(afterWorkingTreeEdit.remoteAssociationToken).toBe(initial.remoteAssociationToken); + + yield* fs.writeFileString(NodePath.join(repoDir, "HEAD-MOVE.md"), "next\n"); + yield* runGit(repoDir, ["add", "HEAD-MOVE.md", "WORKTREE-ONLY.md"]); + yield* runGit(repoDir, ["commit", "-m", "Move head"]); + yield* manager.invalidateLocalStatus(repoDir); + const afterHeadMove = yield* manager.localStatus({ cwd: repoDir }); + expect(afterHeadMove.coherenceToken).not.toBe(initial.coherenceToken); + expect(afterHeadMove.remoteAssociationToken).toBe(initial.remoteAssociationToken); + + yield* runGit(repoDir, ["remote", "remove", "origin"]); + yield* manager.invalidateLocalStatus(repoDir); + const afterPrimaryRemoteRemoval = yield* manager.localStatus({ cwd: repoDir }); + expect(afterPrimaryRemoteRemoval.coherenceToken).not.toBe(afterHeadMove.coherenceToken); + + yield* runGit(repoDir, ["remote", "add", "origin", replacementOriginDir]); + yield* manager.invalidateLocalStatus(repoDir); + const afterPrimaryRemoteAddition = yield* manager.localStatus({ cwd: repoDir }); + expect(afterPrimaryRemoteAddition.coherenceToken).not.toBe( + afterPrimaryRemoteRemoval.coherenceToken, + ); + + yield* runGit(repoDir, ["remote", "add", "fork", originDir]); + yield* runGit(repoDir, ["fetch", "fork", "main"]); + yield* runGit(repoDir, ["branch", "--set-upstream-to=fork/main", "main"]); + yield* manager.invalidateLocalStatus(repoDir); + const afterUpstreamChange = yield* manager.localStatus({ cwd: repoDir }); + expect(afterUpstreamChange.coherenceToken).not.toBe( + afterPrimaryRemoteAddition.coherenceToken, + ); + }), + ); + it.effect("status skips the provider lookup for a branch that was never pushed", () => Effect.gen(function* () { const repoDir = yield* makeTempDir("t3code-git-manager-"); diff --git a/apps/server/src/git/GitManager.ts b/apps/server/src/git/GitManager.ts index 553eda7bb9c3..de3d474d1517 100644 --- a/apps/server/src/git/GitManager.ts +++ b/apps/server/src/git/GitManager.ts @@ -1,3 +1,5 @@ +import * as NodeCrypto from "node:crypto"; + import * as Arr from "effect/Array"; import * as Cache from "effect/Cache"; import * as Context from "effect/Context"; @@ -69,6 +71,11 @@ export interface GitRunStackedActionOptions { readonly progressReporter?: GitActionProgressReporter; } +export interface GitLocalStatusIdentity { + readonly coherenceToken: string; + readonly remoteAssociationToken: string; +} + interface SourceControlTextGenerationSettings { readonly modelSelection: ModelSelection; readonly style: SourceControlWritingStyleSettings; @@ -83,6 +90,9 @@ export class GitManager extends Context.Service< readonly localStatus: ( input: VcsStatusInput, ) => Effect.Effect; + readonly localStatusIdentity: ( + input: VcsStatusInput, + ) => Effect.Effect; readonly remoteStatus: ( input: VcsStatusInput, options?: GitVcsDriver.GitRemoteStatusOptions, @@ -849,6 +859,7 @@ export const make = Effect.gen(function* () { const normalizeStatusCacheKey = canonicalizeExistingPath; const nonRepositoryStatusDetails = { isRepo: false, + headOid: null, hasOriginRemote: false, isDefaultBranch: false, branch: null, @@ -860,6 +871,34 @@ export const make = Effect.gen(function* () { behindCount: 0, aheadOfDefaultCount: 0, } satisfies GitVcsDriver.GitStatusDetails; + const makeLocalStatusIdentity = ( + details: Pick, + primaryRemoteUrlKey: string | null, + ): GitLocalStatusIdentity => { + const remoteAssociationToken = NodeCrypto.createHash("sha256") + .update( + JSON.stringify({ + isRepo: details.isRepo, + refName: details.branch, + upstreamRef: details.upstreamRef, + primaryRemoteUrlKey, + }), + ) + .digest("hex"); + const coherenceToken = NodeCrypto.createHash("sha256") + .update(JSON.stringify({ remoteAssociationToken, headOid: details.headOid })) + .digest("hex"); + return { coherenceToken, remoteAssociationToken }; + }; + const readPrimaryRemoteUrlKey = Effect.fn("readPrimaryRemoteUrlKey")(function* (cwd: string) { + const url = yield* gitCore.readConfigValue(cwd, "remote.origin.url"); + return url === null ? null : normalizeGitRemoteUrl(url); + }); + const readLocalStatusIdentity = Effect.fn("readLocalStatusIdentity")(function* (cwd: string) { + const details = yield* gitCore.statusIdentityDetails(cwd); + const primaryRemoteUrlKey = details.isRepo ? yield* readPrimaryRemoteUrlKey(cwd) : null; + return makeLocalStatusIdentity(details, primaryRemoteUrlKey); + }); const readLocalStatus = Effect.fn("readLocalStatus")(function* (cwd: string) { const details = yield* gitCore .statusDetailsLocal(cwd) @@ -869,6 +908,8 @@ export const make = Effect.gen(function* () { const hostingProvider = details.isRepo ? yield* resolveHostingProvider(cwd, details.branch) : null; + const primaryRemoteUrlKey = details.isRepo ? yield* readPrimaryRemoteUrlKey(cwd) : null; + const identity = makeLocalStatusIdentity(details, primaryRemoteUrlKey); return { isRepo: details.isRepo, @@ -876,6 +917,7 @@ export const make = Effect.gen(function* () { hasPrimaryRemote: details.hasOriginRemote, isDefaultRef: details.isDefaultBranch, refName: details.branch, + ...(details.isRepo ? identity : {}), hasWorkingTreeChanges: details.hasWorkingTreeChanges, workingTree: details.workingTree, } satisfies VcsStatusLocalResult; @@ -1749,6 +1791,12 @@ export const make = Effect.gen(function* () { return yield* Cache.get(localStatusResultCache, cacheKey); }, ); + const localStatusIdentity: GitManager["Service"]["localStatusIdentity"] = Effect.fn( + "localStatusIdentity", + )(function* (input) { + const cacheKey = yield* normalizeStatusCacheKey(input.cwd); + return yield* readLocalStatusIdentity(cacheKey); + }); const remoteStatus: GitManager["Service"]["remoteStatus"] = Effect.fn("remoteStatus")( function* (input, options) { const cacheKey = yield* normalizeStatusCacheKey(input.cwd); @@ -2215,6 +2263,7 @@ export const make = Effect.gen(function* () { return GitManager.of({ localStatus, + localStatusIdentity, remoteStatus, status, invalidateLocalStatus, diff --git a/apps/server/src/git/GitWorkflowService.ts b/apps/server/src/git/GitWorkflowService.ts index da22794951fb..b00b45a7afb9 100644 --- a/apps/server/src/git/GitWorkflowService.ts +++ b/apps/server/src/git/GitWorkflowService.ts @@ -41,6 +41,9 @@ export class GitWorkflowService extends Context.Service< readonly localStatus: ( input: VcsStatusInput, ) => Effect.Effect; + readonly localStatusIdentity: ( + input: VcsStatusInput, + ) => Effect.Effect; readonly remoteStatus: ( input: VcsStatusInput, options?: GitVcsDriver.GitRemoteStatusOptions, @@ -268,6 +271,25 @@ export const make = Effect.gen(function* () { : Effect.succeed(nonRepositoryLocalStatus()), ), ), + localStatusIdentity: (input) => + detectGitRepositoryForStatus("GitWorkflowService.localStatusIdentity", input.cwd).pipe( + Effect.flatMap((isGitRepository) => + isGitRepository + ? gitManager.localStatusIdentity(input) + : Effect.succeed({ + coherenceToken: JSON.stringify({ + isRepo: false, + refName: null, + hasPrimaryRemote: false, + }), + remoteAssociationToken: JSON.stringify({ + isRepo: false, + refName: null, + hasPrimaryRemote: false, + }), + }), + ), + ), remoteStatus: (input, options) => detectGitRepositoryForStatus("GitWorkflowService.remoteStatus", input.cwd).pipe( Effect.flatMap((isGitRepository) => diff --git a/apps/server/src/sourceControl/SourceControlRepositoryService.test.ts b/apps/server/src/sourceControl/SourceControlRepositoryService.test.ts index 861da9a10e05..20b3ba2330fb 100644 --- a/apps/server/src/sourceControl/SourceControlRepositoryService.test.ts +++ b/apps/server/src/sourceControl/SourceControlRepositoryService.test.ts @@ -373,6 +373,7 @@ it.effect("publish succeeds with status remote_added when the local repo has no isRepo: true, hasOriginRemote: true, isDefaultBranch: true, + headOid: null, branch: "main", upstreamRef: null, hasWorkingTreeChanges: false, diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index f256a7dd4e13..e26869f21c34 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -57,6 +57,7 @@ export interface ExecuteGitResult { export interface GitStatusDetails { isRepo: boolean; + headOid: string | null; sourceControlProvider?: VcsStatusResult["sourceControlProvider"]; hasOriginRemote: boolean; isDefaultBranch: boolean; @@ -81,6 +82,13 @@ export interface GitRemoteStatusDetails { aheadOfDefaultCount: number; } +export interface GitStatusIdentityDetails { + isRepo: boolean; + headOid: string | null; + branch: string | null; + upstreamRef: string | null; +} + export interface GitPreparedCommitContext { stagedSummary: string; stagedPatch: string; @@ -202,6 +210,9 @@ export class GitVcsDriver extends Context.Service< readonly status: (input: VcsStatusInput) => Effect.Effect; readonly statusDetails: (cwd: string) => Effect.Effect; readonly statusDetailsLocal: (cwd: string) => Effect.Effect; + readonly statusIdentityDetails: ( + cwd: string, + ) => Effect.Effect; readonly statusDetailsRemote: ( cwd: string, options?: GitRemoteStatusOptions, diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index d39817c0ee1d..0b6a4a22d105 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -76,6 +76,7 @@ const DEFAULT_BASE_BRANCH_CANDIDATES = ["main", "master"] as const; const GIT_LIST_BRANCHES_DEFAULT_LIMIT = 100; const NON_REPOSITORY_STATUS_DETAILS = Object.freeze({ isRepo: false, + headOid: null, hasOriginRemote: false, isDefaultBranch: false, branch: null, @@ -1653,6 +1654,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* ); const statusStdout = statusResult.stdout; + let headOid: string | null = null; let refName: string | null = null; let upstreamRef: string | null = null; let aheadCount = 0; @@ -1662,6 +1664,11 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* const changedFilesWithoutNumstat = new Set(); for (const line of statusStdout.split(/\r?\n/g)) { + if (line.startsWith("# branch.oid ")) { + const value = line.slice("# branch.oid ".length).trim(); + headOid = value === "(initial)" ? null : value; + continue; + } if (line.startsWith("# branch.head ")) { const value = line.slice("# branch.head ".length).trim(); refName = value.startsWith("(") ? null : value; @@ -1731,6 +1738,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* return { isRepo: true, + headOid, hasOriginRemote: hasPrimaryRemote, isDefaultBranch, branch: refName, @@ -2614,6 +2622,55 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* timeToLive: (exit) => (Exit.isSuccess(exit) ? LIST_REFS_SNAPSHOT_CACHE_TTL : Duration.zero), }, ); + + const statusIdentityDetails: GitVcsDriver.GitVcsDriver["Service"]["statusIdentityDetails"] = + Effect.fn("statusIdentityDetails")(function* (cwd) { + const result = yield* executeGitWithStableDiagnostics( + "GitVcsDriver.statusIdentityDetails", + cwd, + ["status", "--porcelain=2", "--branch", "--untracked-files=no"], + { allowNonZeroExit: true }, + ).pipe( + Effect.catchTags({ + GitCommandError: (error) => + isMissingGitCwdError(error) ? Effect.succeed(null) : Effect.fail(error), + }), + ); + + if (result === null || isNonRepositoryGitStderr(result.stderr)) { + return { isRepo: false, headOid: null, branch: null, upstreamRef: null }; + } + if (result.exitCode !== 0) { + return yield* new GitCommandError({ + ...gitCommandContext({ + operation: "GitVcsDriver.statusIdentityDetails", + cwd, + args: ["status", "--porcelain=2", "--branch", "--untracked-files=no"], + }), + detail: "Git status identity lookup failed.", + exitCode: result.exitCode, + stdoutLength: result.stdout.length, + stderrLength: result.stderr.length, + }); + } + + let headOid: string | null = null; + let branch: string | null = null; + let upstreamRef: string | null = null; + for (const line of result.stdout.split(/\r?\n/g)) { + if (line.startsWith("# branch.oid ")) { + const value = line.slice("# branch.oid ".length).trim(); + headOid = value === "(initial)" ? null : value; + } else if (line.startsWith("# branch.head ")) { + const value = line.slice("# branch.head ".length).trim(); + branch = value.startsWith("(") ? null : value; + } else if (line.startsWith("# branch.upstream ")) { + const value = line.slice("# branch.upstream ".length).trim(); + upstreamRef = value.length > 0 ? value : null; + } + } + return { isRepo: true, headOid, branch, upstreamRef }; + }); const listRefsRefreshSnapshotCache = yield* Cache.makeWith( (cacheKey: GitRefsRefreshCacheKey) => Effect.suspend(() => { @@ -3056,6 +3113,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* status, statusDetails, statusDetailsLocal, + statusIdentityDetails, statusDetailsRemote, prepareCommitContext, commit: (cwd, subject, body, options) => diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 6820a29e2c86..09d6d51f48d3 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -7,6 +7,7 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as Option from "effect/Option"; @@ -67,6 +68,30 @@ const baseStatus: VcsStatusResult = { ...baseRemoteStatus, }; +function testLocalIdentity(local: VcsStatusLocalResult) { + const fallback = JSON.stringify({ + isRepo: local.isRepo, + refName: local.refName, + hasPrimaryRemote: local.hasPrimaryRemote, + }); + return { + coherenceToken: local.coherenceToken ?? fallback, + remoteAssociationToken: local.remoteAssociationToken ?? local.coherenceToken ?? fallback, + }; +} + +function makeWorkflowLayer( + service: Partial & + Pick, +) { + return Layer.mock(GitWorkflowService.GitWorkflowService)({ + ...service, + localStatusIdentity: + service.localStatusIdentity ?? + ((input) => service.localStatus(input).pipe(Effect.map(testLocalIdentity))), + }); +} + function makeTestLayer(state: { currentLocalStatus: VcsStatusLocalResult; currentRemoteStatus: VcsStatusRemoteResult | null; @@ -80,12 +105,13 @@ function makeTestLayer(state: { Layer.provideMerge(NodeServices.layer), Layer.provide(makeBackgroundPolicyLayer(() => true)), Layer.provide( - Layer.mock(GitWorkflowService.GitWorkflowService)({ + makeWorkflowLayer({ localStatus: () => Effect.sync(() => { state.localStatusCalls += 1; return state.currentLocalStatus; }), + localStatusIdentity: () => Effect.sync(() => testLocalIdentity(state.currentLocalStatus)), remoteStatus: (_input, options) => Effect.sync(() => { state.remoteStatusCalls += 1; @@ -222,7 +248,7 @@ describe("VcsStatusBroadcaster", () => { Layer.provideMerge(NodeServices.layer), Layer.provide(makeBackgroundPolicyLayer(() => true)), Layer.provide( - Layer.mock(GitWorkflowService.GitWorkflowService)({ + makeWorkflowLayer({ localStatus: () => Effect.sync(() => { state.localStatusCalls += 1; @@ -309,6 +335,30 @@ describe("VcsStatusBroadcaster", () => { ...state.currentLocalStatus, ...baseRemoteStatus, }); + assert.equal(state.localStatusCalls, 3); + assert.equal(state.remoteStatusCalls, 2); + assert.equal(state.localInvalidationCalls, 1); + assert.equal(state.remoteInvalidationCalls, 1); + }).pipe(Effect.provide(makeTestLayer(state))); + }); + + it.effect("reuses a matching cached local identity while filling a remote cache miss", () => { + const state = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.refreshLocalStatus("/repo"); + + const status = yield* broadcaster.getStatus({ cwd: "/repo" }); + + assert.deepStrictEqual(status, baseStatus); assert.equal(state.localStatusCalls, 2); assert.equal(state.remoteStatusCalls, 1); assert.equal(state.localInvalidationCalls, 1); @@ -316,6 +366,577 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(makeTestLayer(state))); }); + it.effect("does not pair a new ref with cached remote status from the previous ref", () => { + const state = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + + state.currentLocalStatus = { + ...baseLocalStatus, + refName: "feature/new-ref", + }; + yield* broadcaster.refreshLocalStatus("/repo"); + + const snapshot = yield* Stream.runHead( + broadcaster.streamStatus( + { cwd: "/repo" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) }, + ), + ); + + assert.isTrue(Option.isSome(snapshot)); + if (Option.isSome(snapshot)) { + assert.equal(snapshot.value._tag, "snapshot"); + if (snapshot.value._tag === "snapshot") { + assert.equal(snapshot.value.local.refName, "feature/new-ref"); + assert.isNull(snapshot.value.remote); + } + } + }).pipe(Effect.provide(makeTestLayer(state))); + }); + + it.effect("discards a stale remote refresh and refetches the current generation", () => { + const state = { + currentLocalStatus: baseLocalStatus, + localStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + let pendingRemote: Deferred.Deferred | null = null; + let remoteCacheUpdated: Deferred.Deferred | null = null; + const testLayer = VcsStatusBroadcaster.layerWithOptions({ + afterRemoteCacheUpdate: ({ accepted }) => + remoteCacheUpdated + ? Deferred.succeed(remoteCacheUpdated, accepted).pipe(Effect.ignore) + : Effect.void, + }).pipe( + Layer.provideMerge(NodeServices.layer), + Layer.provide(makeBackgroundPolicyLayer(() => true)), + Layer.provide( + makeWorkflowLayer({ + localStatus: () => + Effect.sync(() => { + state.localStatusCalls += 1; + return state.currentLocalStatus; + }), + remoteStatus: () => + pendingRemote === null + ? Effect.die("pending remote is not initialized") + : Deferred.await(pendingRemote), + invalidateLocalStatus: () => + Effect.sync(() => { + state.localInvalidationCalls += 1; + }), + invalidateRemoteStatus: () => + Effect.sync(() => { + state.remoteInvalidationCalls += 1; + }), + } satisfies Partial), + ), + ); + + return Effect.gen(function* () { + pendingRemote = yield* Deferred.make(); + remoteCacheUpdated = yield* Deferred.make(); + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const scope = yield* Scope.make(); + const snapshotSeen = yield* Deferred.make(); + const remoteUpdateSeen = yield* Deferred.make(); + yield* Stream.runForEach( + broadcaster.streamStatus( + { cwd: "/repo" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) }, + ), + (event) => { + if (event._tag === "snapshot") { + return Deferred.succeed(snapshotSeen, undefined).pipe(Effect.ignore); + } + if (event._tag === "remoteUpdated") { + return Deferred.succeed(remoteUpdateSeen, event).pipe(Effect.ignore); + } + return Effect.void; + }, + ).pipe(Effect.forkIn(scope)); + + yield* Deferred.await(snapshotSeen); + state.currentLocalStatus = { ...baseLocalStatus, refName: "feature/new-ref" }; + yield* broadcaster.refreshLocalStatus("/repo"); + yield* Deferred.succeed(pendingRemote, baseRemoteStatus); + assert.isFalse(yield* Deferred.await(remoteCacheUpdated)); + + assert.deepStrictEqual(yield* Deferred.await(remoteUpdateSeen), { + _tag: "remoteUpdated", + generation: 1, + remote: baseRemoteStatus, + } satisfies VcsStatusStreamEvent); + yield* Scope.close(scope, Exit.void); + }).pipe(Effect.provide(testLayer)); + }); + + it.effect( + "returns a coherent refresh result when a newer local generation wins the cache race", + () => { + let currentLocal = baseLocalStatus; + let delayRemote = false; + let delayedRemote: Deferred.Deferred | null = null; + let staleLocalRead: Deferred.Deferred | null = null; + const testLayer = VcsStatusBroadcaster.layer.pipe( + Layer.provideMerge(NodeServices.layer), + Layer.provide(makeBackgroundPolicyLayer(() => true)), + Layer.provide( + makeWorkflowLayer({ + localStatus: () => + Effect.sync(() => currentLocal).pipe( + Effect.tap(() => + delayRemote && staleLocalRead + ? Deferred.succeed(staleLocalRead, undefined).pipe(Effect.ignore) + : Effect.void, + ), + ), + remoteStatus: () => + Effect.suspend(() => { + if (!delayRemote) return Effect.succeed(baseRemoteStatus); + return delayedRemote + ? Deferred.await(delayedRemote) + : Effect.die("delayed remote is not initialized"); + }), + invalidateLocalStatus: () => Effect.void, + invalidateRemoteStatus: () => Effect.void, + invalidateStatus: () => Effect.void, + }), + ), + ); + + return Effect.gen(function* () { + delayedRemote = yield* Deferred.make(); + staleLocalRead = yield* Deferred.make(); + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + delayRemote = true; + const staleRefresh = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped); + yield* Deferred.await(staleLocalRead); + + currentLocal = { ...baseLocalStatus, refName: "feature/new-ref" }; + yield* broadcaster.refreshLocalStatus("/repo"); + yield* Deferred.succeed(delayedRemote, baseRemoteStatus); + const refreshed = yield* Fiber.join(staleRefresh); + + assert.equal(refreshed.refName, "feature/new-ref"); + assert.equal(refreshed.hasUpstream, true); + }).pipe(Effect.provide(testLayer)); + }, + ); + + it.effect("returns its coherent read when a local-only writer wins the cache race", () => { + let currentLocal = baseLocalStatus; + let remoteStarted: Deferred.Deferred | null = null; + let releaseRemote: Deferred.Deferred | null = null; + const testLayer = VcsStatusBroadcaster.layer.pipe( + Layer.provideMerge(NodeServices.layer), + Layer.provide(makeBackgroundPolicyLayer(() => true)), + Layer.provide( + makeWorkflowLayer({ + localStatus: () => Effect.sync(() => currentLocal), + remoteStatus: () => + Effect.gen(function* () { + if (remoteStarted) { + yield* Deferred.succeed(remoteStarted, undefined).pipe(Effect.ignore); + } + if (!releaseRemote) { + return yield* Effect.die("release remote is not initialized"); + } + return yield* Deferred.await(releaseRemote); + }), + invalidateLocalStatus: () => Effect.void, + invalidateRemoteStatus: () => Effect.void, + invalidateStatus: () => Effect.void, + }), + ), + ); + + return Effect.gen(function* () { + remoteStarted = yield* Deferred.make(); + releaseRemote = yield* Deferred.make(); + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const read = yield* broadcaster.getStatus({ cwd: "/repo" }).pipe(Effect.forkScoped); + yield* Deferred.await(remoteStarted); + + currentLocal = { + ...baseLocalStatus, + hasWorkingTreeChanges: true, + workingTree: { + files: [{ path: "changed.ts", insertions: 1, deletions: 0 }], + insertions: 1, + deletions: 0, + }, + }; + yield* broadcaster.refreshLocalStatus("/repo"); + yield* Deferred.succeed(releaseRemote, baseRemoteStatus); + const status = yield* Fiber.join(read); + + assert.deepStrictEqual(status.workingTree.files, []); + assert.equal(status.hasUpstream, true); + }).pipe(Effect.provide(testLayer)); + }); + + it.effect("retries a full read when the ref changes between local and remote status", () => { + let localCall = 0; + let remoteCall = 0; + const localB = { ...baseLocalStatus, refName: "feature/new-ref" }; + const remoteB = { ...baseRemoteStatus, aheadCount: 2 }; + const testLayer = VcsStatusBroadcaster.layer.pipe( + Layer.provideMerge(NodeServices.layer), + Layer.provide(makeBackgroundPolicyLayer(() => true)), + Layer.provide( + makeWorkflowLayer({ + localStatus: () => + Effect.sync(() => { + localCall += 1; + return localCall === 1 ? baseLocalStatus : localB; + }), + remoteStatus: () => + Effect.sync(() => { + remoteCall += 1; + return remoteCall === 1 ? baseRemoteStatus : remoteB; + }), + invalidateLocalStatus: () => Effect.void, + invalidateRemoteStatus: () => Effect.void, + invalidateStatus: () => Effect.void, + }), + ), + ); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const status = yield* broadcaster.getStatus({ cwd: "/repo" }); + + assert.equal(status.refName, "feature/new-ref"); + assert.equal(status.aheadCount, 2); + assert.equal(localCall, 4); + assert.equal(remoteCall, 2); + }).pipe(Effect.provide(testLayer)); + }); + + for (const [name, firstToken, confirmedToken] of [ + ["same-branch HEAD move", "head-a|origin|origin/feature", "head-b|origin|origin/feature"], + ["primary remote removal", "head-a|origin|origin/feature", "head-a|none|origin/feature"], + ["upstream change", "head-a|origin|origin/feature", "head-a|origin|fork/feature"], + ] as const) { + it.effect(`retries a coherent read after a ${name}`, () => { + let localCall = 0; + let remoteCall = 0; + const first = { ...baseLocalStatus, coherenceToken: firstToken }; + const confirmed = { ...baseLocalStatus, coherenceToken: confirmedToken }; + const freshRemote = { ...baseRemoteStatus, aheadCount: 7 }; + const testLayer = VcsStatusBroadcaster.layer.pipe( + Layer.provideMerge(NodeServices.layer), + Layer.provide(makeBackgroundPolicyLayer(() => true)), + Layer.provide( + makeWorkflowLayer({ + localStatus: () => + Effect.sync(() => { + localCall += 1; + return localCall === 1 ? first : confirmed; + }), + remoteStatus: () => + Effect.sync(() => { + remoteCall += 1; + return remoteCall === 1 ? baseRemoteStatus : freshRemote; + }), + invalidateLocalStatus: () => Effect.void, + invalidateRemoteStatus: () => Effect.void, + invalidateStatus: () => Effect.void, + }), + ), + ); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const status = yield* broadcaster.getStatus({ cwd: "/repo" }); + + assert.equal(status.aheadCount, 7); + assert.equal(localCall, 4); + assert.equal(remoteCall, 2); + }).pipe(Effect.provide(testLayer)); + }); + } + + it.effect("fails after three continuously changing coherence identities", () => { + let localCall = 0; + let remoteCall = 0; + const testLayer = VcsStatusBroadcaster.layer.pipe( + Layer.provideMerge(NodeServices.layer), + Layer.provide(makeBackgroundPolicyLayer(() => true)), + Layer.provide( + makeWorkflowLayer({ + localStatus: () => + Effect.sync(() => ({ + ...baseLocalStatus, + coherenceToken: `identity-${++localCall}`, + })), + remoteStatus: () => + Effect.sync(() => { + remoteCall += 1; + return baseRemoteStatus; + }), + invalidateLocalStatus: () => Effect.void, + invalidateRemoteStatus: () => Effect.void, + invalidateStatus: () => Effect.void, + }), + ), + ); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const exit = yield* broadcaster.getStatus({ cwd: "/repo" }).pipe(Effect.exit); + + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + const failure = Cause.squash(exit.cause); + assert.instanceOf(failure, GitManagerError); + assert.equal( + (failure as GitManagerError).operation, + "VcsStatusBroadcaster.readCoherentStatus", + ); + } + assert.equal(localCall, 6); + assert.equal(remoteCall, 3); + }).pipe(Effect.provide(testLayer)); + }); + + it.effect("builds the initial snapshot from one cached generation", () => { + const state = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const snapshot = yield* Stream.runHead( + broadcaster.streamStatus( + { cwd: "/repo" }, + { + automaticRemoteRefreshInterval: Effect.succeed(Duration.zero), + beforeInitialSnapshot: Effect.suspend(() => { + state.currentLocalStatus = { ...baseLocalStatus, refName: "feature/new-ref" }; + return broadcaster.refreshLocalStatus("/repo").pipe(Effect.orDie, Effect.asVoid); + }), + }, + ), + ); + + assert.isTrue(Option.isSome(snapshot)); + if (Option.isSome(snapshot) && snapshot.value._tag === "snapshot") { + assert.equal(snapshot.value.generation, 1); + assert.equal(snapshot.value.local.refName, "feature/new-ref"); + assert.isNull(snapshot.value.remote); + } + }).pipe(Effect.provide(makeTestLayer(state))); + }); + + it.effect("publishes a generation bump committed by getStatus", () => { + const state = { + currentLocalStatus: { ...baseLocalStatus, coherenceToken: "head-a" }, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + const published: VcsStatusStreamEvent[] = []; + const testLayer = VcsStatusBroadcaster.layerWithOptions({ + beforePublish: (change) => Effect.sync(() => published.push(change.event)), + }).pipe( + Layer.provideMerge(NodeServices.layer), + Layer.provide(makeBackgroundPolicyLayer(() => true)), + Layer.provide( + makeWorkflowLayer({ + localStatus: () => Effect.succeed(state.currentLocalStatus), + localStatusIdentity: () => Effect.succeed(testLocalIdentity(state.currentLocalStatus)), + remoteStatus: () => Effect.succeed(state.currentRemoteStatus), + invalidateLocalStatus: () => Effect.void, + invalidateRemoteStatus: () => Effect.void, + invalidateStatus: () => Effect.void, + }), + ), + ); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.refreshLocalStatus("/repo"); + state.currentLocalStatus = { ...baseLocalStatus, coherenceToken: "head-b" }; + + yield* broadcaster.getStatus({ cwd: "/repo" }); + + assert.deepStrictEqual(published.at(-1), { + _tag: "snapshot", + generation: 1, + local: state.currentLocalStatus, + remote: baseRemoteStatus, + } satisfies VcsStatusStreamEvent); + }).pipe(Effect.provide(testLayer)); + }); + + it.effect("publishes a new local generation before awaiting remote invalidation", () => { + let currentLocal = baseLocalStatus; + let invalidationGate: Deferred.Deferred | null = null; + const events: VcsStatusStreamEvent[] = []; + const testLayer = VcsStatusBroadcaster.layer.pipe( + Layer.provideMerge(NodeServices.layer), + Layer.provide(makeBackgroundPolicyLayer(() => true)), + Layer.provide( + makeWorkflowLayer({ + localStatus: () => Effect.succeed(currentLocal), + remoteStatus: () => Effect.succeed(baseRemoteStatus), + invalidateLocalStatus: () => Effect.void, + invalidateRemoteStatus: () => + invalidationGate ? Deferred.await(invalidationGate) : Effect.void, + invalidateStatus: () => Effect.void, + }), + ), + ); + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const scope = yield* Scope.make(); + const initialSeen = yield* Deferred.make(); + const localSeen = yield* Deferred.make(); + yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => { + events.push(event); + if (event._tag === "snapshot") { + return Deferred.succeed(initialSeen, undefined).pipe(Effect.ignore); + } + if (event._tag === "localUpdated") { + return Deferred.succeed(localSeen, undefined).pipe(Effect.ignore); + } + return Effect.void; + }).pipe(Effect.forkIn(scope)); + yield* Deferred.await(initialSeen); + + invalidationGate = yield* Deferred.make(); + currentLocal = { ...baseLocalStatus, refName: "feature/new-ref" }; + const refresh = yield* broadcaster.refreshLocalStatus("/repo").pipe(Effect.forkScoped); + yield* Deferred.await(localSeen); + + assert.equal(events.at(-1)?._tag, "localUpdated"); + yield* Deferred.succeed(invalidationGate, undefined); + yield* Fiber.join(refresh); + yield* Scope.close(scope, Exit.void); + }).pipe(Effect.provide(testLayer)); + }); + + it.effect("commits a transition before publishing and serializes the next writer", () => { + const initial = { ...baseLocalStatus, coherenceToken: "stable" }; + const localA = { + ...initial, + hasWorkingTreeChanges: true, + workingTree: { + files: [{ path: "a.ts", insertions: 1, deletions: 0 }], + insertions: 1, + deletions: 0, + }, + }; + const localB = { + ...initial, + hasWorkingTreeChanges: true, + workingTree: { + files: [{ path: "b.ts", insertions: 2, deletions: 0 }], + insertions: 2, + deletions: 0, + }, + }; + let currentLocal = initial; + let publicationGate: Deferred.Deferred | null = null; + let firstPublicationStarted: Deferred.Deferred | null = null; + let secondPublicationStarted: Deferred.Deferred | null = null; + let publishCount = 0; + const events: VcsStatusStreamEvent[] = []; + const testLayer = VcsStatusBroadcaster.layerWithOptions({ + beforePublish: (change) => { + if (change.event._tag !== "localUpdated") return Effect.void; + publishCount += 1; + if (publishCount === 1 && firstPublicationStarted && publicationGate) { + return Deferred.succeed(firstPublicationStarted, undefined).pipe( + Effect.andThen(Deferred.await(publicationGate)), + Effect.asVoid, + ); + } + return secondPublicationStarted + ? Deferred.succeed(secondPublicationStarted, undefined).pipe(Effect.ignore) + : Effect.void; + }, + }).pipe( + Layer.provideMerge(NodeServices.layer), + Layer.provide(makeBackgroundPolicyLayer(() => true)), + Layer.provide( + makeWorkflowLayer({ + localStatus: () => Effect.sync(() => currentLocal), + remoteStatus: () => Effect.succeed(baseRemoteStatus), + invalidateLocalStatus: () => Effect.void, + invalidateRemoteStatus: () => Effect.void, + invalidateStatus: () => Effect.void, + }), + ), + ); + + return Effect.gen(function* () { + publicationGate = yield* Deferred.make(); + firstPublicationStarted = yield* Deferred.make(); + secondPublicationStarted = yield* Deferred.make(); + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const snapshotSeen = yield* Deferred.make(); + const scope = yield* Scope.make(); + yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => { + events.push(event); + return event._tag === "snapshot" + ? Deferred.succeed(snapshotSeen, undefined).pipe(Effect.ignore) + : Effect.void; + }).pipe(Effect.forkIn(scope)); + yield* Deferred.await(snapshotSeen); + + currentLocal = localA; + const writerA = yield* broadcaster.refreshLocalStatus("/repo").pipe(Effect.forkScoped); + yield* Deferred.await(firstPublicationStarted); + const duringPublish = yield* broadcaster.getStatus({ cwd: "/repo" }); + assert.equal(duringPublish.workingTree.files[0]?.path, "a.ts"); + + currentLocal = localB; + const writerB = yield* broadcaster.refreshLocalStatus("/repo").pipe(Effect.forkScoped); + assert.isTrue(Option.isNone(yield* Deferred.poll(secondPublicationStarted))); + + yield* Deferred.succeed(publicationGate, undefined); + yield* Deferred.await(secondPublicationStarted); + yield* Fiber.join(writerA); + yield* Fiber.join(writerB); + const finalStatus = yield* broadcaster.getStatus({ cwd: "/repo" }); + assert.equal(finalStatus.workingTree.files[0]?.path, "b.ts"); + assert.deepStrictEqual( + events + .filter((event) => event._tag === "localUpdated") + .map((event) => + event._tag === "localUpdated" ? event.local.workingTree.files[0]?.path : null, + ), + ["a.ts", "b.ts"], + ); + yield* Scope.close(scope, Exit.void); + }).pipe(Effect.provide(testLayer)); + }); + it.effect("normalizes symlinked CWDs before cache lookup and workflow calls", () => { const seenCwds: string[] = []; const state = { @@ -330,7 +951,7 @@ describe("VcsStatusBroadcaster", () => { Layer.provideMerge(NodeServices.layer), Layer.provide(makeBackgroundPolicyLayer(() => true)), Layer.provide( - Layer.mock(GitWorkflowService.GitWorkflowService)({ + makeWorkflowLayer({ localStatus: (input) => Effect.sync(() => { seenCwds.push(input.cwd); @@ -372,8 +993,8 @@ describe("VcsStatusBroadcaster", () => { yield* broadcaster.getStatus({ cwd: linkDir }); yield* broadcaster.getStatus({ cwd: realDir }); - assert.deepStrictEqual(seenCwds, [realPath, realPath]); - assert.equal(state.localStatusCalls, 1); + assert.deepStrictEqual(seenCwds, [realPath, realPath, realPath]); + assert.equal(state.localStatusCalls, 2); assert.equal(state.remoteStatusCalls, 1); }).pipe(Effect.provide(testLayer)); }); @@ -408,11 +1029,13 @@ describe("VcsStatusBroadcaster", () => { assert.deepStrictEqual(snapshot, { _tag: "snapshot", + generation: 0, local: baseLocalStatus, remote: null, } satisfies VcsStatusStreamEvent); assert.deepStrictEqual(remoteUpdated, { _tag: "remoteUpdated", + generation: 0, remote: baseRemoteStatus, } satisfies VcsStatusStreamEvent); }).pipe(Effect.provide(makeTestLayer(state))); @@ -455,11 +1078,13 @@ describe("VcsStatusBroadcaster", () => { assert.deepStrictEqual(snapshot, { _tag: "snapshot", + generation: 0, local: baseLocalStatus, remote: null, } satisfies VcsStatusStreamEvent); assert.deepStrictEqual(remoteUpdated, { _tag: "remoteUpdated", + generation: 0, remote: remoteStatusWithPr, } satisfies VcsStatusStreamEvent); assert.equal(state.remoteStatusCalls, 1); @@ -474,6 +1099,82 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(Layer.merge(makeTestLayer(state), TestClock.layer()))); }); + it.effect( + "refetches remote immediately after a coherence change with periodic refresh disabled", + () => { + const state = { + currentLocalStatus: { + ...baseLocalStatus, + coherenceToken: "head-a", + remoteAssociationToken: "stable-branch-upstream-origin", + }, + currentRemoteStatus: remoteStatusWithPr, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + remoteStatusRefreshUpstreamValues: [] as Array, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const scope = yield* Scope.make(); + const initialRemoteSeen = yield* Deferred.make(); + const nextLocalSeen = yield* Deferred.make(); + const nextRemoteSeen = yield* Deferred.make(); + yield* Stream.runForEach( + broadcaster.streamStatus( + { cwd: "/repo" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) }, + ), + (event) => { + if (event._tag === "snapshot" && event.generation === 1) { + return Deferred.succeed(nextLocalSeen, event).pipe(Effect.ignore); + } + if (event._tag === "remoteUpdated") { + if (event.generation === 0) { + return Deferred.succeed(initialRemoteSeen, undefined).pipe(Effect.ignore); + } + if (event.generation === 1) { + return Deferred.succeed(nextRemoteSeen, event).pipe(Effect.ignore); + } + } + return Effect.void; + }, + ).pipe(Effect.forkIn(scope)); + + yield* Deferred.await(initialRemoteSeen); + state.currentLocalStatus = { + ...baseLocalStatus, + coherenceToken: "head-b", + remoteAssociationToken: "stable-branch-upstream-origin", + }; + state.currentRemoteStatus = { ...baseRemoteStatus, aheadCount: 5 }; + yield* broadcaster.refreshLocalStatus("/repo"); + const headTransition = yield* Deferred.await(nextLocalSeen); + const nextRemote = yield* Deferred.await(nextRemoteSeen); + + assert.deepStrictEqual(headTransition, { + _tag: "snapshot", + generation: 1, + local: state.currentLocalStatus, + remote: remoteStatusWithPr, + } satisfies VcsStatusStreamEvent); + + assert.deepStrictEqual(nextRemote, { + _tag: "remoteUpdated", + generation: 1, + remote: state.currentRemoteStatus, + } satisfies VcsStatusStreamEvent); + assert.equal(state.remoteStatusCalls, 2); + assert.equal(state.remoteInvalidationCalls, 1); + assert.deepStrictEqual(state.remoteStatusRefreshUpstreamValues, [false, false]); + + yield* Scope.close(scope, Exit.void); + }).pipe(Effect.provide(makeTestLayer(state))); + }, + ); + it.effect("retries the initial remote load when periodic refreshes are disabled", () => { const state = { currentLocalStatus: baseLocalStatus, @@ -494,7 +1195,7 @@ describe("VcsStatusBroadcaster", () => { Layer.provideMerge(NodeServices.layer), Layer.provide(makeBackgroundPolicyLayer(() => true)), Layer.provide( - Layer.mock(GitWorkflowService.GitWorkflowService)({ + makeWorkflowLayer({ localStatus: () => Effect.sync(() => { state.localStatusCalls += 1; @@ -577,6 +1278,7 @@ describe("VcsStatusBroadcaster", () => { assert.deepStrictEqual(remoteUpdated, { _tag: "remoteUpdated", + generation: 0, remote: remoteStatusWithPr, } satisfies VcsStatusStreamEvent); assert.equal(state.remoteStatusCalls, 2); @@ -694,7 +1396,7 @@ describe("VcsStatusBroadcaster", () => { Layer.provideMerge(NodeServices.layer), Layer.provide(makeBackgroundPolicyLayer(() => false)), Layer.provide( - Layer.mock(GitWorkflowService.GitWorkflowService)({ + makeWorkflowLayer({ localStatus: () => Effect.sync(() => { state.localStatusCalls += 1; @@ -747,7 +1449,7 @@ describe("VcsStatusBroadcaster", () => { Layer.provideMerge(NodeServices.layer), Layer.provide(makeBackgroundPolicyLayer(() => true)), Layer.provide( - Layer.mock(GitWorkflowService.GitWorkflowService)({ + makeWorkflowLayer({ localStatus: () => Effect.sync(() => { state.localStatusCalls += 1; diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index f28069f6d8b2..41ac28cfe145 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -7,8 +7,9 @@ import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; -import * as Schedule from "effect/Schedule"; +import * as Semaphore from "effect/Semaphore"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as SynchronizedRef from "effect/SynchronizedRef"; @@ -20,6 +21,7 @@ import type { VcsStatusResult, VcsStatusStreamEvent, } from "@t3tools/contracts"; +import { GitManagerError } from "@t3tools/contracts"; import { mergeGitStatusParts } from "@t3tools/shared/git"; import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; @@ -124,19 +126,27 @@ interface CachedValue { readonly value: T; } +interface CachedRemoteStatus extends CachedValue { + readonly generation: number; +} + interface CachedVcsStatus { + readonly generation: number; readonly local: CachedValue | null; - readonly remote: CachedValue | null; + readonly remote: CachedRemoteStatus | null; } interface ActiveRemotePoller { readonly fiber: Fiber.Fiber; readonly subscriberCount: number; readonly demandCwds: Ref.Ref>; + readonly needsRefresh: Ref.Ref; + readonly refreshRequests: Queue.Queue; } interface StreamStatusOptions { readonly automaticRemoteRefreshInterval?: Effect.Effect; + readonly beforeInitialSnapshot?: Effect.Effect; } export function remoteRefreshFailureDelay( @@ -174,13 +184,38 @@ function fingerprintStatusPart(status: unknown): string { return JSON.stringify(status); } +function localStatusIdentity(status: VcsStatusLocalResult): string { + return ( + status.coherenceToken ?? + JSON.stringify({ + isRepo: status.isRepo, + refName: status.refName, + hasPrimaryRemote: status.hasPrimaryRemote, + }) + ); +} + +function remoteAssociationIdentity(status: VcsStatusLocalResult): string { + return status.remoteAssociationToken ?? localStatusIdentity(status); +} + const normalizeCwd = (cwd: string) => Effect.service(FileSystem.FileSystem).pipe( Effect.flatMap((fs) => fs.realPath(cwd)), Effect.orElseSucceed(() => cwd), ); -export const make = Effect.gen(function* () { +interface MakeOptions { + readonly beforePublish?: (change: VcsStatusChange) => Effect.Effect; + readonly afterRemoteCacheUpdate?: (input: { + readonly cwd: string; + readonly accepted: boolean; + }) => Effect.Effect; +} + +export const makeWithOptions = Effect.fn("VcsStatusBroadcaster.make")(function* ( + makeOptions: MakeOptions = {}, +) { const workflow = yield* GitWorkflowService.GitWorkflowService; const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; const fs = yield* FileSystem.FileSystem; @@ -192,38 +227,116 @@ export const make = Effect.gen(function* () { Scope.close(scope, Exit.void), ); const cacheRef = yield* Ref.make(new Map()); + const transitionLocksRef = yield* SynchronizedRef.make(new Map()); const pollersRef = yield* SynchronizedRef.make(new Map()); + const requestRemoteRefresh = Effect.fn("VcsStatusBroadcaster.requestRemoteRefresh")(function* ( + cwd: string, + ) { + const poller = (yield* SynchronizedRef.get(pollersRef)).get(cwd); + if (!poller) { + return; + } + yield* Ref.set(poller.needsRefresh, true); + yield* Queue.offer(poller.refreshRequests, undefined); + }); + const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* ( cwd: string, ) { return yield* Ref.get(cacheRef).pipe(Effect.map((cache) => cache.get(cwd) ?? null)); }); + const transitionLockFor = Effect.fn("VcsStatusBroadcaster.transitionLockFor")(function* ( + cwd: string, + ) { + return yield* SynchronizedRef.modifyEffect(transitionLocksRef, (locks) => { + const existing = locks.get(cwd); + if (existing) { + return Effect.succeed([existing, locks] as const); + } + return Semaphore.make(1).pipe( + Effect.map((created) => [created, new Map(locks).set(cwd, created)] as const), + ); + }); + }); + + const publishChange = (change: VcsStatusChange) => + Effect.gen(function* () { + yield* makeOptions.beforePublish?.(change) ?? Effect.void; + yield* PubSub.publish(changesPubSub, change); + }); + const updateCachedLocalStatus = Effect.fn("VcsStatusBroadcaster.updateCachedLocalStatus")( function* (cwd: string, local: VcsStatusLocalResult, options?: { publish?: boolean }) { const nextLocal = { fingerprint: fingerprintStatusPart(local), value: local, } satisfies CachedValue; - const shouldPublish = yield* Ref.modify(cacheRef, (cache) => { - const previous = cache.get(cwd) ?? { local: null, remote: null }; - const nextCache = new Map(cache); - nextCache.set(cwd, { - ...previous, - local: nextLocal, - }); - return [previous.local?.fingerprint !== nextLocal.fingerprint, nextCache] as const; - }); - - if (options?.publish && shouldPublish) { - yield* PubSub.publish(changesPubSub, { - cwd, - event: { - _tag: "localUpdated", - local, - }, - }); + const lock = yield* transitionLockFor(cwd); + const update = yield* lock.withPermits(1)( + Effect.uninterruptible( + Effect.gen(function* () { + const committed = yield* Ref.modify(cacheRef, (cache) => { + const previous = cache.get(cwd) ?? { + generation: 0, + local: null, + remote: null, + }; + const refChanged = + previous.local !== null && + localStatusIdentity(previous.local.value) !== localStatusIdentity(local); + const remoteAssociationChanged = + refChanged && + previous.local !== null && + remoteAssociationIdentity(previous.local.value) !== + remoteAssociationIdentity(local); + const generation = previous.generation + (refChanged ? 1 : 0); + // A HEAD-only move cannot change which PR belongs to this branch. + // Carry the last known remote snapshot into the new generation so + // clients do not invent zero counts or lose the PR while the + // forced remote refresh below replaces those last-known counts. + const retainedRemote = + refChanged && !remoteAssociationChanged && previous.remote !== null + ? { ...previous.remote, generation } + : null; + const nextCache = new Map(cache); + nextCache.set(cwd, { + ...previous, + generation, + local: nextLocal, + remote: refChanged ? retainedRemote : previous.remote, + }); + return [ + { + generation, + refChanged, + retainedRemote, + shouldPublish: previous.local?.fingerprint !== nextLocal.fingerprint, + }, + nextCache, + ] as const; + }); + if (options?.publish && committed.shouldPublish) { + yield* publishChange({ + cwd, + event: committed.retainedRemote + ? { + _tag: "snapshot", + generation: committed.generation, + local, + remote: committed.retainedRemote.value, + } + : { _tag: "localUpdated", generation: committed.generation, local }, + }); + } + return committed; + }), + ), + ); + if (update.refChanged) { + yield* workflow.invalidateRemoteStatus(cwd); + yield* requestRemoteRefresh(cwd); } return local; @@ -231,32 +344,65 @@ export const make = Effect.gen(function* () { ); const updateCachedRemoteStatus = Effect.fn("VcsStatusBroadcaster.updateCachedRemoteStatus")( - function* (cwd: string, remote: VcsStatusRemoteResult | null, options?: { publish?: boolean }) { + function* ( + cwd: string, + remote: VcsStatusRemoteResult | null, + generation: number, + options?: { publish?: boolean }, + ) { const nextRemote = { + generation, fingerprint: fingerprintStatusPart(remote), value: remote, - } satisfies CachedValue; - const shouldPublish = yield* Ref.modify(cacheRef, (cache) => { - const previous = cache.get(cwd) ?? { local: null, remote: null }; - const nextCache = new Map(cache); - nextCache.set(cwd, { - ...previous, - remote: nextRemote, - }); - return [previous.remote?.fingerprint !== nextRemote.fingerprint, nextCache] as const; - }); - - if (options?.publish && shouldPublish) { - yield* PubSub.publish(changesPubSub, { - cwd, - event: { - _tag: "remoteUpdated", - remote, - }, - }); - } + } satisfies CachedRemoteStatus; + const lock = yield* transitionLockFor(cwd); + const update = yield* lock.withPermits(1)( + Effect.uninterruptible( + Effect.gen(function* () { + const result = yield* Ref.modify( + cacheRef, + ( + cache, + ): readonly [ + { readonly accepted: boolean; readonly shouldPublish: boolean }, + Map, + ] => { + const previous = cache.get(cwd) ?? { + generation: 0, + local: null, + remote: null, + }; + if (previous.local === null || previous.generation !== generation) { + return [{ accepted: false, shouldPublish: false }, cache] as const; + } + const nextCache = new Map(cache); + nextCache.set(cwd, { ...previous, remote: nextRemote }); + return [ + { + accepted: true, + shouldPublish: + previous.remote?.generation !== generation || + previous.remote.fingerprint !== nextRemote.fingerprint, + }, + nextCache, + ] as const; + }, + ); + if (options?.publish && result.shouldPublish) { + yield* publishChange({ + cwd, + event: { _tag: "remoteUpdated", generation, remote }, + }); + } + return result; + }), + ), + ); + yield* ( + makeOptions.afterRemoteCacheUpdate?.({ cwd, accepted: update.accepted }) ?? Effect.void + ); - return remote; + return update.accepted; }, ); @@ -264,42 +410,76 @@ export const make = Effect.gen(function* () { cwd: string, local: VcsStatusLocalResult, remote: VcsStatusRemoteResult | null, - options?: { publish?: boolean }, + options?: { publish?: boolean; expected?: CachedVcsStatus | null }, ) { const nextLocal = { fingerprint: fingerprintStatusPart(local), value: local, } satisfies CachedValue; - const nextRemote = { - fingerprint: fingerprintStatusPart(remote), - value: remote, - } satisfies CachedValue; - const shouldPublish = yield* Ref.modify(cacheRef, (cache) => { - const previous = cache.get(cwd) ?? { local: null, remote: null }; - const nextCache = new Map(cache); - nextCache.set(cwd, { - local: nextLocal, - remote: nextRemote, - }); - return [ - previous.local?.fingerprint !== nextLocal.fingerprint || - previous.remote?.fingerprint !== nextRemote.fingerprint, - nextCache, - ] as const; - }); - - if (options?.publish && shouldPublish) { - yield* PubSub.publish(changesPubSub, { - cwd, - event: { - _tag: "snapshot", - local, - remote, - }, - }); - } + const lock = yield* transitionLockFor(cwd); + const update = yield* lock.withPermits(1)( + Effect.uninterruptible( + Effect.gen(function* () { + const result = yield* Ref.modify( + cacheRef, + ( + cache, + ): readonly [ + { + readonly accepted: boolean; + readonly generation: number; + readonly shouldPublish: boolean; + }, + Map, + ] => { + const existing = cache.get(cwd) ?? null; + if (options && "expected" in options && existing !== options.expected) { + return [ + { + accepted: false, + generation: existing?.generation ?? 0, + shouldPublish: false, + }, + cache, + ] as const; + } + const previous = existing ?? { generation: 0, local: null, remote: null }; + const refChanged = + previous.local === null || + localStatusIdentity(previous.local.value) !== localStatusIdentity(local); + const generation = previous.generation + (refChanged ? 1 : 0); + const nextRemote = { + generation, + fingerprint: fingerprintStatusPart(remote), + value: remote, + } satisfies CachedRemoteStatus; + const nextCache = new Map(cache); + nextCache.set(cwd, { generation, local: nextLocal, remote: nextRemote }); + return [ + { + accepted: true, + generation, + shouldPublish: + previous.local?.fingerprint !== nextLocal.fingerprint || + previous.remote?.generation !== generation || + previous.remote.fingerprint !== nextRemote.fingerprint, + }, + nextCache, + ] as const; + }, + ); + if (options?.publish && result.shouldPublish) { + yield* publishChange({ + cwd, + event: { _tag: "snapshot", generation: result.generation, local, remote }, + }); + } + return result; + }), + ), + ); - return mergeGitStatusParts(local, remote); + return update.accepted ? mergeGitStatusParts(local, remote) : null; }); const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* ( @@ -321,22 +501,45 @@ export const make = Effect.gen(function* () { const withFileSystem = Effect.provideService(FileSystem.FileSystem, fs); + const readCoherentStatus = Effect.fn("VcsStatusBroadcaster.readCoherentStatus")(function* ( + cwd: string, + ) { + for (let attempt = 0; attempt < 3; attempt += 1) { + const local = yield* workflow.localStatus({ cwd }); + const remote = yield* workflow.remoteStatus({ cwd }); + const confirmedIdentity = yield* workflow.localStatusIdentity({ cwd }); + if (localStatusIdentity(local) === confirmedIdentity.coherenceToken) { + return { local, remote }; + } + yield* workflow.invalidateLocalStatus(cwd); + yield* workflow.invalidateRemoteStatus(cwd); + } + return yield* Effect.fail( + new GitManagerError({ + operation: "VcsStatusBroadcaster.readCoherentStatus", + cwd, + detail: "The current ref changed repeatedly while reading VCS status", + }), + ); + }); + const getStatus: VcsStatusBroadcaster["Service"]["getStatus"] = Effect.fn( "VcsStatusBroadcaster.getStatus", )(function* (input) { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); const cached = yield* getCachedStatus(cwd); - if (cached?.local && cached.remote) { + if (cached?.local && cached.remote?.generation === cached.generation) { return mergeGitStatusParts(cached.local.value, cached.remote.value); } - const [local, remote] = yield* Effect.all( - [ - cached?.local ? Effect.succeed(cached.local.value) : workflow.localStatus({ cwd }), - cached?.remote ? Effect.succeed(cached.remote.value) : workflow.remoteStatus({ cwd }), - ], - { concurrency: "unbounded" }, - ); - return yield* updateCachedStatus(cwd, local, remote); + const { local, remote } = yield* readCoherentStatus(cwd); + const updated = yield* updateCachedStatus(cwd, local, remote, { + publish: true, + expected: cached, + }); + if (updated !== null) { + return updated; + } + return mergeGitStatusParts(local, remote); }); const refreshLocalStatusCore = Effect.fn("VcsStatusBroadcaster.refreshLocalStatusCore")( @@ -361,46 +564,54 @@ export const make = Effect.gen(function* () { if (options?.refreshUpstream !== false) { yield* workflow.invalidateRemoteStatus(cwd); } + yield* getOrLoadLocalStatus(cwd); + const generation = (yield* getCachedStatus(cwd))?.generation ?? 0; const remote = yield* workflow.remoteStatus({ cwd }, options); - return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); + return yield* updateCachedRemoteStatus(cwd, remote, generation, { publish: true }); }); const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn( "VcsStatusBroadcaster.refreshStatus", )(function* (rawCwd) { const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); + const expected = yield* getCachedStatus(cwd); // invalidateStatus (not the two partial invalidations) so an explicit // refresh also bypasses GitManager's slow PR-lookup cache. yield* workflow.invalidateStatus(cwd); - const [local, remote] = yield* Effect.all( - [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd })], - { concurrency: "unbounded" }, - ); - return yield* updateCachedStatus(cwd, local, remote, { publish: true }); + const { local, remote } = yield* readCoherentStatus(cwd); + const updated = yield* updateCachedStatus(cwd, local, remote, { + publish: true, + expected, + }); + if (updated !== null) { + return updated; + } + return mergeGitStatusParts(local, remote); }); const makeRemoteRefreshLoop = ( cwd: string, demandCwdsRef: Ref.Ref>, + needsRefreshRef: Ref.Ref, + refreshRequests: Queue.Queue, automaticRemoteRefreshInterval: Effect.Effect, refreshImmediately: boolean, ) => { return Effect.gen(function* () { const consecutiveFailuresRef = yield* Ref.make(0); - const needsInitialRefreshRef = yield* Ref.make(refreshImmediately); const refreshRemoteStatusIfEnabled = Effect.gen(function* () { const configuredInterval = yield* automaticRemoteRefreshInterval; const activeInterval = Duration.isZero(configuredInterval) ? DEFAULT_VCS_STATUS_REFRESH_INTERVAL : configuredInterval; - const needsInitialRefresh = yield* Ref.get(needsInitialRefreshRef); - if (Duration.isZero(configuredInterval) && !needsInitialRefresh) { + const needsRefresh = yield* Ref.get(needsRefreshRef); + if (Duration.isZero(configuredInterval) && !needsRefresh) { return activeInterval; } const demandCwds = yield* Ref.get(demandCwdsRef); const shouldRun = - needsInitialRefresh || + needsRefresh || (yield* Effect.all( [...demandCwds.keys()].map((demandCwd) => backgroundPolicy.shouldRunScopeWork({ @@ -418,7 +629,9 @@ export const make = Effect.gen(function* () { refreshUpstream: !Duration.isZero(configuredInterval), }).pipe(Effect.exit); if (Exit.isSuccess(exit)) { - yield* Ref.set(needsInitialRefreshRef, false); + if (exit.value) { + yield* Ref.set(needsRefreshRef, false); + } yield* Ref.set(consecutiveFailuresRef, 0); return activeInterval; } @@ -444,21 +657,20 @@ export const make = Effect.gen(function* () { if (!refreshImmediately) { const configuredInterval = yield* automaticRemoteRefreshInterval; - yield* Effect.sleep( - Duration.isZero(configuredInterval) - ? DEFAULT_VCS_STATUS_REFRESH_INTERVAL - : configuredInterval, + yield* Effect.raceFirst( + Effect.sleep( + Duration.isZero(configuredInterval) + ? DEFAULT_VCS_STATUS_REFRESH_INTERVAL + : configuredInterval, + ), + Queue.take(refreshRequests), ); } - return yield* refreshRemoteStatusIfEnabled.pipe( - Effect.repeat( - Schedule.identity().pipe( - Schedule.addDelay(({ output: delay }) => Effect.succeed(delay)), - ), - ), - Effect.asVoid, - ); + while (true) { + const nextDelay = yield* refreshRemoteStatusIfEnabled; + yield* Effect.raceFirst(Effect.sleep(nextDelay), Queue.take(refreshRequests)); + } }); }; @@ -471,27 +683,36 @@ export const make = Effect.gen(function* () { yield* SynchronizedRef.modifyEffect(pollersRef, (activePollers) => { const existing = activePollers.get(cwd); if (existing) { - return Ref.update(existing.demandCwds, (demandCwds) => { - const next = new Map(demandCwds); - next.set(demandCwd, (next.get(demandCwd) ?? 0) + 1); - return next; - }).pipe( - Effect.map(() => { - const nextPollers = new Map(activePollers); - nextPollers.set(cwd, { - ...existing, - subscriberCount: existing.subscriberCount + 1, - }); - return [undefined, nextPollers] as const; - }), - ); + return Effect.gen(function* () { + yield* Ref.update(existing.demandCwds, (demandCwds) => { + const next = new Map(demandCwds); + next.set(demandCwd, (next.get(demandCwd) ?? 0) + 1); + return next; + }); + if (refreshImmediately) { + yield* Ref.set(existing.needsRefresh, true); + yield* Queue.offer(existing.refreshRequests, undefined); + } + const nextPollers = new Map(activePollers); + nextPollers.set(cwd, { + ...existing, + subscriberCount: existing.subscriberCount + 1, + }); + return [undefined, nextPollers] as const; + }); } - return Ref.make>(new Map([[demandCwd, 1]])).pipe( - Effect.flatMap((demandCwds) => + return Effect.all({ + demandCwds: Ref.make>(new Map([[demandCwd, 1]])), + needsRefresh: Ref.make(refreshImmediately), + refreshRequests: Queue.sliding(1), + }).pipe( + Effect.flatMap(({ demandCwds, needsRefresh, refreshRequests }) => makeRemoteRefreshLoop( cwd, demandCwds, + needsRefresh, + refreshRequests, automaticRemoteRefreshInterval, refreshImmediately, ).pipe( @@ -502,6 +723,8 @@ export const make = Effect.gen(function* () { fiber, subscriberCount: 1, demandCwds, + needsRefresh, + refreshRequests, }); return [undefined, nextPollers] as const; }), @@ -558,9 +781,15 @@ export const make = Effect.gen(function* () { Effect.gen(function* () { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); const subscription = yield* PubSub.subscribe(changesPubSub); - const initialLocal = yield* getOrLoadLocalStatus(cwd); + yield* getOrLoadLocalStatus(cwd); + yield* options?.beforeInitialSnapshot ?? Effect.void; const cachedStatus = yield* getCachedStatus(cwd); - const initialRemote = cachedStatus?.remote?.value ?? null; + const initialLocal = yield* cachedStatus?.local + ? Effect.succeed(cachedStatus.local.value) + : Effect.die("Local VCS status was unavailable after loading"); + const generation = cachedStatus?.generation ?? 0; + const initialRemote = + cachedStatus?.remote?.generation === generation ? cachedStatus.remote.value : null; yield* retainRemotePoller( cwd, input.cwd, @@ -574,6 +803,7 @@ export const make = Effect.gen(function* () { return Stream.concat( Stream.make({ _tag: "snapshot" as const, + generation, local: initialLocal, remote: initialRemote, }), @@ -593,4 +823,9 @@ export const make = Effect.gen(function* () { }); }); +export const make = makeWithOptions(); + +export const layerWithOptions = (options: MakeOptions) => + Layer.effect(VcsStatusBroadcaster, makeWithOptions(options)); + export const layer = Layer.effect(VcsStatusBroadcaster, make); diff --git a/apps/swift-ios/App/NativeFeatureClient.swift b/apps/swift-ios/App/NativeFeatureClient.swift index afcb76857fea..61159c8be68a 100644 --- a/apps/swift-ios/App/NativeFeatureClient.swift +++ b/apps/swift-ios/App/NativeFeatureClient.swift @@ -1870,6 +1870,40 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, } func sourceControlStatus(threadID: String) async throws -> FeatureSourceControlStatus { + let route = try threadRoute(for: threadID) + let context = try workspaceContext(route: route) + let cached = try await withThrowingTaskGroup(of: FeatureSourceControlStatus?.self) { group in + group.addTask { + let events = await route.client.vcsStatusEvents(cwd: context.cwd) + var accumulator = NativeSourceControlStatusAccumulator() + do { + for try await event in events { + if let status = accumulator.consume(event) { return status } + } + } catch is CancellationError { + throw CancellationError() + } catch { + try Task.checkCancellation() + return nil + } + try Task.checkCancellation() + return nil + } + group.addTask { + try await Task.sleep(for: .seconds(2)) + return nil + } + + let result = try await group.next() ?? nil + group.cancelAll() + return result + } + if let cached { return cached } + try Task.checkCancellation() + return try await refreshSourceControlStatus(threadID: threadID) + } + + func refreshSourceControlStatus(threadID: String) async throws -> FeatureSourceControlStatus { let route = try threadRoute(for: threadID) let context = try workspaceContext(route: route) return NativeWorkspaceMapper.sourceControl( diff --git a/apps/swift-ios/App/NativeWorkspaceMapper.swift b/apps/swift-ios/App/NativeWorkspaceMapper.swift index 8a7ea4ff616f..edc1444d5384 100644 --- a/apps/swift-ios/App/NativeWorkspaceMapper.swift +++ b/apps/swift-ios/App/NativeWorkspaceMapper.swift @@ -90,6 +90,30 @@ enum NativeWorkspaceMapper { ) } + static func sourceControl( + local: VCSLocalStatus, + remote: VCSRemoteStatus? + ) -> FeatureSourceControlStatus? { + guard !local.hasPrimaryRemote || remote != nil else { return nil } + return FeatureSourceControlStatus( + isRepository: local.isRepo, + branch: local.refName, + aheadCount: remote?.aheadCount ?? 0, + behindCount: remote?.behindCount ?? 0, + files: local.workingTree.files.map { + FeatureSourceControlFile(path: $0.path, state: .modified, isStaged: false) + }, + pullRequest: remote?.pr.map { + FeaturePullRequest( + number: $0.number, + title: $0.title, + state: $0.state, + url: URL(string: $0.url) + ) + } + ) + } + static func gitAction(_ action: FeatureSourceControlAction) -> GitStackedAction { switch action { case .commit: .commit @@ -342,3 +366,47 @@ enum NativeWorkspaceMapper { ) } } + +struct NativeSourceControlStatusAccumulator { + private var generation: Int? + private var acceptsLegacyRemote = false + private var local: VCSLocalStatus? + private var remote: VCSRemoteStatus? + + mutating func consume(_ event: VCSStatusEvent) -> FeatureSourceControlStatus? { + switch event { + case let .snapshot(nextGeneration, nextLocal, nextRemote): + if let generation, let nextGeneration, nextGeneration < generation { break } + let canStartLegacyStream = local == nil || acceptsLegacyRemote + generation = nextGeneration + acceptsLegacyRemote = nextGeneration == nil && canStartLegacyStream + local = nextLocal + remote = nextRemote + case let .localUpdated(nextGeneration, nextLocal): + if let generation, let nextGeneration, nextGeneration < generation { break } + if nextGeneration == nil + || generation != nextGeneration + || local?.refName != nextLocal.refName { + remote = nil + } + generation = nextGeneration + if nextGeneration != nil { + acceptsLegacyRemote = false + } else if local == nil { + acceptsLegacyRemote = true + } + local = nextLocal + case let .remoteUpdated(nextGeneration, nextRemote): + if let nextGeneration { + acceptsLegacyRemote = false + guard generation == nextGeneration else { break } + } else { + guard acceptsLegacyRemote else { break } + } + remote = nextRemote + } + + guard let local else { return nil } + return NativeWorkspaceMapper.sourceControl(local: local, remote: remote) + } +} diff --git a/apps/swift-ios/Core/WorkspaceModels.swift b/apps/swift-ios/Core/WorkspaceModels.swift index 8df7221c2c54..13797f6d752b 100644 --- a/apps/swift-ios/Core/WorkspaceModels.swift +++ b/apps/swift-ios/Core/WorkspaceModels.swift @@ -358,11 +358,11 @@ public struct VCSStatus: Codable, Equatable, Sendable { } public enum VCSStatusEvent: Decodable, Sendable { - case snapshot(local: VCSLocalStatus, remote: VCSRemoteStatus?) - case localUpdated(VCSLocalStatus) - case remoteUpdated(VCSRemoteStatus?) + case snapshot(generation: Int?, local: VCSLocalStatus, remote: VCSRemoteStatus?) + case localUpdated(generation: Int?, local: VCSLocalStatus) + case remoteUpdated(generation: Int?, remote: VCSRemoteStatus?) - private enum CodingKeys: String, CodingKey { case _tag, local, remote } + private enum CodingKeys: String, CodingKey { case _tag, generation, local, remote } public init(from decoder: any Decoder) throws { let container = try decoder.container(keyedBy: CodingKeys.self) @@ -370,14 +370,19 @@ public enum VCSStatusEvent: Decodable, Sendable { switch tag { case "snapshot": self = .snapshot( + generation: try container.decodeIfPresent(Int.self, forKey: .generation), local: try container.decode(VCSLocalStatus.self, forKey: .local), remote: try container.decodeIfPresent(VCSRemoteStatus.self, forKey: .remote) ) case "localUpdated": - self = .localUpdated(try container.decode(VCSLocalStatus.self, forKey: .local)) + self = .localUpdated( + generation: try container.decodeIfPresent(Int.self, forKey: .generation), + local: try container.decode(VCSLocalStatus.self, forKey: .local) + ) case "remoteUpdated": self = .remoteUpdated( - try container.decodeIfPresent(VCSRemoteStatus.self, forKey: .remote) + generation: try container.decodeIfPresent(Int.self, forKey: .generation), + remote: try container.decodeIfPresent(VCSRemoteStatus.self, forKey: .remote) ) default: throw DecodingError.dataCorruptedError( diff --git a/apps/swift-ios/Features/Shared/FeatureClient.swift b/apps/swift-ios/Features/Shared/FeatureClient.swift index dc282ed4de29..7336fe5eb55b 100644 --- a/apps/swift-ios/Features/Shared/FeatureClient.swift +++ b/apps/swift-ios/Features/Shared/FeatureClient.swift @@ -116,6 +116,7 @@ public protocol FeatureClient: AnyObject { ) async throws -> FeatureReviewFileContents? func sourceControlStatus(threadID: String) async throws -> FeatureSourceControlStatus + func refreshSourceControlStatus(threadID: String) async throws -> FeatureSourceControlStatus func performSourceControlAction( threadID: String, action: FeatureSourceControlAction, @@ -330,6 +331,10 @@ public extension FeatureClient { throw FeatureCapabilityUnavailable("Source control") } + func refreshSourceControlStatus(threadID: String) async throws -> FeatureSourceControlStatus { + try await sourceControlStatus(threadID: threadID) + } + func performSourceControlAction( threadID: String, action: FeatureSourceControlAction, diff --git a/apps/swift-ios/Features/SourceControl/FeatureSourceControlView.swift b/apps/swift-ios/Features/SourceControl/FeatureSourceControlView.swift index ea26b9fcd97b..6b35164639c3 100644 --- a/apps/swift-ios/Features/SourceControl/FeatureSourceControlView.swift +++ b/apps/swift-ios/Features/SourceControl/FeatureSourceControlView.swift @@ -41,7 +41,9 @@ public struct FeatureSourceControlView: View { .navigationBarTitleDisplayMode(.inline) .toolbar { ToolbarItem(placement: .topBarTrailing) { - Button { Task { await load() } } label: { Image(systemName: "arrow.clockwise") } + Button { Task { await load(refresh: true) } } label: { + Image(systemName: "arrow.clockwise") + } .disabled(isLoading || isRunningAction) .accessibilityLabel("Reload source control") } @@ -60,7 +62,7 @@ public struct FeatureSourceControlView: View { } .disabled(commitMessage.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty) } - .task { await load() } + .task { await load(refresh: false) } } private func statusList(_ status: FeatureSourceControlStatus) -> some View { @@ -131,7 +133,7 @@ public struct FeatureSourceControlView: View { } .listStyle(.insetGrouped) .scrollContentBackground(.hidden) - .refreshable { await load() } + .refreshable { await load(refresh: true) } .overlay { if isRunningAction { ProgressView() @@ -150,11 +152,15 @@ public struct FeatureSourceControlView: View { } } - private func load() async { + private func load(refresh: Bool) async { isLoading = true defer { isLoading = false } do { - status = try await client.sourceControlStatus(threadID: threadID) + status = if refresh { + try await client.refreshSourceControlStatus(threadID: threadID) + } else { + try await client.sourceControlStatus(threadID: threadID) + } errorMessage = nil } catch { errorMessage = error.localizedDescription diff --git a/apps/swift-ios/Features/Workspace/DailyUXModels.swift b/apps/swift-ios/Features/Workspace/DailyUXModels.swift index 9fb6ba598164..a367d1d17442 100644 --- a/apps/swift-ios/Features/Workspace/DailyUXModels.swift +++ b/apps/swift-ios/Features/Workspace/DailyUXModels.swift @@ -635,9 +635,12 @@ extension FeatureThread { .first(where: { $0.id == projectID })? .environmentID let resolvedEnvironmentID = environmentID ?? projectEnvironmentID - let providers = resolvedEnvironmentID.flatMap { - snapshot.providersByEnvironment?[$0] - } ?? [] + let providers: [FeatureProvider] + if let providersByEnvironment = snapshot.providersByEnvironment { + providers = resolvedEnvironmentID.flatMap { providersByEnvironment[$0] } ?? [] + } else { + providers = snapshot.providers + } return providers.first(where: { $0.id == providerID })?.name ?? providerID } diff --git a/apps/swift-ios/Features/Workspace/WorkspaceView.swift b/apps/swift-ios/Features/Workspace/WorkspaceView.swift index 34c82e049c2c..73a8007bedf3 100644 --- a/apps/swift-ios/Features/Workspace/WorkspaceView.swift +++ b/apps/swift-ios/Features/Workspace/WorkspaceView.swift @@ -780,9 +780,12 @@ struct HomeThreadRowContext: Equatable { let explicitProvider = thread.providerName? .trimmingCharacters(in: .whitespacesAndNewlines) let configuredProvider = thread.providerID.flatMap { providerID in - environmentID.flatMap { - snapshot.providersByEnvironment?[$0]?.first(where: { $0.id == providerID }) + if let providersByEnvironment = snapshot.providersByEnvironment { + return environmentID.flatMap { + providersByEnvironment[$0]?.first(where: { $0.id == providerID }) + } } + return snapshot.providers.first(where: { $0.id == providerID }) } let providerName = (explicitProvider?.isEmpty == false ? explicitProvider : nil) ?? configuredProvider?.name diff --git a/apps/swift-ios/Tests/CoreTests/WorkspaceContractTests.swift b/apps/swift-ios/Tests/CoreTests/WorkspaceContractTests.swift index 39cc9653818a..79019a2db2f3 100644 --- a/apps/swift-ios/Tests/CoreTests/WorkspaceContractTests.swift +++ b/apps/swift-ios/Tests/CoreTests/WorkspaceContractTests.swift @@ -8,6 +8,7 @@ final class WorkspaceContractTests: XCTestCase { """ { "_tag": "snapshot", + "generation": 7, "local": { "isRepo": true, "sourceControlProvider": { @@ -37,14 +38,38 @@ final class WorkspaceContractTests: XCTestCase { ) let event = try JSONDecoder.t3.decode(VCSStatusEvent.self, from: data) - guard case let .snapshot(local, remote) = event else { + guard case let .snapshot(generation, local, remote) = event else { return XCTFail("Expected snapshot") } + XCTAssertEqual(generation, 7) XCTAssertEqual(local.refName, "feat/swift") XCTAssertEqual(local.workingTree.files.first?.insertions, 12) XCTAssertEqual(remote?.aheadCount, 1) } + func testVCSStatusEventDecodesLegacyGenerationlessShape() throws { + let data = Data( + """ + { + "_tag": "remoteUpdated", + "remote": { + "hasUpstream": true, + "aheadCount": 1, + "behindCount": 0, + "pr": null + } + } + """.utf8 + ) + + let event = try JSONDecoder.t3.decode(VCSStatusEvent.self, from: data) + guard case let .remoteUpdated(generation, remote) = event else { + return XCTFail("Expected remoteUpdated") + } + XCTAssertNil(generation) + XCTAssertEqual(remote?.aheadCount, 1) + } + func testTerminalAttachEventsDecodeSnapshotAndOutputShapes() throws { let snapshotData = Data( """ diff --git a/apps/swift-ios/Tests/FeatureTests/FeatureToolStateTests.swift b/apps/swift-ios/Tests/FeatureTests/FeatureToolStateTests.swift index 871f08d926b8..92507f14cc7e 100644 --- a/apps/swift-ios/Tests/FeatureTests/FeatureToolStateTests.swift +++ b/apps/swift-ios/Tests/FeatureTests/FeatureToolStateTests.swift @@ -3,6 +3,141 @@ import Testing @Suite("Thread tool state") struct FeatureToolStateTests { + private func vcsLocal(refName: String, hasPrimaryRemote: Bool = true) -> VCSLocalStatus { + VCSLocalStatus( + isRepo: true, + sourceControlProvider: nil, + hasPrimaryRemote: hasPrimaryRemote, + isDefaultRef: false, + refName: refName, + hasWorkingTreeChanges: false, + workingTree: VCSWorkingTree(files: [], insertions: 0, deletions: 0) + ) + } + + private func vcsRemote(aheadCount: Int) -> VCSRemoteStatus { + VCSRemoteStatus( + hasUpstream: true, + aheadCount: aheadCount, + behindCount: 0, + aheadOfDefaultCount: nil, + pr: nil + ) + } + + @Test + func cachedSourceControlWaitsForRemoteWhenRepositoryHasPrimaryRemote() { + var accumulator = NativeSourceControlStatusAccumulator() + + let status = accumulator.consume( + .snapshot(generation: 1, local: vcsLocal(refName: "feature/a"), remote: nil) + ) + + #expect(status == nil) + } + + @Test + func cachedSourceControlRejectsLateRemoteFromPreviousRef() { + var accumulator = NativeSourceControlStatusAccumulator() + _ = accumulator.consume( + .snapshot( + generation: 1, + local: vcsLocal(refName: "feature/a"), + remote: vcsRemote(aheadCount: 4) + ) + ) + _ = accumulator.consume( + .localUpdated(generation: 2, local: vcsLocal(refName: "feature/b")) + ) + + let status = accumulator.consume( + .remoteUpdated(generation: 1, remote: vcsRemote(aheadCount: 4)) + ) + + #expect(status == nil) + } + + @Test + func cachedSourceControlClearsRemoteWhenRefChangesWithinGeneration() { + var accumulator = NativeSourceControlStatusAccumulator() + _ = accumulator.consume( + .snapshot( + generation: 1, + local: vcsLocal(refName: "feature/a"), + remote: vcsRemote(aheadCount: 4) + ) + ) + + let status = accumulator.consume( + .localUpdated(generation: 1, local: vcsLocal(refName: "feature/b")) + ) + + #expect(status == nil) + } + + @Test + func cachedSourceControlAcceptsRemoteForCurrentRefGeneration() throws { + var accumulator = NativeSourceControlStatusAccumulator() + _ = accumulator.consume( + .snapshot(generation: 2, local: vcsLocal(refName: "feature/b"), remote: nil) + ) + + let consumed = accumulator.consume( + .remoteUpdated(generation: 2, remote: vcsRemote(aheadCount: 2)) + ) + let status = try #require(consumed) + + #expect(status.branch == "feature/b") + #expect(status.aheadCount == 2) + } + + @Test + func legacySourceControlAcceptsGenerationlessRemoteAfterLocalDelta() { + var accumulator = NativeSourceControlStatusAccumulator() + _ = accumulator.consume( + .snapshot( + generation: nil, + local: vcsLocal(refName: "feature/a"), + remote: nil + ) + ) + let initialRemote = accumulator.consume( + .remoteUpdated(generation: nil, remote: vcsRemote(aheadCount: 4)) + ) + let localOnly = accumulator.consume( + .localUpdated(generation: nil, local: vcsLocal(refName: "feature/b")) + ) + let afterRemote = accumulator.consume( + .remoteUpdated(generation: nil, remote: vcsRemote(aheadCount: 99)) + ) + + #expect(initialRemote?.aheadCount == 4) + #expect(localOnly == nil) + #expect(afterRemote?.branch == "feature/b") + #expect(afterRemote?.aheadCount == 99) + } + + @Test + func sourceControlStopsAcceptingLegacyRemoteAfterGenerationAppears() { + var accumulator = NativeSourceControlStatusAccumulator() + _ = accumulator.consume( + .snapshot( + generation: nil, + local: vcsLocal(refName: "feature/a"), + remote: nil + ) + ) + let mismatchedGeneration = accumulator.consume( + .remoteUpdated(generation: 1, remote: vcsRemote(aheadCount: 4)) + ) + let legacyRemote = accumulator.consume( + .remoteUpdated(generation: nil, remote: vcsRemote(aheadCount: 99)) + ) + + #expect(mismatchedGeneration == nil) + #expect(legacyRemote == nil) + } + @Test func fileFilteringKeepsDirectoriesFirstAndHonorsHiddenFiles() { let entries = [ diff --git a/apps/swift-ios/Tests/FeatureTests/HomeThreadMetadataTests.swift b/apps/swift-ios/Tests/FeatureTests/HomeThreadMetadataTests.swift index c63745ee727a..9802f896b472 100644 --- a/apps/swift-ios/Tests/FeatureTests/HomeThreadMetadataTests.swift +++ b/apps/swift-ios/Tests/FeatureTests/HomeThreadMetadataTests.swift @@ -195,4 +195,36 @@ struct HomeThreadMetadataTests { ) == nil ) } + + @Test + func environmentCatalogMissDoesNotUseLegacyProviders() throws { + let thread = FeatureThread( + id: "thread", + projectID: "project", + title: "Use provider", + providerID: "shared-id" + ) + let snapshot = FeatureSnapshot( + projects: [ + FeatureProject( + id: "project", + environmentID: "remote", + name: "Remote", + path: "/remote" + ), + ], + threads: [thread], + providers: [ + FeatureProvider(id: "shared-id", name: "Wrong environment", driver: "codex"), + ], + providersByEnvironment: ["local": [ + FeatureProvider(id: "shared-id", name: "Local only", driver: "codex"), + ]] + ) + + #expect(thread.homeProviderLabel(in: snapshot) == "shared-id") + let context = try #require(HomeThreadRowContext.index(snapshot: snapshot)[thread.id]) + #expect(context.providerName == "shared-id") + #expect(context.providerDriver == "shared-id") + } } diff --git a/packages/client-runtime/src/state/vcs.ts b/packages/client-runtime/src/state/vcs.ts index a0d4510be7f5..a17c57cda315 100644 --- a/packages/client-runtime/src/state/vcs.ts +++ b/packages/client-runtime/src/state/vcs.ts @@ -2,10 +2,9 @@ import { type EnvironmentId, type VcsListRefsInput, type VcsListRefsResult, - type VcsStatusResult, WS_METHODS, } from "@t3tools/contracts"; -import { applyGitStatusStreamEvent } from "@t3tools/shared/git"; +import { applyGitStatusStreamState } from "@t3tools/shared/git"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; @@ -275,10 +274,10 @@ export function createVcsEnvironmentAtoms( subscribe: (input: EnvironmentRpcInput) => subscribe(WS_METHODS.subscribeVcsStatus, input).pipe( Stream.mapAccum( - () => null as VcsStatusResult | null, + () => null as ReturnType, (current, event) => { - const next = applyGitStatusStreamEvent(current, event); - return [next, [next]] as const; + const next = applyGitStatusStreamState(current, event); + return [next, next === null ? [] : [next.status]] as const; }, ), ), diff --git a/packages/contracts/src/git.test.ts b/packages/contracts/src/git.test.ts index 4ea86670ff8f..e6822811a1de 100644 --- a/packages/contracts/src/git.test.ts +++ b/packages/contracts/src/git.test.ts @@ -7,6 +7,7 @@ import { GitRunStackedActionResult, GitRunStackedActionInput, GitResolvePullRequestResult, + VcsStatusStreamEvent, } from "./git.ts"; const decodeCreateWorktreeInput = Schema.decodeUnknownSync(VcsCreateWorktreeInput); @@ -16,6 +17,25 @@ const decodePreparePullRequestThreadInput = Schema.decodeUnknownSync( const decodeRunStackedActionInput = Schema.decodeUnknownSync(GitRunStackedActionInput); const decodeRunStackedActionResult = Schema.decodeUnknownSync(GitRunStackedActionResult); const decodeResolvePullRequestResult = Schema.decodeUnknownSync(GitResolvePullRequestResult); +const decodeStatusStreamEvent = Schema.decodeUnknownSync(VcsStatusStreamEvent); + +describe("VcsStatusStreamEvent", () => { + it("decodes legacy and generation-aware status updates", () => { + expect( + decodeStatusStreamEvent({ + _tag: "remoteUpdated", + remote: null, + }), + ).toEqual({ _tag: "remoteUpdated", remote: null }); + expect( + decodeStatusStreamEvent({ + _tag: "remoteUpdated", + generation: 3, + remote: null, + }), + ).toEqual({ _tag: "remoteUpdated", generation: 3, remote: null }); + }); +}); describe("VcsCreateWorktreeInput", () => { it("accepts omitted newRefName for existing-refName worktrees", () => { diff --git a/packages/contracts/src/git.ts b/packages/contracts/src/git.ts index 2e0552740a6c..f1e05899d21d 100644 --- a/packages/contracts/src/git.ts +++ b/packages/contracts/src/git.ts @@ -205,6 +205,8 @@ const VcsStatusLocalShape = { hasPrimaryRemote: Schema.Boolean, isDefaultRef: Schema.Boolean, refName: Schema.NullOr(TrimmedNonEmptyStringSchema), + coherenceToken: Schema.optional(TrimmedNonEmptyStringSchema), + remoteAssociationToken: Schema.optional(TrimmedNonEmptyStringSchema), hasWorkingTreeChanges: Schema.Boolean, workingTree: Schema.Struct({ files: Schema.Array( @@ -241,13 +243,16 @@ export type VcsStatusResult = typeof VcsStatusResult.Type; export const VcsStatusStreamEvent = Schema.Union([ Schema.TaggedStruct("snapshot", { + generation: Schema.optional(NonNegativeInt), local: VcsStatusLocalResult, remote: Schema.NullOr(VcsStatusRemoteResult), }), Schema.TaggedStruct("localUpdated", { + generation: Schema.optional(NonNegativeInt), local: VcsStatusLocalResult, }), Schema.TaggedStruct("remoteUpdated", { + generation: Schema.optional(NonNegativeInt), remote: Schema.NullOr(VcsStatusRemoteResult), }), ]); diff --git a/packages/shared/src/git.test.ts b/packages/shared/src/git.test.ts index 96539f0aae24..5e576a5a708e 100644 --- a/packages/shared/src/git.test.ts +++ b/packages/shared/src/git.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from "vite-plus/test"; import { applyGitStatusStreamEvent, + applyGitStatusStreamState, buildTemporaryWorktreeBranchName, isTemporaryWorktreeBranch, normalizeGitRemoteUrl, @@ -110,7 +111,9 @@ describe("applyGitStatusStreamEvent", () => { pr: null, }; - expect(applyGitStatusStreamEvent(null, { _tag: "remoteUpdated", remote })).toEqual({ + expect( + applyGitStatusStreamEvent(null, { _tag: "remoteUpdated", generation: 0, remote }), + ).toEqual({ isRepo: true, hasPrimaryRemote: false, isDefaultRef: false, @@ -154,7 +157,9 @@ describe("applyGitStatusStreamEvent", () => { pr: null, }; - expect(applyGitStatusStreamEvent(current, { _tag: "remoteUpdated", remote })).toEqual({ + expect( + applyGitStatusStreamEvent(current, { _tag: "remoteUpdated", generation: 0, remote }), + ).toEqual({ ...current, hasUpstream: true, aheadCount: 2, @@ -163,3 +168,117 @@ describe("applyGitStatusStreamEvent", () => { }); }); }); + +describe("applyGitStatusStreamState", () => { + const local = { + isRepo: true, + hasPrimaryRemote: true, + isDefaultRef: false, + refName: "feature/new-ref", + hasWorkingTreeChanges: false, + workingTree: { files: [], insertions: 0, deletions: 0 }, + } as const; + const remote: VcsStatusRemoteResult = { + hasUpstream: true, + aheadCount: 2, + behindCount: 1, + pr: null, + }; + + it("clears remote state when a newer local generation arrives", () => { + const initial = applyGitStatusStreamState(null, { + _tag: "snapshot", + generation: 1, + local: { ...local, refName: "feature/old-ref" }, + remote, + }); + const updated = applyGitStatusStreamState(initial, { + _tag: "localUpdated", + generation: 2, + local, + }); + + expect(updated?.status.refName).toBe("feature/new-ref"); + expect(updated?.status.aheadCount).toBe(0); + expect(updated?.status.pr).toBeNull(); + }); + + it("rejects a late remote update from an older generation", () => { + const current = applyGitStatusStreamState(null, { + _tag: "snapshot", + generation: 2, + local, + remote: null, + }); + + expect( + applyGitStatusStreamState(current, { + _tag: "remoteUpdated", + generation: 1, + remote, + }), + ).toEqual(current); + }); + + it("accepts a remote update for the current generation", () => { + const current = applyGitStatusStreamState(null, { + _tag: "snapshot", + generation: 2, + local, + remote: null, + }); + const updated = applyGitStatusStreamState(current, { + _tag: "remoteUpdated", + generation: 2, + remote, + }); + + expect(updated?.status.aheadCount).toBe(2); + expect(updated?.status.behindCount).toBe(1); + }); + + it("keeps accepting generation-less remote updates after a legacy local delta", () => { + const initial = applyGitStatusStreamState(null, { + _tag: "snapshot", + local, + remote: null, + }); + const initialRemote = applyGitStatusStreamState(initial, { + _tag: "remoteUpdated", + remote, + }); + const localOnly = applyGitStatusStreamState(initialRemote, { + _tag: "localUpdated", + local: { ...local, refName: "feature/legacy-next" }, + }); + const afterRemote = applyGitStatusStreamState(localOnly, { + _tag: "remoteUpdated", + remote: { ...remote, aheadCount: 99 }, + }); + + expect(initial?.generation).toBeNull(); + expect(initialRemote?.status.aheadCount).toBe(2); + expect(localOnly?.status.aheadCount).toBe(0); + expect(afterRemote?.status.aheadCount).toBe(99); + }); + + it("stops accepting generation-less remote updates after a generation-bearing event", () => { + const initial = applyGitStatusStreamState(null, { + _tag: "snapshot", + local, + remote: null, + }); + const generationSeen = applyGitStatusStreamState(initial, { + _tag: "remoteUpdated", + generation: 1, + remote, + }); + const legacyRemote = applyGitStatusStreamState(generationSeen, { + _tag: "remoteUpdated", + remote: { ...remote, aheadCount: 99 }, + }); + + expect(generationSeen?.acceptsLegacyRemote).toBe(false); + expect(legacyRemote).toEqual(generationSeen); + }); +}); diff --git a/packages/shared/src/git.ts b/packages/shared/src/git.ts index 71fe2e806cfc..48578100d942 100644 --- a/packages/shared/src/git.ts +++ b/packages/shared/src/git.ts @@ -252,6 +252,10 @@ function toLocalStatusPart(status: VcsStatusResult): VcsStatusLocalResult { hasPrimaryRemote: status.hasPrimaryRemote, isDefaultRef: status.isDefaultRef, refName: status.refName, + ...(status.coherenceToken === undefined ? {} : { coherenceToken: status.coherenceToken }), + ...(status.remoteAssociationToken === undefined + ? {} + : { remoteAssociationToken: status.remoteAssociationToken }), hasWorkingTreeChanges: status.hasWorkingTreeChanges, workingTree: status.workingTree, }; @@ -283,3 +287,64 @@ export function applyGitStatusStreamEvent( return mergeGitStatusParts(toLocalStatusPart(current), event.remote); } } + +export interface GitStatusStreamState { + readonly generation: number | null; + readonly acceptsLegacyRemote: boolean; + readonly status: VcsStatusResult; +} + +export function applyGitStatusStreamState( + current: GitStatusStreamState | null, + event: VcsStatusStreamEvent, +): GitStatusStreamState | null { + if ( + current !== null && + current.generation !== null && + event.generation !== undefined && + event.generation < current.generation + ) { + return current; + } + + switch (event._tag) { + case "snapshot": + return { + generation: event.generation ?? null, + acceptsLegacyRemote: + event.generation === undefined && (current === null || current.acceptsLegacyRemote), + status: mergeGitStatusParts(event.local, event.remote), + }; + case "localUpdated": { + const remote = + event.generation !== undefined && + current !== null && + current.generation === event.generation + ? toRemoteStatusPart(current.status) + : null; + return { + generation: event.generation ?? null, + acceptsLegacyRemote: + event.generation === undefined && (current === null || current.acceptsLegacyRemote), + status: mergeGitStatusParts(event.local, remote), + }; + } + case "remoteUpdated": { + if ( + current === null || + (event.generation === undefined + ? !current.acceptsLegacyRemote + : current.generation !== event.generation) + ) { + return current?.acceptsLegacyRemote === true && event.generation !== undefined + ? { ...current, acceptsLegacyRemote: false } + : current; + } + return { + generation: event.generation ?? null, + acceptsLegacyRemote: event.generation === undefined && current.acceptsLegacyRemote, + status: mergeGitStatusParts(toLocalStatusPart(current.status), event.remote), + }; + } + } +}