diff --git a/apps/server/src/git/GitManager.test.ts b/apps/server/src/git/GitManager.test.ts index e1f705dea0e2..a6dfc69d67d3 100644 --- a/apps/server/src/git/GitManager.test.ts +++ b/apps/server/src/git/GitManager.test.ts @@ -1423,7 +1423,7 @@ it.layer(layerGitManagerTest)("GitManager", (it) => { }), ); - it.effect("turn-end refresh finds a new PR and keeps known PRs cached", () => + it.effect("turn-end refresh keeps a no-PR answer until its TTL, then finds the new PR", () => Effect.gen(function* () { const repoDir = yield* makeTempDir("t3code-git-manager-"); yield* initRepo(repoDir); @@ -1454,6 +1454,14 @@ it.layer(layerGitManagerTest)("GitManager", (it) => { (yield* manager.remoteStatus({ cwd: repoDir }, { refreshUpstream: false }))?.pr, ).toBeNull(); + // Fork patch (board #15036): the run end reads the cached answer, no GitHub call. + const atRunEnd = yield* manager.remoteStatus( + { cwd: repoDir }, + { refreshUpstream: false, refreshMissingPullRequest: true }, + ); + expect(atRunEnd?.pr).toBeNull(); + expect(ghCalls.filter((call) => call.startsWith("pr list "))).toHaveLength(1); + yield* TestClock.adjust("10 minutes"); const refreshed = yield* manager.remoteStatus( { cwd: repoDir }, { refreshUpstream: false, refreshMissingPullRequest: true }, @@ -2186,57 +2194,55 @@ it.layer(layerGitManagerTest)("GitManager", (it) => { }), ); - it.effect("branch PR lookup rechecks open PRs every minute and settled answers less often", () => - Effect.gen(function* () { - const repoDir = yield* makeTempDir("t3code-git-manager-"); - yield* initRepo(repoDir); - const remoteDir = yield* createBareRemote(); - yield* runGit(repoDir, ["remote", "add", "origin", remoteDir]); - for (const branch of ["feature/open-pr", "feature/merged-pr", "feature/no-pr"]) { - yield* runGit(repoDir, ["checkout", "-b", branch, "main"]); - yield* runGit(repoDir, ["push", "-u", "origin", branch]); - } - const pullRequest = (number: number, headRefName: string, state: string) => ({ - number, - title: headRefName, - url: `https://github.com/pingdotgg/codething-mvp/pull/${number}`, - baseRefName: "main", - headRefName, - state, - updatedAt: "2026-04-07T15:00:00Z", - }); - const { manager, ghCalls } = yield* makeManager({ - ghScenario: { - prListByHeadSelector: { - "feature/open-pr": encodeCliJson([pullRequest(301, "feature/open-pr", "OPEN")]), - "feature/merged-pr": encodeCliJson([pullRequest(302, "feature/merged-pr", "MERGED")]), + it.effect( + "branch PR lookup keeps open and settled answers for ten minutes (fork patch #15036)", + () => + Effect.gen(function* () { + const repoDir = yield* makeTempDir("t3code-git-manager-"); + yield* initRepo(repoDir); + const remoteDir = yield* createBareRemote(); + yield* runGit(repoDir, ["remote", "add", "origin", remoteDir]); + for (const branch of ["feature/open-pr", "feature/merged-pr", "feature/no-pr"]) { + yield* runGit(repoDir, ["checkout", "-b", branch, "main"]); + yield* runGit(repoDir, ["push", "-u", "origin", branch]); + } + const pullRequest = (number: number, headRefName: string, state: string) => ({ + number, + title: headRefName, + url: `https://github.com/pingdotgg/codething-mvp/pull/${number}`, + baseRefName: "main", + headRefName, + state, + updatedAt: "2026-04-07T15:00:00Z", + }); + const { manager, ghCalls } = yield* makeManager({ + ghScenario: { + prListByHeadSelector: { + "feature/open-pr": encodeCliJson([pullRequest(301, "feature/open-pr", "OPEN")]), + "feature/merged-pr": encodeCliJson([pullRequest(302, "feature/merged-pr", "MERGED")]), + }, }, - }, - }); - const lookupAll = Effect.forEach( - ["feature/open-pr", "feature/merged-pr", "feature/no-pr"], - (branch) => manager.branchPullRequest({ cwd: repoDir, branch }), - ); - const prListCalls = () => ghCalls.filter((call) => call.startsWith("pr list ")); - - yield* lookupAll; - expect(prListCalls()).toHaveLength(3); + }); + const lookupAll = Effect.forEach( + ["feature/open-pr", "feature/merged-pr", "feature/no-pr"], + (branch) => manager.branchPullRequest({ cwd: repoDir, branch }), + ); + const prListCalls = () => ghCalls.filter((call) => call.startsWith("pr list ")); - yield* TestClock.adjust("61 seconds"); - yield* lookupAll; - expect(prListCalls()).toHaveLength(4); - expect(prListCalls().at(-1)).toContain("--head feature/open-pr"); + yield* lookupAll; + expect(prListCalls()).toHaveLength(3); - yield* TestClock.adjust("238 seconds"); - yield* lookupAll; - expect(prListCalls()).toHaveLength(5); - expect(prListCalls().at(-1)).toContain("--head feature/open-pr"); + // Just inside ten minutes nothing is asked again, not even the open PR. + yield* TestClock.adjust("599 seconds"); + yield* lookupAll; + expect(prListCalls()).toHaveLength(3); - yield* TestClock.adjust("2 seconds"); - yield* lookupAll; - expect(prListCalls()).toHaveLength(7); - expect(prListCalls().slice(-2).join("\n")).not.toContain("--head feature/open-pr"); - }), + // Just past it all three answers expire. + yield* TestClock.adjust("2 seconds"); + yield* lookupAll; + expect(prListCalls()).toHaveLength(6); + expect(prListCalls().slice(-3).join("\n")).toContain("--head feature/open-pr"); + }), ); it.effect("branch PR lookup announces a pull request when it reads it merged", () => @@ -2278,7 +2284,8 @@ it.layer(layerGitManagerTest)("GitManager", (it) => { yield* lookup("feature/merges-later"); yield* lookup("feature/already-merged"); - yield* TestClock.adjust("61 seconds"); + // Open answers are re-read after ten minutes (fork patch #15036). + yield* TestClock.adjust("601 seconds"); expect((yield* lookup("feature/merges-later"))?.state).toBe("merged"); const announced = yield* Stream.runCollect(Stream.take(changes, 2)); diff --git a/apps/server/src/git/GitManager.ts b/apps/server/src/git/GitManager.ts index fdc2bf2f88b7..dbad8357aa04 100644 --- a/apps/server/src/git/GitManager.ts +++ b/apps/server/src/git/GitManager.ts @@ -165,8 +165,19 @@ const SHORT_SHA_LENGTH = 7; const TOAST_DESCRIPTION_MAX = 72; const STATUS_RESULT_CACHE_TTL = Duration.seconds(1); const STATUS_RESULT_CACHE_CAPACITY = 2_048; -const PR_LOOKUP_CACHE_TTL = Duration.seconds(60); -const PR_LOOKUP_NO_OPEN_PR_CACHE_TTL = Duration.minutes(5); +// Fork patch (MattRiddell/t3code, board #15036): an open PR's lookup is kept +// for ten minutes, not one. Upstream matched the one-minute settlement sweep +// so an external merge settled within about a minute, but that re-asked +// GitHub for every open-PR branch every minute (500 to 1,100 GraphQL calls an +// hour on the fleet box). A PR badge or an external merge can now lag up to +// ten minutes. Git actions, turn ends and user refreshes still bypass this +// cache, and failed lookups still back off via prLookupFailureTtl. +const PR_LOOKUP_CACHE_TTL = Duration.minutes(10); +// Answers without an open PR ("no PR yet", merged, closed) only change when +// someone opens a PR, and the paths that do that in-app (turn end, push, +// create PR, user refresh) bypass this cache. They wait at least as long as +// an open PR's answer (fork patch, board #15036; upstream: five minutes). +const PR_LOOKUP_NO_OPEN_PR_CACHE_TTL = Duration.minutes(10); const PR_LOOKUP_FAILURE_BASE_TTL = Duration.seconds(20); const PR_LOOKUP_FAILURE_MAX_TTL = Duration.minutes(15); const PR_LOOKUP_CACHE_CAPACITY = 2_048; @@ -1264,14 +1275,11 @@ export const make = Effect.gen(function* () { // `push -u`) must not orphan the fallback value for the same branch. const branchKey = `${cwd}\u0000${details.branch}`; const cacheKey = prLookupCacheKey(cwd, details); - if (refreshMissingPullRequest) { - const cached = yield* Cache.getOption(prLookupCache, cacheKey).pipe( - Effect.orElseSucceed(() => Option.none()), - ); - if (Option.isSome(cached) && cached.value.latest === null) { - yield* Cache.invalidate(prLookupCache, cacheKey); - } - } + // Fork patch (board #15036): a finished run no longer drops a cached "no pull request" + // answer. Every run end cost one GitHub read; a pull request opened during the run now + // shows when the answer's PR_LOOKUP_NO_OPEN_PR_CACHE_TTL ends (Matt, q3440: PR badges + // can lag up to 10 min). A user-driven refresh still bumps the epoch (invalidateStatus). + void refreshMissingPullRequest; return yield* getPrLookup(cacheKey).pipe( Effect.map(({ latest, headContext }) => { if (!latest) return { pr: null, headContext }; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 4222c3c27021..8303c0547456 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -178,6 +178,8 @@ export type ProjectionSettlementCandidate = Pick< OrchestrationV2ThreadShell, | "id" | "projectId" + | "lineage" + | "creationSource" | "branch" | "worktreePath" | "pullRequests" diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 841e87c274fa..671bc88bfb9e 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -24,6 +24,7 @@ import { assert, describe, it } from "@effect/vitest"; import * as Clock from "effect/Clock"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -295,7 +296,7 @@ const sweepAgain = Effect.fn("sweepPullRequestSyncHarness")(function* ( fixture: Harness, reactor: PullRequestSyncReactor.PullRequestSyncReactor["Service"], ) { - yield* TestClock.adjust("1 minute"); + yield* TestClock.adjust(PullRequestSyncReactor.nextPassDelay(yield* Clock.currentTimeMillis)); yield* Queue.take(fixture.snapshotReads); yield* reactor.drain; }); @@ -658,7 +659,7 @@ describe("PullRequestSyncReactor", () => { ), ); - it.effect("syncs a pull request a reader saw merge without waiting for the sweep", () => + it.effect("syncs a pull request a reader saw merge, fresh, on the next pass", () => Effect.scoped( Effect.gen(function* () { yield* TestClock.setTime(Date.parse(NOW)); @@ -687,6 +688,14 @@ describe("PullRequestSyncReactor", () => { repository: "owner/repository", number: 7, }); + // Fork patch (board #15036): the change waits for the next pass, not a read of its own. + yield* TestClock.adjust("1 minute"); + yield* reactor.drain; + assert.deepStrictEqual(yield* Ref.get(invalidated), []); + assert.deepStrictEqual(yield* Ref.get(fixture.syncCommands), []); + + yield* TestClock.adjust("9 minutes"); + yield* TestClock.adjust("1 second"); yield* Queue.take(fixture.snapshotReads); yield* reactor.drain; @@ -700,6 +709,37 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect("sweeps just after each ten-minute tick, not every minute", () => + Effect.scoped( + Effect.gen(function* () { + // Starts three minutes past a tick: the next pass waits for the tick, not ten minutes. + yield* TestClock.setTime(Date.parse(NOW) + 3 * 60_000); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }), + ]), + }); + const reads = Ref.get(fixture.summaryCalls).pipe(Effect.map((calls) => calls.length)); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + assert.strictEqual(yield* reads, 1); + yield* TestClock.adjust("1 minute"); + assert.strictEqual(yield* reads, 1); + yield* TestClock.adjust("6 minutes"); + assert.strictEqual(yield* reads, 1); + yield* TestClock.adjust("1 second"); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.strictEqual(yield* reads, 2); + assert.strictEqual( + Duration.toMillis(PullRequestSyncReactor.nextPassDelay(yield* Clock.currentTimeMillis)), + 10 * 60_000, + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("discovers externally reopened pull requests after fifteen minutes", () => Effect.scoped( Effect.gen(function* () { @@ -716,7 +756,8 @@ describe("PullRequestSyncReactor", () => { const reactor = yield* startAndSweep(fixture); assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 1); yield* Ref.set(state, "open"); - for (let index = 0; index < 14; index += 1) yield* sweepAgain(fixture, reactor); + // The pass ten minutes later is too soon; the next one, past fifteen, reads it. + yield* sweepAgain(fixture, reactor); assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 1); yield* sweepAgain(fixture, reactor); assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); @@ -981,7 +1022,7 @@ describe("PullRequestSyncReactor", () => { const parts = Array.isArray(message) ? message : [message]; if (logLevel === "Warn" && parts[0] === "pull request sync skipped") skips.push(parts); }); - const retryAt = Date.parse(NOW) + 3 * 60_000; + const retryAt = Date.parse(NOW) + 25 * 60_000; return Effect.scoped( Effect.gen(function* () { yield* TestClock.setTime(Date.parse(NOW)); @@ -1059,7 +1100,7 @@ describe("PullRequestSyncReactor", () => { Effect.scoped( Effect.gen(function* () { yield* TestClock.setTime(Date.parse(NOW)); - const retryAt = Date.parse(NOW) + 3 * 60_000; + const retryAt = Date.parse(NOW) + 25 * 60_000; const fixture = yield* makeHarness({ snapshot: makeSnapshot([makeThread("one", { pullRequests: [makeLink(7, null)] })]), summary: (input) => Effect.succeed(makeSummary(input, { state: "open" })), @@ -1099,56 +1140,72 @@ describe("PullRequestSyncReactor", () => { ), ); - it.effect("reads open links fresh only when a run that ran a merge command ends", () => - Effect.scoped( - Effect.gen(function* () { - yield* TestClock.setTime(Date.parse(NOW)); - const state = yield* Ref.make("open"); - const invalidated = yield* Ref.make>([]); - const fixture = yield* makeHarness({ - snapshot: makeSnapshot([ - makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }), - makeThread("other", { pullRequests: [makeLink(9, { state: "open" })] }), - ]), - summary: (input) => - Ref.get(state).pipe( - Effect.map((current) => - makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}), + it.effect( + "reads open links fresh on the next pass only when a run that ran a merge command ends", + () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const state = yield* Ref.make("open"); + const invalidated = yield* Ref.make>([]); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }), + makeThread("other", { pullRequests: [makeLink(9, { state: "open" })] }), + ]), + summary: (input) => + Ref.get(state).pipe( + Effect.map((current) => + makeSummary( + input, + current === "merged" && input.number === 7 + ? { state: current, mergedAt: NOW } + : {}, + ), + ), ), - ), - invalidate: ({ reference }) => - Ref.update(invalidated, (numbers) => [...numbers, reference?.number ?? -1]), - }); - - yield* Effect.gen(function* () { - const reactor = yield* startAndSweep(fixture); - yield* Ref.set(state, "merged"); - yield* Ref.set(fixture.summaryCalls, []); - - yield* Queue.offerAll(fixture.domainEvents, [ - commandRan(ThreadId.make("other"), "gh pr view 9"), - runUpdated(ThreadId.make("other"), "completed"), - commandRan(ThreadId.make("agent"), "gh pr merge 7 --squash 2>&1 | tail -3"), - runUpdated(ThreadId.make("agent"), "completed"), - ]); - yield* Queue.take(fixture.snapshotReads); - yield* Queue.take(fixture.snapshotReads); - yield* reactor.drain; + invalidate: ({ reference }) => + Ref.update(invalidated, (numbers) => [...numbers, reference?.number ?? -1]), + }); - assert.deepStrictEqual(yield* Ref.get(invalidated), [7]); - assert.deepStrictEqual( - (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number), - [7], - ); - assert.deepStrictEqual( - (yield* Ref.get(fixture.syncCommands)).map((command) => [ - command.number, - command.snapshot.state, - ]), - [[7, "merged"]], - ); - }).pipe(Effect.provide(fixture.layer)); - }), - ), + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + yield* Ref.set(state, "merged"); + yield* Ref.set(fixture.summaryCalls, []); + + yield* Queue.offerAll(fixture.domainEvents, [ + commandRan(ThreadId.make("other"), "gh pr view 9"), + runUpdated(ThreadId.make("other"), "completed"), + commandRan(ThreadId.make("agent"), "gh pr merge 7 --squash 2>&1 | tail -3"), + runUpdated(ThreadId.make("agent"), "completed"), + ]); + // The agent thread's lookup. Fork patch (board #15036): no read of its own; the links + // wait for the next pass. + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.deepStrictEqual(yield* Ref.get(invalidated), []); + assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), []); + + yield* TestClock.adjust("10 minutes"); + yield* TestClock.adjust("1 second"); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + + // Only the merged link is read fresh; the pass reads the other open link as usual. + assert.deepStrictEqual(yield* Ref.get(invalidated), [7]); + assert.deepStrictEqual( + (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number).toSorted(), + [7, 9], + ); + assert.deepStrictEqual( + (yield* Ref.get(fixture.syncCommands)).map((command) => [ + command.number, + command.snapshot.state, + ]), + [[7, "merged"]], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), ); }); diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index c2cfe497d35a..0198d45293bd 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -20,16 +20,17 @@ import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; -import * as Schedule from "effect/Schedule"; import type * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as GitManager from "../git/GitManager.ts"; import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider"; +import { entryExpiresAt } from "../pullRequest/PullRequestReadCache.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import { forkParked } from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; @@ -37,6 +38,30 @@ import * as ProjectionStore from "./ProjectionStore.ts"; import { isTerminalRunStatus } from "./ThreadManagementService.ts"; const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000; +/** + * Fork patch (board #15036): the periodic pass runs just after each tick of the PR cache's + * ten-minute clock (`entryExpiresAt`), not every minute. A one-minute pass found a few stale + * pull requests each minute (entries expired one by one, and every finished run invalidates its + * project's entries) and sent each handful as its own GraphQL document: about 50 an hour, 1,213 + * on 2026-10-13. Every entry now expires on the tick, so the pass finds them stale together and + * they share one document of 25. Background requests wait for the pass too + * (`DEFER_BACKGROUND_SYNC_TO_PASS`); a client's own refresh reads at once. A badge's data is at most one tick old when the pass + * writes it (Matt, q3440: PR badges can lag up to 10 min). + */ +const PASS_AFTER_TICK_MS = 1_000; +export const nextPassDelay = (nowMs: number): Duration.Duration => + Duration.millis(entryExpiresAt(nowMs) - nowMs + PASS_AFTER_TICK_MS); +/** Shell commands that can merge or close a pull request without a merge notification. */ +/** + * Fork patch (board #15036): background reasons to re-read a pull request (a state change a + * branch lookup or detail read saw, a new link with no snapshot, a run that ran `gh pr merge`) + * mark it for the next periodic pass instead of reading it at once. Each read at once was its + * own GraphQL document between ticks: about 160 a day on 2026-10-14, most in bursts while + * batches merged. The pass reads them with every other due pull request, at most one tick + * later (Matt, q3440: PR badges can lag up to 10 min). A client's own refresh still reads at + * once through `requestSync`. + */ +export const DEFER_BACKGROUND_SYNC_TO_PASS = true; const PULL_REQUEST_CLOSE_COMMAND = /\b(?:gh\s+pr|glab\s+mr)\s+(?:merge|close)\b/u; const isPullRequestProviderError = Schema.is(PullRequestProviderError); @@ -393,7 +418,7 @@ export const make = Effect.gen(function* () { (link) => link.snapshot?.state === "open", ), ), - requestSync, + deferSync, { discard: true }, ), ), @@ -404,12 +429,14 @@ export const make = Effect.gen(function* () { "PullRequestSyncReactor.start", )(function* () { const events = engine.streamDomainEvents; + // A client reading a pull request, or its branch status, can see it merge or close before + // the next sweep does. The next pass reads it fresh. const stateChanges = Stream.merge( yield* pullRequests.subscribeStateChanges, yield* git.subscribePullRequestStateChanges, ); yield* forkParked( - Stream.runForEach(stateChanges, requestSync).pipe( + Stream.runForEach(stateChanges, deferSync).pipe( Effect.catchCause(logSkipped("pull request state change stream failed", {})), ), ); @@ -423,13 +450,13 @@ export const make = Effect.gen(function* () { visibleThreadPullRequests(event.payload.pullRequests ?? []).filter( (link) => link.snapshot === null, ), - requestSync, + deferSync, { discard: true }, ); // An agent can merge or close its pull request from a shell (`gh pr merge`), which - // sends no merge notification. When a run that ran such a command ends, read the - // thread's open links fresh, so settlement does not wait for the next sweep and the - // cached summary. Other runs add no host reads. + // sends no merge notification. When a run that ran such a command ends, the next pass + // reads the thread's open links fresh rather than from the cached summary. Other runs + // add no host reads. case "turn-item.updated": if ( event.payload.type === "command_execution" && @@ -452,7 +479,8 @@ export const make = Effect.gen(function* () { Effect.gen(function* () { yield* worker.enqueue("all"); yield* worker.drain; - }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid), + yield* Effect.sleep(nextPassDelay(yield* Clock.currentTimeMillis)); + }).pipe(Effect.forever, Effect.asVoid), ); }); @@ -464,6 +492,14 @@ export const make = Effect.gen(function* () { return worker.enqueue("requested"); }); + // Marks the key for the next periodic pass, which reads requested keys with the rest. + const deferSync = (key: ThreadPullRequestKey): Effect.Effect => + DEFER_BACKGROUND_SYNC_TO_PASS + ? Effect.sync(() => { + requested.set(threadPullRequestKeyOf(key), ++requestGeneration); + }) + : requestSync(key); + return { start, drain: worker.drain, requestSync } satisfies PullRequestSyncReactor["Service"]; }); diff --git a/apps/server/src/orchestration-v2/ThreadPullRequestService.test.ts b/apps/server/src/orchestration-v2/ThreadPullRequestService.test.ts index 25afb151e280..2387ccead4e1 100644 --- a/apps/server/src/orchestration-v2/ThreadPullRequestService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadPullRequestService.test.ts @@ -257,6 +257,57 @@ describe("ThreadPullRequestServiceV2 reads", () => { ), ); + it("periodic passes skip a branch whose pull request is known and open", () => { + // Fork patch (board #15036): the pull request sync pass already reads that pull request. + const link = (number: number, state: "open" | "merged" | "closed" | null) => ({ + host: "github.com", + repository: "Owner/Repo", + number, + url: `https://github.com/Owner/Repo/pull/${number}`, + source: "created" as const, + linkedAt: "2026-09-20T00:00:00.000Z", + snapshot: + state === null + ? null + : { + state, + title: `Pull request ${number}`, + headBranch: "feature/x", + baseBranch: "main", + isDraft: false, + updatedAt: "2026-09-20T00:00:00.000Z", + syncedAt: "2026-09-20T00:00:00.000Z", + }, + stack: null, + }); + const thread = ( + branchNumber: number | null, + links: ReadonlyArray>, + ): OrchestrationV2ThreadShell => ({ + ...threadShell("thread"), + branch: "feature/x", + branchPullRequest: + branchNumber === null + ? null + : { + projectId: ProjectId.make("project-1"), + repository: "owner/repo", + number: branchNumber, + url: `https://github.com/Owner/Repo/pull/${branchNumber}`, + }, + pullRequests: links, + }); + const skips = ThreadPullRequestService.hasOpenBranchPullRequest; + expect(skips(thread(7, [link(7, "open")]))).toBe(true); + // Must say no: anything a new lookup could still change. + expect(skips(thread(7, [link(7, "merged")]))).toBe(false); + expect(skips(thread(7, [link(7, "closed")]))).toBe(false); + expect(skips(thread(7, [link(7, null)]))).toBe(false); + expect(skips(thread(7, [link(8, "open")]))).toBe(false); + expect(skips(thread(null, [link(7, "open")]))).toBe(false); + expect(skips(thread(7, []))).toBe(false); + }); + it.effect("periodic sweeps read only unsettled threads once backfill is done", () => Effect.scoped( Effect.gen(function* () { @@ -307,7 +358,8 @@ describe("ThreadPullRequestServiceV2 reads", () => { yield* Deferred.succeed(activation, undefined); expect(yield* Queue.take(reads)).toEqual({ location: "active", unsettledOnly: false }); yield* service.drain; - yield* TestClock.adjust("1 minute"); + // Its project is gone, so backfill finishes it on the first pass. + yield* TestClock.adjust(ThreadPullRequestService.PERIODIC_PASS_INTERVAL); expect(yield* Queue.take(reads)).toEqual({ location: "active", unsettledOnly: true }); yield* service.drain; }).pipe(Effect.provide(layerDependencies)); diff --git a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts index 21387243b46c..a124b6900481 100644 --- a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts +++ b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts @@ -4,6 +4,7 @@ import { } from "@t3tools/shared/sourceControl"; import { CommandId, + isProviderNativeSubagentThread, type OrchestrationProjectShell, type OrchestrationV2DomainEvent, type OrchestrationV2ThreadShell, @@ -90,7 +91,43 @@ export function projectWorkspaceMatchesSnapshot( ); } +/** + * Fork patch (board #15036): the periodic pass skips a thread whose branch pull request is + * known and open. The pull request sync pass already reads that pull request every ten minutes, + * and a head with an open pull request gets no new one until it merges or closes, which that + * pass sees. Looking the branch up again cost a GraphQL document on most passes. Threads with + * no known pull request, or a merged or closed one, are still looked up every pass. + */ +export const hasOpenBranchPullRequest = (thread: OrchestrationV2ThreadShell): boolean => { + const known = thread.branchPullRequest; + if (known == null) return false; + return (thread.pullRequests ?? []).some( + (link) => + link.number === known.number && + link.repository.toLowerCase() === known.repository.toLowerCase() && + link.snapshot?.state === "open", + ); +}; + const BACKFILL_ATTEMPTS = 5; +/** + * Fork patch (board #15036): the periodic pass over every unsettled thread runs every ten + * minutes, not every minute. Branch lookups are cached for ten minutes, so a one-minute pass + * found a few expired branches each minute and sent each handful to GitHub as its own GraphQL + * document. A ten-minute pass finds them expired together, and they share documents. + */ +export const PERIODIC_PASS_INTERVAL = "10 minutes"; + +/** + * Fork patch (board #15036): a provider's own subagent works on its root thread's branch. Once + * the root is archived or settled the subagent is left behind, unarchived and unsettled, and + * every pass looked up its branch again. The fleet's daily bench archived its roots and left + * 493 such threads, 231 of the 267 branch lookups a pass made. + */ +const isLeftBehindSubagent = ( + thread: Pick, + activeThreadIds: ReadonlySet, +) => isProviderNativeSubagentThread(thread) && !activeThreadIds.has(thread.lineage.rootThreadId); interface RefreshRequest { readonly threadId: ThreadId | null; @@ -161,9 +198,14 @@ export const make = Effect.gen(function* () { for (const threadId of checkedIds) { if (!visibleThreadIds.has(threadId)) pendingBackfill.delete(threadId); } + // A pass over all threads reads only active ones, so a root missing from it is archived + // or settled. A single-thread request cannot tell, and keeps its thread. + const activeThreadIds = new Set(threadSnapshot.threads.map((thread) => thread.id)); const threads = threadSnapshot.threads.filter( (thread) => thread.archivedAt === null && + (request.threadId !== null || !isLeftBehindSubagent(thread, activeThreadIds)) && + (request.threadId !== null || !hasOpenBranchPullRequest(thread)) && ((thread.settledOverride !== "settled" && thread.settledAt === null) || request.threadId !== null || pendingBackfill.has(thread.id)) && @@ -367,8 +409,12 @@ export const make = Effect.gen(function* () { case "thread.metadata-updated": return worker.enqueue({ threadId: event.threadId, refresh: false }); case "thread.unsettled": - case "checkpoint.captured": return worker.enqueue({ threadId: event.threadId, refresh: true }); + // Fork patch (board #15036): a checkpoint or a finished run reads the cached branch + // lookup. Forcing a fresh read cost one GitHub document per run end; a pull request + // opened during the run shows within the cache's ten minutes (Matt, q3440). + case "checkpoint.captured": + return worker.enqueue({ threadId: event.threadId, refresh: false }); case "run.updated": if ( event.payload.status === "completed" || @@ -376,7 +422,7 @@ export const make = Effect.gen(function* () { event.payload.status === "cancelled" || event.payload.status === "interrupted" ) { - return worker.enqueue({ threadId: event.threadId, refresh: true }); + return worker.enqueue({ threadId: event.threadId, refresh: false }); } break; } @@ -394,7 +440,10 @@ export const make = Effect.gen(function* () { yield* Effect.gen(function* () { yield* worker.enqueue({ threadId: null, refresh: false }); yield* worker.drain; - }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.delay("1 minute")); + }).pipe( + Effect.repeat(Schedule.spaced(PERIODIC_PASS_INTERVAL)), + Effect.delay(PERIODIC_PASS_INTERVAL), + ); }).pipe(Effect.asVoid), ); }); diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index b4ca137f3f3a..3b32e7a45f7c 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -636,6 +636,12 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: : snapshot.threads.filter((thread) => thread.id === threadId), ), ), + getThreadShell: (threadId) => + Ref.get(snapshots).pipe( + Effect.map( + (snapshot) => snapshot.threads.find((thread) => thread.id === threadId) ?? null, + ), + ), getThread: (threadId) => { const thread = options.currentThreads?.find((candidate) => candidate.id === threadId); return thread @@ -1059,6 +1065,49 @@ describe("ThreadSettlementServiceV2 worker", () => { }), ), ); + + it.effect("skips the branch lookup for a provider subagent whose root is gone", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const subagent = (id: string, rootThreadId: string, creationSource: "provider" | "mcp") => + makeThread(id, { + branch: `feature/${id}`, + creationSource, + lineage: { + rootThreadId: ThreadId.make(rootThreadId), + parentThreadId: ThreadId.make(rootThreadId), + relationshipToParent: "subagent", + }, + }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("live-root", { branch: "feature/live-root" }), + subagent("live-child", "live-root", "provider"), + subagent("orphan-child", "archived-root", "provider"), + makeThread("settled-root", { settledOverride: "settled" }), + subagent("settled-root-child", "settled-root", "provider"), + subagent("delegated-child", "archived-root", "mcp"), + ]), + settings: { + ...DEFAULT_SERVER_SETTINGS, + sidebarAutoSettleAfterDays: null, + sidebarAutoSettleOnMerge: true, + }, + }); + + yield* Effect.gen(function* () { + const reactor = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(reactor, fixture.activation, fixture.snapshotReads); + + assert.deepStrictEqual( + new Set((yield* Ref.get(fixture.branchCalls)).map((call) => call.branch)), + new Set(["feature/live-root", "feature/live-child", "feature/delegated-child"]), + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); }); describe("ThreadSettlementServiceV2 terminals", () => { diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index bba955cb2e05..fe3e1ef5c601 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -3,6 +3,7 @@ import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; import { CommandId, + isProviderNativeSubagentThread, type ThreadId, type OrchestrationV2DomainEvent, type OrchestrationV2ThreadShell, @@ -37,6 +38,18 @@ export interface SettlementPullRequest { const DAY_MS = 24 * 60 * 60 * 1_000; export const QUEUED_TURN_START_GRACE_MS = 2 * 60 * 1_000; +/** + * Fork patch (board #15036): the full sweep ran every minute and each pass read + * GitHub for every branch candidate whose cached answer had expired. PR answers + * now live ten minutes (Matt, q3440: PR badges can lag up to 10 min), so a + * faster sweep only finds the same answers. + */ +export const SETTLEMENT_SWEEP_INTERVAL = "10 minutes"; +/** + * Branch lookups per sweep run together so GitHub's batched head lookups fill a + * document (25 per background document) instead of eight at a time. + */ +const SETTLEMENT_LOOKUP_CONCURRENCY = 25; function toMillis(value: DateTime.Utc | null | undefined): number | null { return value == null ? null : DateTime.toEpochMillis(value); @@ -351,13 +364,40 @@ export const make = Effect.gen(function* () { ), ); - const lookupCandidates = (yield* Effect.forEach( + // Inactivity is entirely projection-backed. Complete those decisions before + // a source-control lookup can delay or fail an otherwise eligible thread. + const settledCandidates = (yield* Effect.forEach( candidates, (thread) => settleThread(thread, null), { concurrency: 8 }, )) .filter((thread) => thread !== null) .filter((thread) => visibleThreadPullRequests(thread.pullRequests ?? []).length === 0); + // Fork patch (board #15036): a provider-native subagent whose root thread is + // archived, settled or gone stays unarchived and unsettled (493 of 547 candidates on + // 2026-10-12). Nobody sees its pull request, so it settles by age above and + // costs no GitHub read here. + const rootLeftBehind = new Map(); + const isLeftBehind = (thread: (typeof settledCandidates)[number]) => + Effect.gen(function* () { + if (!isProviderNativeSubagentThread(thread)) return false; + const rootThreadId = thread.lineage.rootThreadId; + if (rootThreadId === thread.id) return false; + const known = rootLeftBehind.get(rootThreadId); + if (known !== undefined) return known; + const root = yield* projections.getThreadShell(rootThreadId); + const leftBehind = + root === null || + root.archivedAt !== null || + root.settledOverride === "settled" || + root.settledAt !== null; + rootLeftBehind.set(rootThreadId, leftBehind); + return leftBehind; + }); + const lookupCandidates = yield* Effect.filter(settledCandidates, (thread) => + Effect.map(isLeftBehind(thread), (leftBehind) => !leftBehind), + ); + // Use the same cwd as the sidebar so both paths share GitManager's PR cache. const lookupCwdByThreadId = new Map(); yield* Effect.forEach( lookupCandidates, @@ -512,7 +552,7 @@ export const make = Effect.gen(function* () { }), ), ), - { concurrency: 8, discard: true }, + { concurrency: SETTLEMENT_LOOKUP_CONCURRENCY, discard: true }, ); }); @@ -609,7 +649,7 @@ export const make = Effect.gen(function* () { Effect.gen(function* () { yield* worker.enqueue(undefined); yield* worker.drain; - }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid), + }).pipe(Effect.repeat(Schedule.spaced(SETTLEMENT_SWEEP_INTERVAL)), Effect.asVoid), ); yield* forkParked( Stream.runForEach(settingsChanges, (settings) => { diff --git a/apps/server/src/project/RepositoryIdentityResolver.test.ts b/apps/server/src/project/RepositoryIdentityResolver.test.ts index 404b3c5b5b65..91bfabb879e2 100644 --- a/apps/server/src/project/RepositoryIdentityResolver.test.ts +++ b/apps/server/src/project/RepositoryIdentityResolver.test.ts @@ -203,6 +203,28 @@ it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => { }).pipe(Effect.provide(RepositoryIdentityResolver.layer)), ); + it.effect("resolves a partial clone whose remote lines carry the filter", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const cwd = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-repository-identity-test-", + }); + + yield* git(cwd, ["init"]); + yield* git(cwd, ["remote", "add", "origin", "git@github.com:T3Tools/t3code.git"]); + // `git fetch --filter=blob:none` sets these; `git remote -v` then prints + // `origin (fetch) [blob:none]`. + yield* git(cwd, ["config", "remote.origin.promisor", "true"]); + yield* git(cwd, ["config", "remote.origin.partialclonefilter", "blob:none"]); + + const resolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver; + const identity = yield* resolver.resolve(cwd); + + expect(identity?.canonicalKey).toBe("github.com/t3tools/t3code"); + expect(identity?.provider).toBe("github"); + }).pipe(Effect.provide(RepositoryIdentityResolver.layer)), + ); + it.effect("returns the git top-level root path when resolving from a nested workspace", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; diff --git a/apps/server/src/project/RepositoryIdentityResolver.ts b/apps/server/src/project/RepositoryIdentityResolver.ts index 57084dd179f3..474aaf3b45b6 100644 --- a/apps/server/src/project/RepositoryIdentityResolver.ts +++ b/apps/server/src/project/RepositoryIdentityResolver.ts @@ -43,7 +43,8 @@ function parseRemoteFetchUrls(stdout: string): Map { for (const line of stdout.split("\n")) { const trimmed = line.trim(); if (trimmed.length === 0) continue; - const match = /^(\S+)\s+(\S+)\s+\((fetch|push)\)$/.exec(trimmed); + // A partial clone's lines end with its filter, e.g. `origin (fetch) [blob:none]`. + const match = /^(\S+)\s+(\S+)\s+\((fetch|push)\)(?:\s+\[[^\]]*\])?$/.exec(trimmed); if (!match) continue; const [, remoteName = "", remoteUrl = "", direction = ""] = match; if (direction !== "fetch" || remoteName.length === 0 || remoteUrl.length === 0) { diff --git a/apps/server/src/pullRequest/PullRequestReadCache.test.ts b/apps/server/src/pullRequest/PullRequestReadCache.test.ts index a4c0c8b17879..5f713057fa59 100644 --- a/apps/server/src/pullRequest/PullRequestReadCache.test.ts +++ b/apps/server/src/pullRequest/PullRequestReadCache.test.ts @@ -99,6 +99,10 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => { ), ); + it("keeps an entry for ten minutes (fork patch #15036)", () => { + assert.strictEqual(Duration.toMillis(PullRequestReadCache.ENTRY_TTL), 600_000); + }); + it.effect("reuses files after restart and respects the original expiry", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; @@ -108,7 +112,9 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => { const first = yield* cacheLayer(directory); const key = "long/repository/key".repeat(100); assert.strictEqual(yield* first.get(key, lookup), "1"); - yield* TestClock.adjust("59 seconds"); + yield* TestClock.adjust( + Duration.subtract(PullRequestReadCache.ENTRY_TTL, Duration.seconds(1)), + ); const restarted = yield* cacheLayer(directory); assert.strictEqual(yield* restarted.get(key, lookup), "1"); yield* TestClock.adjust("1 second"); @@ -118,6 +124,36 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => { }), ); + it.effect("expires entries read at different times together at the ten-minute tick", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" }); + yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe("2026-10-01T00:03:00Z"))); + const cache = yield* cacheLayer(directory); + let reads = 0; + const lookup = Effect.sync(() => String(++reads)); + assert.strictEqual(yield* cache.get("early", lookup), "1"); + yield* TestClock.adjust("6 minutes"); + assert.strictEqual(yield* cache.get("late", lookup), "2"); + // 00:09:59: both are still fresh. + yield* TestClock.adjust("59 seconds"); + assert.strictEqual(yield* cache.get("early", lookup), "1"); + assert.strictEqual(yield* cache.get("late", lookup), "2"); + // 00:10:00: both went stale at the tick, though "late" is one minute old. + yield* TestClock.adjust("1 second"); + assert.strictEqual(yield* cache.get("early", lookup), "3"); + assert.strictEqual(yield* cache.get("late", lookup), "4"); + }), + ); + + it("never keeps an entry longer than ENTRY_TTL", () => { + const ttl = Duration.toMillis(PullRequestReadCache.ENTRY_TTL); + const tick = Date.parse("2026-10-01T00:10:00Z"); + assert.strictEqual(PullRequestReadCache.entryExpiresAt(tick), tick + ttl); + assert.strictEqual(PullRequestReadCache.entryExpiresAt(tick - 1), tick); + assert.strictEqual(PullRequestReadCache.entryExpiresAt(tick + 1), tick + ttl); + }); + it.effect("clears in-flight reads before a new service can reuse them", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; @@ -216,13 +252,17 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => { Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" }); + // Off the ten-minute grid, so the scopes expire (at 00:15) before the fresh entry (00:20). + yield* TestClock.setTime(5 * 60_000); const cache = yield* cacheLayer(directory); yield* cache.get("summary", Effect.succeed("old"), ["pr"]); yield* cache.invalidate("pr"); for (let index = 0; index < 100; index++) yield* cache.invalidate(`pr-${index}`); assert.strictEqual((yield* fs.readDirectory(directory)).length, 2); const before = (yield* fs.stat(`${directory}/revisions`)).size; - yield* TestClock.adjust("59 seconds"); + yield* TestClock.adjust( + Duration.subtract(PullRequestReadCache.ENTRY_TTL, Duration.seconds(1)), + ); assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh"); yield* TestClock.adjust("1 second"); yield* cache.invalidate("other-pr"); diff --git a/apps/server/src/pullRequest/PullRequestReadCache.ts b/apps/server/src/pullRequest/PullRequestReadCache.ts index 16860c78ee9a..7006bde46ce9 100644 --- a/apps/server/src/pullRequest/PullRequestReadCache.ts +++ b/apps/server/src/pullRequest/PullRequestReadCache.ts @@ -29,10 +29,29 @@ const CONCURRENT_READS = 512; const STORE_ID = "pr-v2"; const ENTRY_FILE_NAME = new RegExp(`^${STORE_ID}[0-9a-f]{64}$`); /** - * A file untouched for a day is long expired; the day only leaves slack for - * clock changes. + * Entries expire at most ENTRY_TTL after they are written, and a write sets the file's + * mtime. A file untouched for a day is long expired; the day only leaves slack + * for clock changes. Pruning a live entry would cost one refetch. */ export const ENTRY_FILE_MAX_AGE = Duration.days(1); +/** + * Fork patch (MattRiddell/t3code, board #15036): a pull request read is kept + * for ten minutes, not one, so the settlement and sync sweeps stop re-reading + * every PR from GitHub each minute. A scope invalidation lasts as long as an + * entry, so an entry read before the invalidation can never outlive it. + */ +export const ENTRY_TTL = Duration.minutes(10); +const ENTRY_TTL_MILLIS = Duration.toMillis(ENTRY_TTL); +/** + * Fork patch (MattRiddell/t3code, board #15036): an entry expires at the next + * multiple of ENTRY_TTL since the epoch, not ENTRY_TTL after its own read. + * Entries read at different times then go stale together, so the sync + * reactor's pass just after each tick reads them together, in shared GraphQL + * documents. No entry lives longer than ENTRY_TTL, so a scope invalidation + * still outlives every entry read before it. + */ +export const entryExpiresAt = (nowMillis: number): number => + (Math.floor(nowMillis / ENTRY_TTL_MILLIS) + 1) * ENTRY_TTL_MILLIS; type ReadError = PullRequestOperationError | PullRequestUnavailableError; const revisionCodec = Schema.fromJsonString( Schema.Record( @@ -105,7 +124,7 @@ export const make = Effect.gen(function* () { request.lookup.pipe( Effect.map((payload) => ({ payload, - expiresAt: clock.currentTimeMillisUnsafe() + 60_000, + expiresAt: entryExpiresAt(clock.currentTimeMillisUnsafe()), revision: request.revision, })), ), @@ -160,7 +179,7 @@ export const make = Effect.gen(function* () { const next = Object.fromEntries( Object.entries(current).filter(([, value]) => value.expiresAt > now), ); - next[scope] = { revision: yield* crypto.randomUUIDv4(), expiresAt: now + 60_000 }; + next[scope] = { revision: yield* crypto.randomUUIDv4(), expiresAt: now + ENTRY_TTL_MILLIS }; const encoded = yield* Schema.encodeEffect(revisionCodec)(next); yield* backing.set("revisions", encoded); yield* Cache.set(revisions, undefined, next); diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index 098b1ebcf649..e4a3a8590f33 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -464,7 +464,8 @@ function parseGitRemoteVerboseOutput( continue; } - const match = /^(\S+)\s+(\S+)\s+\((fetch|push)\)$/.exec(trimmed); + // A partial clone's lines end with its filter, e.g. `origin (fetch) [blob:none]`. + const match = /^(\S+)\s+(\S+)\s+\((fetch|push)\)(?:\s+\[[^\]]*\])?$/.exec(trimmed); if (!match) { continue; } diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index db56c68b48eb..f5982a45a70f 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -344,7 +344,8 @@ function parseRemoteFetchUrls(stdout: string): Map { for (const line of stdout.split("\n")) { const trimmed = line.trim(); if (trimmed.length === 0) continue; - const match = /^(\S+)\s+(\S+)\s+\((fetch|push)\)$/.exec(trimmed); + // A partial clone's lines end with its filter, e.g. `origin (fetch) [blob:none]`. + const match = /^(\S+)\s+(\S+)\s+\((fetch|push)\)(?:\s+\[[^\]]*\])?$/.exec(trimmed); if (!match) continue; const [, remoteName = "", remoteUrl = "", direction = ""] = match; if (direction !== "fetch" || remoteName.length === 0 || remoteUrl.length === 0) { diff --git a/packages/provider-grok/src/server/xaiAcpExtension.test.ts b/packages/provider-grok/src/server/xaiAcpExtension.test.ts index c31d9438972c..e93b29afeb38 100644 --- a/packages/provider-grok/src/server/xaiAcpExtension.test.ts +++ b/packages/provider-grok/src/server/xaiAcpExtension.test.ts @@ -664,6 +664,181 @@ describe("XAiAcpExtension", () => { ]); }); + it("hydrates an auto-backgrounded shell command whose task id is its call id (#16214)", () => { + // Live 2026-10-08 (lcr-bench fe-14222 grok-4.7): Grok moved a foreground + // run_terminal_command to the background after 15s and reused the tool + // call id as the task id. The UUID-only match dropped this TaskOutput, the + // tool row stayed running and deferred finalize never ended the turn. + const taskId = "call-586d8645-9bbb-44f9-a337-ed45d3b87124-70"; + expect( + extractXAiMonitorTaskId({ + toolCallId: taskId, + title: "[bg] yarn compile (call-586)", + status: "completed", + data: { + rawOutput: { + type: "BackgroundTaskStarted", + task_id: taskId, + task_type: "bash", + status: "running", + command: "yarn compile", + }, + }, + }), + ).toBe(taskId); + expect( + extractXAiBackgroundTaskCompletion({ + toolCallId: "call-fe86669b-694a-434d-b9b2-c7ef58375311-71", + title: "yarn compile (call-586)", + status: "completed", + data: { + rawInput: { variant: "TaskOutput", task_ids: [taskId], timeout_ms: 300000 }, + rawOutput: { + type: "TaskOutput", + Result: { + task_id: taskId, + command: "yarn compile", + status: "failed", + exit_code: 2, + output: "error TS2304: Cannot find name 'x'.\n", + }, + }, + }, + }), + ).toEqual([{ taskId, status: "failed", appendOutput: "error TS2304: Cannot find name 'x'." }]); + }); + + it("hydrates a call-id task from the request list and the task header (#16214)", () => { + const taskId = "call-586d8645-9bbb-44f9-a337-ed45d3b87124-70"; + expect( + extractXAiBackgroundTaskCompletion({ + toolCallId: "call-get-call-id-request", + title: "get_command_or_subagent_output", + status: "completed", + data: { + rawInput: { variant: "TaskOutput", task_ids: [taskId] }, + rawOutput: { type: "TaskOutput", Result: { status: "completed", output: "done\n" } }, + }, + }), + ).toMatchObject([{ taskId, status: "completed" }]); + expect( + extractXAiBackgroundTaskCompletion({ + toolCallId: "call-get-call-id-header", + title: "get_command_or_subagent_output", + status: "completed", + data: { + rawInput: { variant: "TaskOutput" }, + content: [ + { + type: "content", + content: { type: "text", text: `=== Task ${taskId} ===\nStatus: completed\ndone` }, + }, + ], + }, + }), + ).toMatchObject([{ taskId }]); + }); + + it("hydrates each task of a wait_all MultiResult from its own entry (#16214)", () => { + // Live 2026-10-14 (lcr-bench fe-10784 grok-4.7-medium): one TaskOutput + // waited on two auto-backgrounded commands. Grok answered with + // MultiResult.results[] and no Result or text header, so both tasks were + // reported running and the turn never ended. + const first = "call-c0b011f1-de42-43de-ac33-a3417c0a748f-117"; + const second = "call-c0b011f1-de42-43de-ac33-a3417c0a748f-118"; + expect( + extractXAiBackgroundTaskCompletion({ + toolCallId: "call-5ef78a10-7efb-43a2-9820-bf461bb83e6c-119", + title: "multi-wait (wait_all)", + status: "completed", + data: { + rawInput: { variant: "TaskOutput", task_ids: [first, second], timeout_ms: 300000 }, + rawOutput: { + type: "TaskOutput", + MultiResult: { + mode: "wait_all", + results: [ + { task_id: first, status: "completed", exit_code: 0, output: "tests ok\n" }, + { task_id: second, status: "failed", exit_code: 1, output: "catalog drift\n" }, + ], + summary: "2/2 tasks completed (wait_all)", + }, + }, + }, + }), + ).toEqual([ + { taskId: first, status: "completed", appendOutput: "tests ok" }, + { taskId: second, status: "failed", appendOutput: "catalog drift" }, + ]); + }); + + it("keeps a MultiResult entry without a terminal status running (#16214)", () => { + const done = "019f44a6-4820-7402-925d-bc862ee711dd"; + const open = "call-c0b011f1-de42-43de-ac33-a3417c0a748f-118"; + expect( + extractXAiBackgroundTaskCompletion({ + toolCallId: "call-get-multi-any", + title: "multi-wait (wait_any)", + status: "completed", + data: { + rawInput: { variant: "TaskOutput", task_ids: [done, open, "not-a-task"] }, + rawOutput: { + type: "TaskOutput", + MultiResult: { + mode: "wait_any", + results: [ + { task_id: done, exit_code: 0, output: "" }, + { task_id: open, status: "running", output: "still going\n" }, + { task_id: "not-a-task", status: "completed" }, + ], + }, + }, + }, + }), + ).toEqual([ + { taskId: done, status: "completed", appendOutput: "" }, + { taskId: open, status: "running", appendOutput: "still going" }, + ]); + }); + + it("still ignores TaskOutput ids that are neither a UUID nor a Grok call id", () => { + for (const taskId of [ + "call-get-mon", + "not-a-task", + "call-586d8645-9bbb-44f9-a337-ed45d3b87124-x", + ]) { + expect( + extractXAiBackgroundTaskCompletion({ + toolCallId: "call-get-bogus", + title: "get_command_or_subagent_output", + status: "completed", + data: { + rawInput: { variant: "TaskOutput", task_ids: [taskId] }, + rawOutput: { type: "TaskOutput", Result: { task_id: taskId, status: "completed" } }, + }, + }), + ).toEqual([]); + } + }); + + it("does not report a call-id TaskOutput as a subagent (#16214)", () => { + const taskId = "call-586d8645-9bbb-44f9-a337-ed45d3b87124-70"; + expect( + extractXAiAcpSubagentUpdate({ + toolCallId: "call-fe86669b-694a-434d-b9b2-c7ef58375311-71", + title: "yarn compile (call-586)", + status: "completed", + data: { + rawInput: { variant: "TaskOutput", task_ids: [taskId] }, + rawOutput: { + type: "TaskOutput", + Result: { task_id: taskId, status: "failed", exit_code: 2, output: "x\n" }, + }, + }, + }), + ).toBeUndefined(); + }); + it("hydrates monitor completion from a standalone TaskOutput frame without rawInput", () => { // Post-settle shape observed live (2026-07-12): the final // get_command_or_subagent_output update carries only rawOutput; the frame diff --git a/packages/provider-grok/src/server/xaiAcpExtension.ts b/packages/provider-grok/src/server/xaiAcpExtension.ts index c392b88cb0db..cabfab997fc8 100644 --- a/packages/provider-grok/src/server/xaiAcpExtension.ts +++ b/packages/provider-grok/src/server/xaiAcpExtension.ts @@ -122,6 +122,11 @@ export interface XAiAcpSubagentEndNotice { } const XAI_UUID_RE = "[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}"; +// Background task ids: a bare UUID (monitors, subagents) or, when Grok +// auto-backgrounds a foreground shell command, its tool call id +// (`call--`). Matching only the UUID form left those commands running +// forever, so deferred finalize never ended the turn (board #16214). +const XAI_BACKGROUND_TASK_ID_RE = `(?:call-)?${XAI_UUID_RE}(?:-\\d+)?`; function unknownRecord(value: unknown): Record | undefined { return typeof value === "object" && value !== null && !Array.isArray(value) @@ -328,19 +333,35 @@ export function extractXAiBackgroundTaskCompletion(toolCall: AcpToolCallState): }> { const rawInput = unknownRecord(toolCall.data.rawInput); if (!isXAiGetSubagentOutputTool(toolCall, rawInput)) return []; + // Caller matches taskId against registered background tools (monitors). + // Subagent hydration stays on extractXAiAcpSubagentUpdate. + // One call can wait on several task_ids; Grok then reports each task in its + // own MultiResult entry and sends no Result or task header (board #16214). + const multiResults = multiResultsFromGetOutputTool(toolCall); + if (multiResults !== undefined) { + return multiResults.flatMap((entry) => { + const taskId = nonEmptyString(entry.task_id) ?? nonEmptyString(entry.taskId); + if (taskId === undefined || !new RegExp(`^${XAI_BACKGROUND_TASK_ID_RE}$`, "i").test(taskId)) { + return []; + } + const status = + toolCall.status === "failed" ? "failed" : (structuredTaskStatus(entry) ?? "running"); + return [{ taskId, status, appendOutput: structuredTaskOutput(entry) ?? "" }]; + }); + } const text = xAiToolOutputText(toolCall); const status = statusFromGetOutputTool(toolCall, text); const appendOutput = resultFromGetOutputTool(toolCall, text) ?? ""; // Prefer the completed identity from the result/header over the request list // (one call can poll multiple task_ids). - const resultTaskId = resultTaskIdFromGetOutputTool(toolCall, text); + const resultTaskId = resultTaskIdFromGetOutputTool(toolCall, text, XAI_BACKGROUND_TASK_ID_RE); if (resultTaskId !== undefined) { return [{ taskId: resultTaskId, status, appendOutput }]; } const requestIds: string[] = []; const push = (value: unknown) => { const id = nonEmptyString(value); - if (id !== undefined && new RegExp(`^${XAI_UUID_RE}$`, "i").test(id)) { + if (id !== undefined && new RegExp(`^${XAI_BACKGROUND_TASK_ID_RE}$`, "i").test(id)) { requestIds.push(id); } }; @@ -745,6 +766,7 @@ function taskIdsFromGetOutputTool( function resultTaskIdFromGetOutputTool( toolCall: AcpToolCallState, output: string | undefined, + idPattern: string = XAI_UUID_RE, ): string | undefined { const rawOutput = unknownRecord(toolCall.data.rawOutput); const result = unknownRecord(rawOutput?.Result) ?? unknownRecord(rawOutput?.result); @@ -753,35 +775,72 @@ function resultTaskIdFromGetOutputTool( nonEmptyString(result?.taskId) ?? nonEmptyString(rawOutput?.task_id) ?? nonEmptyString(rawOutput?.taskId); - if (structured !== undefined && new RegExp(`^${XAI_UUID_RE}$`, "i").test(structured)) { + if (structured !== undefined && new RegExp(`^${idPattern}$`, "i").test(structured)) { return structured; } if (output === undefined) return undefined; const taskHeader = firstUuidMatch( output, - new RegExp(`===\\s*Task\\s+(${XAI_UUID_RE})\\s*===`, "i"), + new RegExp(`===\\s*Task\\s+(${idPattern})\\s*===`, "i"), ); return taskHeader; } -function resultFromGetOutputTool( +/** The per-task entries of a multi-task get_output envelope, when present. */ +function multiResultsFromGetOutputTool( toolCall: AcpToolCallState, - output: string | undefined, -): string | null { +): ReadonlyArray> | undefined { const rawOutput = unknownRecord(toolCall.data.rawOutput); - const result = unknownRecord(rawOutput?.Result) ?? unknownRecord(rawOutput?.result); + const multi = unknownRecord(rawOutput?.MultiResult) ?? unknownRecord(rawOutput?.multiResult); + if (!Array.isArray(multi?.results)) return undefined; + return multi.results.flatMap((entry: unknown) => { + const record = unknownRecord(entry); + return record === undefined ? [] : [record]; + }); +} + +/** + * Output text of one structured task result. `undefined` when the result has no + * output field; `null` when it has one that is empty after cleaning. + */ +function structuredTaskOutput( + result: Record | undefined, +): string | null | undefined { const structured = nonEmptyString(result?.output) ?? nonEmptyString(result?.text) ?? decodeByteText(result?.output); - if (structured !== undefined) { - const cleaned = structured - .replace(/[\s\S]*?<\/subagent_meta>/gi, "") - .replace(/[\s\S]*?<\/subagent_result>/gi, "") - .trim(); - if (cleaned.length > 0) return cleaned; - return null; + if (structured === undefined) return undefined; + // Drop trailing machine meta blocks when present. + const cleaned = structured + .replace(/[\s\S]*?<\/subagent_meta>/gi, "") + .replace(/[\s\S]*?<\/subagent_result>/gi, "") + .trim(); + return cleaned.length > 0 ? cleaned : null; +} + +/** Task status from a structured result's status or exit code, if it has one. */ +function structuredTaskStatus( + result: Record | undefined, +): "running" | "completed" | "failed" | undefined { + const structuredStatus = nonEmptyString(result?.status)?.toLowerCase(); + if (structuredStatus === "completed" || structuredStatus === "success") return "completed"; + if (structuredStatus === "failed" || structuredStatus === "error") return "failed"; + if (structuredStatus === "running" || structuredStatus === "pending") return "running"; + if (typeof result?.exit_code === "number") { + return result.exit_code === 0 ? "completed" : "failed"; } + return undefined; +} + +function resultFromGetOutputTool( + toolCall: AcpToolCallState, + output: string | undefined, +): string | null { + const rawOutput = unknownRecord(toolCall.data.rawOutput); + const result = unknownRecord(rawOutput?.Result) ?? unknownRecord(rawOutput?.result); + const structured = structuredTaskOutput(result); + if (structured !== undefined) return structured; if (output === undefined) return null; const marker = output.match(/=== Output ===\s*([\s\S]*)$/i); const body = (marker?.[1] ?? output).trim(); @@ -795,13 +854,8 @@ function statusFromGetOutputTool( if (toolCall.status === "failed") return "failed"; const rawOutput = unknownRecord(toolCall.data.rawOutput); const result = unknownRecord(rawOutput?.Result) ?? unknownRecord(rawOutput?.result); - const structuredStatus = nonEmptyString(result?.status)?.toLowerCase(); - if (structuredStatus === "completed" || structuredStatus === "success") return "completed"; - if (structuredStatus === "failed" || structuredStatus === "error") return "failed"; - if (structuredStatus === "running" || structuredStatus === "pending") return "running"; - if (typeof result?.exit_code === "number") { - return result.exit_code === 0 ? "completed" : "failed"; - } + const structuredStatus = structuredTaskStatus(result); + if (structuredStatus !== undefined) return structuredStatus; if (output !== undefined) { if (/Status:\s*failed/i.test(output) || /Exit Code:\s*-?(?!0\b)\d+/i.test(output)) { return "failed"; diff --git a/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts b/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts index 705be8fb556b..b52f2a368668 100644 --- a/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts +++ b/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts @@ -906,7 +906,7 @@ layer("GitHubPullRequestApi.layer", (it) => { }), { concurrency: "unbounded" }, ).pipe(Effect.forkChild); - yield* TestClock.adjust("10 millis"); + yield* TestClock.adjust(`${GitHubPullRequestApi.SUMMARY_BATCH_WINDOW_MILLIS} millis`); const [seven, eight] = yield* Fiber.join(reads); assert.deepStrictEqual( @@ -946,6 +946,64 @@ layer("GitHubPullRequestApi.layer", (it) => { }), ); + it.effect("reads summaries that reach the resolver 65 ms apart in one document", () => + Effect.gen(function* () { + // Fork patch (board #15036): a pass's reads arrive spread out by their git lookups. + const node = (number: number) => ({ + number, + title: `Pull request ${number}`, + url: `https://github.com/acme/web/pull/${number}`, + author: { + __typename: "User", + login: "octocat", + name: "Octo Cat", + avatarUrl: "https://avatars.githubusercontent.com/u/1", + }, + baseRefName: "main", + headRefName: `feat/${number}`, + state: "OPEN", + isDraft: false, + mergeable: "MERGEABLE", + reviewDecision: null, + latestReviews: { nodes: [] }, + additions: 1, + deletions: 1, + changedFiles: 1, + updatedAt: "2026-08-24T12:34:56.000Z", + mergedAt: null, + closedAt: null, + commits: { nodes: [{ commit: { statusCheckRollup: { state: "SUCCESS" } } }] }, + }); + mockedExecute.mockReturnValueOnce( + Effect.succeed( + output( + encodeJson({ + data: { + _0_repository: { pullRequest: node(7) }, + _1_repository: { pullRequest: node(8) }, + }, + }), + ), + ), + ); + const cli = yield* GitHubPullRequestApi.GitHubPullRequestApi; + const read = (number: number) => + cli + .getPullRequestSummary({ cwd: "/w", repository: "acme/web", host: "github.com", number }) + .pipe(Effect.forkChild); + + const early = yield* read(7); + yield* TestClock.adjust("65 millis"); + const late = yield* read(8); + yield* TestClock.adjust(`${GitHubPullRequestApi.SUMMARY_BATCH_WINDOW_MILLIS} millis`); + + assert.strictEqual((yield* Fiber.join(early))?.headBranch, "feat/7"); + assert.strictEqual((yield* Fiber.join(late))?.headBranch, "feat/8"); + expect(mockedExecute).toHaveBeenCalledOnce(); + expect(varsAt(0)).toMatchObject({ _0_number: 7, _1_number: 8 }); + }), + ); + it.effect("fingerprints watched pull requests on one host in one read", () => Effect.gen(function* () { const node = (comments: number) => ({ @@ -982,7 +1040,7 @@ layer("GitHubPullRequestApi.layer", (it) => { }), { concurrency: "unbounded" }, ).pipe(Effect.forkChild); - yield* TestClock.adjust("10 millis"); + yield* TestClock.adjust(`${GitHubPullRequestApi.SUMMARY_BATCH_WINDOW_MILLIS} millis`); const [seven, eight] = yield* Fiber.join(reads); expect(seven?.remarks.startsWith("1 ")).toBe(true); @@ -1056,7 +1114,7 @@ layer("GitHubPullRequestApi.layer", (it) => { .pipe(Effect.result), { concurrency: "unbounded" }, ).pipe(Effect.forkChild); - yield* TestClock.adjust("10 millis"); + yield* TestClock.adjust(`${GitHubPullRequestApi.SUMMARY_BATCH_WINDOW_MILLIS} millis`); const [seven, eight] = yield* Fiber.join(reads); expect(Result.isFailure(seven!) ? seven.failure._tag : null).toBe("GitHubApiNotFoundError"); diff --git a/packages/source-control-github/src/server/GitHubPullRequestApi.ts b/packages/source-control-github/src/server/GitHubPullRequestApi.ts index 0e73babffc08..109d61b1d082 100644 --- a/packages/source-control-github/src/server/GitHubPullRequestApi.ts +++ b/packages/source-control-github/src/server/GitHubPullRequestApi.ts @@ -371,8 +371,14 @@ const STAT_ALIASES_PER_REQUEST = 25; * The background sync asks for every linked pull request at once, and each read reaches the * resolver after its own cache check, so a batch needs a moment longer than one scheduler * tick to gather them. + * + * Fork patch (board #15036): 250 ms, not 10. Reads whose repository identity was cached reach + * the resolver about 65 ms before reads that resolve it with git, so a 10 ms window split one + * pass into two documents on a third of the ticks of 2026-10-14. A summary read that reaches + * the host takes a second or more, so the wait adds little. */ -const SUMMARY_BATCH_WINDOW = "10 millis"; +export const SUMMARY_BATCH_WINDOW_MILLIS = 250; +const SUMMARY_BATCH_WINDOW = `${SUMMARY_BATCH_WINDOW_MILLIS} millis` as const; export interface GitHubPullRequestSearchBatch { readonly items: ReadonlyArray;