From 5df3b88ed07a4afbbaff692ac7d491baa3cbc2a8 Mon Sep 17 00:00:00 2001 From: Matthew Riddell Date: Fri, 9 Oct 2026 19:25:39 -0500 Subject: [PATCH 1/6] fix(server): keep PR-status lookups for ten minutes, not one (fork patch, board #15036) The settlement and thread-PR sweeps run every minute and re-read every open PR from GitHub once their one-minute cache entries expire: 500 to 1,100 GraphQL calls an hour on the fleet box (listPullRequestsByHead from ThreadPullRequestServiceV2.synchronize and ThreadSettlementServiceV2.sweep, getPullRequestSummary from PullRequestReadCache). - GitManager: PR_LOOKUP_CACHE_TTL 60 s -> 10 min; the no-open-PR TTL 5 min -> 10 min. Refresh paths (turn end, push, create PR, user refresh) still bypass the cache; failure backoff is unchanged. - PullRequestReadCache: entries and scope invalidations last ENTRY_TTL (10 min) instead of 60 s, so an invalidation still outlives any entry read before it. PR badges and external merges can lag up to ten minutes. Matt, q3440 (Telegram 2026-10-10T00:17:56Z): "q3440: 1". (cherry picked from commit 393c055cab6dfa880cdbfb3fade6762514bc8f15) --- apps/server/src/git/GitManager.test.ts | 95 +++++++++---------- apps/server/src/git/GitManager.ts | 15 ++- .../pullRequest/PullRequestReadCache.test.ts | 12 ++- .../src/pullRequest/PullRequestReadCache.ts | 17 +++- 4 files changed, 83 insertions(+), 56 deletions(-) diff --git a/apps/server/src/git/GitManager.test.ts b/apps/server/src/git/GitManager.test.ts index e1f705dea0e2e..ec30c8d742354 100644 --- a/apps/server/src/git/GitManager.test.ts +++ b/apps/server/src/git/GitManager.test.ts @@ -2186,57 +2186,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 +2276,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 fdc2bf2f88b73..4d3f9c651a938 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; diff --git a/apps/server/src/pullRequest/PullRequestReadCache.test.ts b/apps/server/src/pullRequest/PullRequestReadCache.test.ts index a4c0c8b178798..e5f14266a4ded 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"); @@ -222,7 +228,9 @@ it.layer(NodeServices.layer)("PR filesystem cache", (it) => { 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 16860c78ee9a2..04c4c39c16d3b 100644 --- a/apps/server/src/pullRequest/PullRequestReadCache.ts +++ b/apps/server/src/pullRequest/PullRequestReadCache.ts @@ -29,10 +29,19 @@ 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 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); type ReadError = PullRequestOperationError | PullRequestUnavailableError; const revisionCodec = Schema.fromJsonString( Schema.Record( @@ -105,7 +114,7 @@ export const make = Effect.gen(function* () { request.lookup.pipe( Effect.map((payload) => ({ payload, - expiresAt: clock.currentTimeMillisUnsafe() + 60_000, + expiresAt: clock.currentTimeMillisUnsafe() + ENTRY_TTL_MILLIS, revision: request.revision, })), ), @@ -160,7 +169,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); From 95d006291db2ef8628023aef2dcee3004f6d7330 Mon Sep 17 00:00:00 2001 From: Matthew Riddell Date: Fri, 9 Oct 2026 19:37:01 -0500 Subject: [PATCH 2/6] fix(grok): match auto-backgrounded shell task ids (call--) When Grok moves a foreground run_terminal_command to the background after its 15s limit, it reuses the tool call id (call--) as the task id. T3 registered that id as running background work, but TaskOutput hydration only accepted a bare UUID, so the completion never matched. The tool row stayed running, hasDeferredBackgroundWork stayed true and the turn never ended until the hung-turn timer stopped the thread. Accept the call-id form in extractXAiBackgroundTaskCompletion (result task_id, task header and request ids). Subagent and monitor parsing keep the UUID-only match. Evidence: lcr-bench 2026-10-08/09, 26 of 28 grok arms with call-id tasks hung; 0 of 13 arms with UUID or no background tasks hung. Board #16214. Co-Authored-By: Claude Opus 5.5 (cherry picked from commit 533aa6a797a88e218e33817fa70336def786bbc3) --- .../src/server/xaiAcpExtension.test.ts | 113 ++++++++++++++++++ .../src/server/xaiAcpExtension.ts | 14 ++- 2 files changed, 123 insertions(+), 4 deletions(-) diff --git a/packages/provider-grok/src/server/xaiAcpExtension.test.ts b/packages/provider-grok/src/server/xaiAcpExtension.test.ts index c31d9438972c4..5ec07c0f241e6 100644 --- a/packages/provider-grok/src/server/xaiAcpExtension.test.ts +++ b/packages/provider-grok/src/server/xaiAcpExtension.test.ts @@ -664,6 +664,119 @@ 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("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 c392b88cb0db9..a703484e77c8b 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) @@ -333,14 +338,14 @@ export function extractXAiBackgroundTaskCompletion(toolCall: AcpToolCallState): 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 +750,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,13 +759,13 @@ 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; } From 281cca8992dae0dbc4db0738b487568c2028bdaa Mon Sep 17 00:00:00 2001 From: Matthew Riddell Date: Mon, 12 Oct 2026 19:48:48 -0500 Subject: [PATCH 3/6] fix(server): cut background GitHub PR lookups (fork patch, board #15036) The 10-minute PR cache (393c055ca) did not cut t3-server's GraphQL reads: 2026-10-12 05-10Z still ran 106-147 calls an hour against a 283/day target. - The periodic thread PR pass and the settlement sweep run every 10 minutes, not every minute, so expired branch lookups fall due together and share GraphQL documents. Settlement lookups run 25 at a time to fill a document. - A finished run or captured checkpoint reads the cached branch answer; it no longer forces a fresh read or drops a cached "no pull request" answer. A pull request opened during a run shows within the cache's 10 minutes (Matt, q3440: PR badges can lag up to 10 min). User refreshes still bypass. - Provider-native subagents whose root thread is archived, settled or gone get no branch lookup (493 of 547 settlement candidates; 231 of 267 keys per pass). They still settle by age. - `git remote -v` lines of a partial clone end with `[blob:none]`; the remote regex rejected them, so the checkout resolved to provider "unknown" and its PR badges and merge settlement stopped. The three copies accept the suffix. Tests: GitManager, ThreadPullRequestService, ThreadSettlementService, RepositoryIdentityResolver, GitVcsDriver, GitVcsDriverCore: 362 passed (GIT_CONFIG_GLOBAL=/dev/null; eniac's global core.hooksPath fails 5 hook tests on the base too). New tests fail on the base: 3 of 3. Co-Authored-By: Claude Opus 5.5 (cherry picked from commit ebddc2c6d9f787cefb42bc35dfa1c907b18b5b96) --- apps/server/src/git/GitManager.test.ts | 10 +++- apps/server/src/git/GitManager.ts | 13 ++--- .../src/orchestration-v2/ProjectionStore.ts | 2 + .../ThreadPullRequestService.test.ts | 3 +- .../ThreadPullRequestService.ts | 36 ++++++++++++-- .../ThreadSettlementService.test.ts | 49 +++++++++++++++++++ .../ThreadSettlementService.ts | 46 +++++++++++++++-- .../RepositoryIdentityResolver.test.ts | 22 +++++++++ .../src/project/RepositoryIdentityResolver.ts | 3 +- apps/server/src/vcs/GitVcsDriver.ts | 3 +- apps/server/src/vcs/GitVcsDriverCore.ts | 3 +- 11 files changed, 171 insertions(+), 19 deletions(-) diff --git a/apps/server/src/git/GitManager.test.ts b/apps/server/src/git/GitManager.test.ts index ec30c8d742354..a6dfc69d67d34 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 }, diff --git a/apps/server/src/git/GitManager.ts b/apps/server/src/git/GitManager.ts index 4d3f9c651a938..dbad8357aa04f 100644 --- a/apps/server/src/git/GitManager.ts +++ b/apps/server/src/git/GitManager.ts @@ -1275,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 4222c3c27021a..8303c05474564 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/ThreadPullRequestService.test.ts b/apps/server/src/orchestration-v2/ThreadPullRequestService.test.ts index 25afb151e2808..704383167add6 100644 --- a/apps/server/src/orchestration-v2/ThreadPullRequestService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadPullRequestService.test.ts @@ -307,7 +307,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 21387243b46cb..fc2e92b038733 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, @@ -91,6 +92,24 @@ export function projectWorkspaceMatchesSnapshot( } 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 +180,13 @@ 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)) && ((thread.settledOverride !== "settled" && thread.settledAt === null) || request.threadId !== null || pendingBackfill.has(thread.id)) && @@ -367,8 +390,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 +403,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 +421,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 b4ca137f3f3ac..3b32e7a45f7c8 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 bba955cb2e05e..fe3e1ef5c6016 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 404b3c5b5b656..91bfabb879e2b 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 57084dd179f3c..474aaf3b45b61 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/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index 098b1ebcf6499..e4a3a8590f33c 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 db56c68b48ebf..f5982a45a70fc 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) { From 4b876341dd1e2f9dfc560062cd263ce8ebdedfd0 Mon Sep 17 00:00:00 2001 From: Matthew Riddell Date: Wed, 14 Oct 2026 00:34:51 -0500 Subject: [PATCH 4/6] fix(15036): PR sync pass every 10 min on a shared cache tick, not every minute PullRequestSyncReactor swept every minute. Open pull requests are always due, cache entries expired 10 min after their own read, and every finished run invalidates its project's entries, so each minute found a few stale PRs and sent them as their own GraphQL document: ~50/h, 1,213 getPullRequestSummary calls on 2026-10-13. - PullRequestReadCache: entries expire at the next 10-min tick since the epoch (entryExpiresAt), so entries read at different times go stale together. No entry lives longer than ENTRY_TTL. - PullRequestSyncReactor: the periodic pass runs 1 s after each tick (nextPassDelay), so it finds them stale together and reads them in shared documents of 25. Requested syncs still run at once. Badge data is at most one tick old when the pass writes it (Matt, q3440: PR badges can lag up to 10 min). Co-Authored-By: Claude Opus 5.5 (cherry picked from commit 0f53a26f77ceb39ac954ffe5433a13ae8d9ff7c7) --- .../PullRequestSyncReactor.test.ts | 42 +++++++++++++++++-- .../PullRequestSyncReactor.ts | 20 ++++++++- .../pullRequest/PullRequestReadCache.test.ts | 32 ++++++++++++++ .../src/pullRequest/PullRequestReadCache.ts | 14 ++++++- 4 files changed, 100 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 841e87c274fa3..702c8d7de9fa9 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; }); @@ -682,6 +683,7 @@ describe("PullRequestSyncReactor", () => { const reactor = yield* startAndSweep(fixture); yield* Ref.set(state, "merged"); + // The clock stays put: the next sweep is still ten minutes away. yield* Queue.offer(fixture.stateChanges, { host: "github.com", repository: "owner/repository", @@ -700,6 +702,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 +749,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 +1015,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 +1093,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" })), diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index c2cfe497d35ae..de829121c2277 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,20 @@ 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. Requested reads (a new link, a merge a reader saw, a run that + * ran `gh pr merge`) still run 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. */ const PULL_REQUEST_CLOSE_COMMAND = /\b(?:gh\s+pr|glab\s+mr)\s+(?:merge|close)\b/u; const isPullRequestProviderError = Schema.is(PullRequestProviderError); @@ -452,7 +467,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), ); }); diff --git a/apps/server/src/pullRequest/PullRequestReadCache.test.ts b/apps/server/src/pullRequest/PullRequestReadCache.test.ts index e5f14266a4ded..5f713057fa592 100644 --- a/apps/server/src/pullRequest/PullRequestReadCache.test.ts +++ b/apps/server/src/pullRequest/PullRequestReadCache.test.ts @@ -124,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; @@ -222,6 +252,8 @@ 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"); diff --git a/apps/server/src/pullRequest/PullRequestReadCache.ts b/apps/server/src/pullRequest/PullRequestReadCache.ts index 04c4c39c16d3b..7006bde46ce93 100644 --- a/apps/server/src/pullRequest/PullRequestReadCache.ts +++ b/apps/server/src/pullRequest/PullRequestReadCache.ts @@ -29,7 +29,7 @@ const CONCURRENT_READS = 512; const STORE_ID = "pr-v2"; const ENTRY_FILE_NAME = new RegExp(`^${STORE_ID}[0-9a-f]{64}$`); /** - * Entries expire ENTRY_TTL after they are written, and a write sets the file's + * 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. */ @@ -42,6 +42,16 @@ export const ENTRY_FILE_MAX_AGE = Duration.days(1); */ 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( @@ -114,7 +124,7 @@ export const make = Effect.gen(function* () { request.lookup.pipe( Effect.map((payload) => ({ payload, - expiresAt: clock.currentTimeMillisUnsafe() + ENTRY_TTL_MILLIS, + expiresAt: entryExpiresAt(clock.currentTimeMillisUnsafe()), revision: request.revision, })), ), From a2bc0d2bff5849a8db58f81739cd80a357e5a55c Mon Sep 17 00:00:00 2001 From: Matthew Riddell Date: Thu, 15 Oct 2026 01:56:07 -0500 Subject: [PATCH 5/6] fix(15036): one GitHub document per PR sync tick; no between-tick reads sync15036 (#7) made 507 GraphQL calls on 2026-10-14 (target 283): getPullRequestSummary 354, listPullRequestsByHead 154. - GitHubPullRequestApi: summary batch window 10 ms -> 250 ms (SUMMARY_BATCH_WINDOW_MILLIS). Reads whose repository identity was cached reached the resolver ~65 ms before the rest, so a third of the ticks sent two documents. - PullRequestSyncReactor: background re-reads (a state change a lookup or detail read saw, a new link without a snapshot, a run that ran `gh pr merge`) mark the key for the next pass instead of reading at once (DEFER_BACKGROUND_SYNC_TO_PASS). ~157 between-tick documents a day. A client's own refresh (ws invalidate / runAction) still reads at once. - ThreadPullRequestService: the periodic branch pass skips a thread whose branch pull request is known and open (hasOpenBranchPullRequest); the sync pass already reads it. Badge data stays at most one tick old (Matt, q3440: PR badges can lag up to 10 min). Co-Authored-By: Claude Opus 5.5 (cherry picked from commit 99539ef0995987f10959f892a4fa87833a5b5f98) --- .../PullRequestSyncReactor.test.ts | 125 +++++++++++------- .../PullRequestSyncReactor.ts | 36 +++-- .../ThreadPullRequestService.test.ts | 51 +++++++ .../ThreadPullRequestService.ts | 19 +++ .../src/server/GitHubPullRequestApi.test.ts | 56 +++++++- .../src/server/GitHubPullRequestApi.ts | 8 +- 6 files changed, 232 insertions(+), 63 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 702c8d7de9fa9..671bc88bfb9ee 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -659,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)); @@ -683,12 +683,19 @@ describe("PullRequestSyncReactor", () => { const reactor = yield* startAndSweep(fixture); yield* Ref.set(state, "merged"); - // The clock stays put: the next sweep is still ten minutes away. yield* Queue.offer(fixture.stateChanges, { host: "github.com", 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; @@ -1133,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 de829121c2277..0198d45293bd4 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -44,14 +44,24 @@ const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000; * 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. Requested reads (a new link, a merge a reader saw, a run that - * ran `gh pr merge`) still run at once. A badge's data is at most one tick old when the pass + * 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); @@ -408,7 +418,7 @@ export const make = Effect.gen(function* () { (link) => link.snapshot?.state === "open", ), ), - requestSync, + deferSync, { discard: true }, ), ), @@ -419,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", {})), ), ); @@ -438,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" && @@ -480,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 704383167add6..2387ccead4e12 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* () { diff --git a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts index fc2e92b038733..a124b6900481e 100644 --- a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts +++ b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts @@ -91,6 +91,24 @@ 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 @@ -187,6 +205,7 @@ export const make = Effect.gen(function* () { (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)) && diff --git a/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts b/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts index 705be8fb556bb..3614e1db6771c 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,56 @@ 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: null }, + 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: { s0: { pullRequest: node(7) }, s1: { 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({ s0_number: 7, s1_number: 8 }); + }), + ); + it.effect("fingerprints watched pull requests on one host in one read", () => Effect.gen(function* () { const node = (comments: number) => ({ @@ -982,7 +1032,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 +1106,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 0e73babffc082..109d61b1d0828 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; From 1a0b478539ed0c67447bb9d05c55140d9a62c71f Mon Sep 17 00:00:00 2001 From: Matthew Riddell Date: Thu, 15 Oct 2026 02:57:09 -0500 Subject: [PATCH 6/6] fix(grok): end each task of a multi-task TaskOutput wait (board #16214) One get_command_or_subagent_output call can wait on several task ids. Grok then answers with rawOutput.MultiResult.results[] (one entry per task, each with its own status, exit_code and output) and no Result or task header. The completion matcher read only Result, so every task in the wait was reported running, deferred finalize kept waiting, and the turn never ended. Live 2026-10-14 (lcr-bench fe-10784 grok-4.7-medium, T3 thread 1b658b27): the one post-restart Grok arm with call-id tasks that still hung, of 19. Read each MultiResult entry as its own task: its id (UUID or call--), its status or exit code, its output. An entry with no terminal status stays running. The single-Result path is unchanged. Co-Authored-By: Claude Opus 5.5 (cherry picked from commit 3d3222eee9669cf97ff28ba1c8205a635f914b69) --- .../src/server/xaiAcpExtension.test.ts | 62 ++++++++++++++ .../src/server/xaiAcpExtension.ts | 84 +++++++++++++++---- .../src/server/GitHubPullRequestApi.test.ts | 14 +++- 3 files changed, 139 insertions(+), 21 deletions(-) diff --git a/packages/provider-grok/src/server/xaiAcpExtension.test.ts b/packages/provider-grok/src/server/xaiAcpExtension.test.ts index 5ec07c0f241e6..e93b29afeb380 100644 --- a/packages/provider-grok/src/server/xaiAcpExtension.test.ts +++ b/packages/provider-grok/src/server/xaiAcpExtension.test.ts @@ -739,6 +739,68 @@ describe("XAiAcpExtension", () => { ).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", diff --git a/packages/provider-grok/src/server/xaiAcpExtension.ts b/packages/provider-grok/src/server/xaiAcpExtension.ts index a703484e77c8b..cabfab997fc86 100644 --- a/packages/provider-grok/src/server/xaiAcpExtension.ts +++ b/packages/provider-grok/src/server/xaiAcpExtension.ts @@ -333,6 +333,22 @@ 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) ?? ""; @@ -770,24 +786,61 @@ function resultTaskIdFromGetOutputTool( 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(); @@ -801,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 3614e1db6771c..b52f2a368668e 100644 --- a/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts +++ b/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts @@ -953,7 +953,12 @@ layer("GitHubPullRequestApi.layer", (it) => { number, title: `Pull request ${number}`, url: `https://github.com/acme/web/pull/${number}`, - author: { __typename: "User", login: "octocat", name: "Octo Cat", avatarUrl: null }, + author: { + __typename: "User", + login: "octocat", + name: "Octo Cat", + avatarUrl: "https://avatars.githubusercontent.com/u/1", + }, baseRefName: "main", headRefName: `feat/${number}`, state: "OPEN", @@ -973,7 +978,10 @@ layer("GitHubPullRequestApi.layer", (it) => { Effect.succeed( output( encodeJson({ - data: { s0: { pullRequest: node(7) }, s1: { pullRequest: node(8) } }, + data: { + _0_repository: { pullRequest: node(7) }, + _1_repository: { pullRequest: node(8) }, + }, }), ), ), @@ -992,7 +1000,7 @@ layer("GitHubPullRequestApi.layer", (it) => { 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({ s0_number: 7, s1_number: 8 }); + expect(varsAt(0)).toMatchObject({ _0_number: 7, _1_number: 8 }); }), );