| author | |
| committer | |
| log | 999161ef844a7a38c8a6d08195c1240759169832 |
| tree | 90e91380b09d0523baa8eded102461827f4ef42a |
| parent | ee18e56f902f2ccd74dffdc50363051be02ae0ab |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>5 files changed, 100 insertions(+), 36 deletions(-)
apps/server/src/orchestration/ThreadPullRequestReactor.test.ts+45| ... | ... | @@ -406,6 +406,51 @@ describe("ThreadPullRequestReactor", () => { |
| 406 | 406 | ), |
| 407 | 407 | ); |
| 408 | 408 | |
| 409 | it.effect("refreshes the project identity when a turn adds the remote", () => | |
| 410 | Effect.scoped( | |
| 411 | Effect.gen(function* () { | |
| 412 | const current = thread("new-remote"); | |
| 413 | const fixture = yield* makeHarness({ | |
| 414 | threads: [current], | |
| 415 | project: { ...project, repositoryIdentity: null }, | |
| 416 | branchPullRequest: () => Effect.succeed(branchPullRequest()), | |
| 417 | resolveRepositoryIdentity: (_cwd, options) => | |
| 418 | Effect.succeed(options?.refresh ? project.repositoryIdentity : null), | |
| 419 | }); | |
| 420 | yield* Effect.gen(function* () { | |
| 421 | const reactor = yield* fixture.start(); | |
| 422 | expect(yield* Ref.get(fixture.commands)).toHaveLength(0); | |
| 423 | ||
| 424 | yield* fixture.publish({ | |
| 425 | type: "thread.turn-diff-completed", | |
| 426 | sequence: 2, | |
| 427 | eventId: EventId.make("checkpoint-finished"), | |
| 428 | aggregateKind: "thread", | |
| 429 | aggregateId: current.id, | |
| 430 | occurredAt: NOW, | |
| 431 | commandId: null, | |
| 432 | causationEventId: null, | |
| 433 | correlationId: null, | |
| 434 | metadata: {}, | |
| 435 | payload: { | |
| 436 | threadId: current.id, | |
| 437 | turnId: TurnId.make("turn"), | |
| 438 | checkpointTurnCount: 1, | |
| 439 | checkpointRef: CheckpointRef.make("checkpoint"), | |
| 440 | status: "ready", | |
| 441 | files: [], | |
| 442 | assistantMessageId: null, | |
| 443 | completedAt: NOW, | |
| 444 | }, | |
| 445 | }); | |
| 446 | yield* Queue.take(fixture.reads); | |
| 447 | yield* reactor.drain; | |
| 448 | expect((yield* Ref.get(fixture.commands))[0]?.branchPullRequest).toEqual(reference(42)); | |
| 449 | }).pipe(Effect.provide(fixture.layer)); | |
| 450 | }), | |
| 451 | ), | |
| 452 | ); | |
| 453 | ||
| 409 | 454 | it.effect("uses live worktrees and falls back to the project for removed worktrees", () => |
| 410 | 455 | Effect.scoped( |
| 411 | 456 | Effect.gen(function* () { |
apps/server/src/orchestration/ThreadPullRequestReactor.ts+13-2| ... | ... | @@ -164,8 +164,19 @@ export const make = Effect.gen(function* () { |
| 164 | 164 | (group) => |
| 165 | 165 | Effect.gen(function* () { |
| 166 | 166 | const first = group[0]!; |
| 167 | const project = projects.get(first.projectId); | |
| 168 | if (project === undefined) return finishBackfill(group); | |
| 167 | const snapshotProject = projects.get(first.projectId); | |
| 168 | if (snapshotProject === undefined) return finishBackfill(group); | |
| 169 | // A finished turn may have added the remote this PR lives on. A failed | |
| 170 | // refresh resolves to null, so keep the snapshot's identity then. | |
| 171 | const project = request.refresh | |
| 172 | ? { | |
| 173 | ...snapshotProject, | |
| 174 | repositoryIdentity: | |
| 175 | (yield* repositoryIdentities.resolve(snapshotProject.workspaceRoot, { | |
| 176 | refresh: true, | |
| 177 | })) ?? snapshotProject.repositoryIdentity, | |
| 178 | } | |
| 179 | : snapshotProject; | |
| 169 | 180 | const repository = sourceControlRepositorySelector(project.repositoryIdentity); |
| 170 | 181 | if (first.branch !== null && repository === null) return finishBackfill(group); |
| 171 | 182 | const worktreeExists = |
apps/server/src/project/RepositoryIdentityResolver.test.ts+7-3| ... | ... | @@ -94,6 +94,8 @@ it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => { |
| 94 | 94 | const resolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver; |
| 95 | 95 | const first = yield* resolver.resolve("/repo/packages/web"); |
| 96 | 96 | rootPath = "/repo/packages/web"; |
| 97 | // Longer than the one-minute cadence of the background sweeps. | |
| 98 | yield* TestClock.adjust(Duration.minutes(10)); | |
| 97 | 99 | const second = yield* resolver.resolve("/repo/packages/web"); |
| 98 | 100 | |
| 99 | 101 | expect(first?.canonicalKey).toBe("github.com/t3tools/t3code"); |
| ... | ... | @@ -123,10 +125,10 @@ it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => { |
| 123 | 125 | const unavailable = yield* resolver.resolve(rootPath, { refresh: true }); |
| 124 | 126 | expect(unavailable?.webUrl).toBeUndefined(); |
| 125 | 127 | expect(unavailable?.canonicalKey).toBe("ssh.forge.test/team/repo"); |
| 126 | }).pipe(Effect.provide(resolverLayer)); | |
| 128 | }).pipe(Effect.provide(Layer.merge(TestClock.layer(), resolverLayer))); | |
| 127 | 129 | }); |
| 128 | 130 | |
| 129 | it.effect("retries Git root discovery after a failed lookup", () => { | |
| 131 | it.effect("retries Git root discovery after the negative TTL", () => { | |
| 130 | 132 | const calls: Array<ReadonlyArray<string>> = []; |
| 131 | 133 | let rootAttempts = 0; |
| 132 | 134 | const processRunner = Layer.succeed(ProcessRunner.ProcessRunner, { |
| ... | ... | @@ -159,7 +161,9 @@ it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => { |
| 159 | 161 | return Effect.gen(function* () { |
| 160 | 162 | const resolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver; |
| 161 | 163 | expect(yield* resolver.resolve("/repo/packages/web")).toBeNull(); |
| 164 | expect(yield* resolver.resolve("/repo/packages/web")).toBeNull(); | |
| 162 | 165 | |
| 166 | yield* TestClock.adjust(Duration.minutes(1)); | |
| 163 | 167 | const recovered = yield* resolver.resolve("/repo/packages/web"); |
| 164 | 168 | expect(recovered?.rootPath).toBe("/repo"); |
| 165 | 169 | expect(calls).toEqual([ |
| ... | ... | @@ -167,7 +171,7 @@ it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => { |
| 167 | 171 | ["-C", "/repo/packages/web", "rev-parse", "--show-toplevel"], |
| 168 | 172 | ["-C", "/repo", "remote", "-v"], |
| 169 | 173 | ]); |
| 170 | }).pipe(Effect.provide(resolverLayer)); | |
| 174 | }).pipe(Effect.provide(Layer.merge(TestClock.layer(), resolverLayer))); | |
| 171 | 175 | }); |
| 172 | 176 | |
| 173 | 177 | it.effect("normalizes equivalent GitHub remotes into a stable repository identity", () => |
apps/server/src/project/RepositoryIdentityResolver.ts+28-28| ... | ... | @@ -13,7 +13,11 @@ import * as Layer from "effect/Layer"; |
| 13 | 13 | import * as ProcessRunner from "../processRunner.ts"; |
| 14 | 14 | |
| 15 | 15 | const DEFAULT_REPOSITORY_IDENTITY_CACHE_CAPACITY = 512; |
| 16 | const DEFAULT_POSITIVE_CACHE_TTL = Duration.minutes(1); | |
| 16 | // Background sweeps resolve every project each minute. A long TTL keeps them | |
| 17 | // from spawning git each time. Clone, publish, and PR discovery (after a turn | |
| 18 | // and before it saves links) resolve with `refresh: true`. | |
| 19 | const DEFAULT_POSITIVE_CACHE_TTL = Duration.minutes(15); | |
| 20 | // Short, so a folder that gains a repository or a remote shows up quickly. | |
| 17 | 21 | const DEFAULT_NEGATIVE_CACHE_TTL = Duration.minutes(1); |
| 18 | 22 | |
| 19 | 23 | export interface RepositoryIdentityResolverOptions { |
| ... | ... | @@ -142,20 +146,23 @@ export const make = Effect.fn("RepositoryIdentityResolver.make")(function* ( |
| 142 | 146 | const processRunner = yield* ProcessRunner.ProcessRunner; |
| 143 | 147 | const cacheCapacity = options.cacheCapacity ?? DEFAULT_REPOSITORY_IDENTITY_CACHE_CAPACITY; |
| 144 | 148 | const refine = options.refine ?? Effect.succeed; |
| 149 | // Git errors and timeouts resolve to null, so they use the negative TTL like | |
| 150 | // "no repository" or "no remote". Only interrupts and defects skip the cache. | |
| 151 | const timeToLive = (exit: Exit.Exit<unknown>) => | |
| 152 | Exit.match(exit, { | |
| 153 | onSuccess: (value) => | |
| 154 | value === null | |
| 155 | ? (options.negativeCacheTtl ?? DEFAULT_NEGATIVE_CACHE_TTL) | |
| 156 | : (options.positiveCacheTtl ?? DEFAULT_POSITIVE_CACHE_TTL), | |
| 157 | onFailure: () => Duration.zero, | |
| 158 | }); | |
| 145 | 159 | |
| 146 | 160 | const repositoryRootCache = yield* Cache.makeWith<string, string | null>( |
| 147 | 161 | (cwd) => |
| 148 | 162 | resolveRepositoryIdentityCacheKey(cwd).pipe( |
| 149 | 163 | Effect.provideService(ProcessRunner.ProcessRunner, processRunner), |
| 150 | 164 | ), |
| 151 | { | |
| 152 | capacity: cacheCapacity, | |
| 153 | timeToLive: Exit.match({ | |
| 154 | onSuccess: (value) => | |
| 155 | value === null ? Duration.zero : (options.positiveCacheTtl ?? DEFAULT_POSITIVE_CACHE_TTL), | |
| 156 | onFailure: () => Duration.zero, | |
| 157 | }), | |
| 158 | }, | |
| 165 | { capacity: cacheCapacity, timeToLive }, | |
| 159 | 166 | ); |
| 160 | 167 | |
| 161 | 168 | const repositoryIdentityCache = yield* Cache.makeWith<string, RepositoryIdentity | null>( |
| ... | ... | @@ -167,27 +174,20 @@ export const make = Effect.fn("RepositoryIdentityResolver.make")(function* ( |
| 167 | 174 | (identity) => refine(identity).pipe(Effect.orElseSucceed(() => identity)), |
| 168 | 175 | ), |
| 169 | 176 | ), |
| 170 | { | |
| 171 | capacity: cacheCapacity, | |
| 172 | timeToLive: Exit.match({ | |
| 173 | onSuccess: (value) => | |
| 174 | value === null | |
| 175 | ? (options.negativeCacheTtl ?? DEFAULT_NEGATIVE_CACHE_TTL) | |
| 176 | : (options.positiveCacheTtl ?? DEFAULT_POSITIVE_CACHE_TTL), | |
| 177 | onFailure: () => Duration.zero, | |
| 178 | }), | |
| 179 | }, | |
| 177 | { capacity: cacheCapacity, timeToLive }, | |
| 180 | 178 | ); |
| 181 | 179 | |
| 182 | const resolve: RepositoryIdentityResolver["Service"]["resolve"] = Effect.fn( | |
| 183 | "RepositoryIdentityResolver.resolve", | |
| 184 | )(function* (cwd, options) { | |
| 185 | if (options?.refresh) yield* Cache.invalidate(repositoryRootCache, cwd); | |
| 186 | const cacheKey = yield* Cache.get(repositoryRootCache, cwd); | |
| 187 | if (cacheKey === null) return null; | |
| 188 | if (options?.refresh) yield* Cache.invalidate(repositoryIdentityCache, cacheKey); | |
| 189 | return yield* Cache.get(repositoryIdentityCache, cacheKey); | |
| 190 | }); | |
| 180 | // Untraced because almost every call is a cache hit. The lookups that spawn | |
| 181 | // git keep their own spans. | |
| 182 | const resolve: RepositoryIdentityResolver["Service"]["resolve"] = Effect.fnUntraced( | |
| 183 | function* (cwd, options) { | |
| 184 | if (options?.refresh) yield* Cache.invalidate(repositoryRootCache, cwd); | |
| 185 | const cacheKey = yield* Cache.get(repositoryRootCache, cwd); | |
| 186 | if (cacheKey === null) return null; | |
| 187 | if (options?.refresh) yield* Cache.invalidate(repositoryIdentityCache, cacheKey); | |
| 188 | return yield* Cache.get(repositoryIdentityCache, cacheKey); | |
| 189 | }, | |
| 190 | ); | |
| 191 | 191 | |
| 192 | 192 | return RepositoryIdentityResolver.of({ resolve }); |
| 193 | 193 | }); |
apps/server/src/ws.ts+7-3| ... | ... | @@ -3038,9 +3038,13 @@ const makeWsRpcLayer = ( |
| 3038 | 3038 | [WS_METHODS.sourceControlPublishRepository]: (input) => |
| 3039 | 3039 | observeRpcEffect( |
| 3040 | 3040 | WS_METHODS.sourceControlPublishRepository, |
| 3041 | sourceControlRepositories | |
| 3042 | .publishRepository(input) | |
| 3043 | .pipe(Effect.tap(() => refreshGitStatus(input.cwd))), | |
| 3041 | sourceControlRepositories.publishRepository(input).pipe( | |
| 3042 | // A new remote can change the cached identity. Only the `cwd` entry | |
| 3043 | // refreshes, so after a publish from a linked worktree the project | |
| 3044 | // root entry waits for its TTL. | |
| 3045 | Effect.tap(() => repositoryIdentityResolver.resolve(input.cwd, { refresh: true })), | |
| 3046 | Effect.tap(() => refreshGitStatus(input.cwd)), | |
| 3047 | ), | |
| 3044 | 3048 | { |
| 3045 | 3049 | "rpc.aggregate": "source-control", |
| 3046 | 3050 | }, |