From 564c71017f9dd8da30c0a03b33678d2a62832b14 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Tue, 27 Oct 2026 20:22:32 -0700 Subject: [PATCH 1/8] fix(server): watched PRs wake the agent when they merge A watched PR's merge ended the watch without telling the agent, so it could keep saying it was still watching a merged PR. A merge or close now wakes the agent once, naming the merge commit and, where the host reports it, who merged, and then ends the watch. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestWatchReactor.ts | 40 ++-- .../orchestration-v2/pullRequestWatch.test.ts | 40 ++++ .../src/orchestration-v2/pullRequestWatch.ts | 27 +++ .../src/orchestration-v2/runtimeLayer.test.ts | 196 ++++++++++++++++++ .../src/pullRequest/PullRequestService.ts | 4 + docs/user/source-control.md | 5 +- packages/contracts/src/pullRequest.ts | 2 + .../orchestrationV2PendingBackgroundWork.ts | 2 +- .../src/server/PullRequestProvider.ts | 2 + .../src/server/ForgejoPullRequestProvider.ts | 2 + .../ForgejoSourceControlProvider.test.ts | 25 ++- .../src/server/forgejoPullRequestJson.ts | 2 + .../server/GitCafePullRequestProvider.test.ts | 4 +- .../src/server/GitCafePullRequestProvider.ts | 1 + .../src/server/gitCafePullRequestJson.ts | 10 +- .../src/server/GitHubPullRequestApi.test.ts | 2 + .../src/server/gitHubPullRequestJson.test.ts | 22 ++ .../src/server/gitHubPullRequestJson.ts | 4 + .../src/server/pullRequestReads.graphql | 10 + .../src/server/pullRequestReads.graphql.ts | 76 ++++--- 20 files changed, 416 insertions(+), 60 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index ddfe74160459a..6a584aa3958b1 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -4,6 +4,7 @@ import { type OrchestrationV2Notification, type PullRequestActivity, type PullRequestComment, + type PullRequestDetail, type PullRequestRef, type PullRequestThreadCommentsResult, type ThreadPullRequestLink, @@ -41,6 +42,7 @@ import { combinePullRequestWatchWakes, evaluatePullRequestWatch, PULL_REQUEST_WATCH_NO_CHECKS_MS, + pullRequestWatchEndMessage, pullRequestWatchMessage, } from "./pullRequestWatch.ts"; @@ -517,30 +519,26 @@ export const make = Effect.gen(function* () { ); }); - const closed = (groupKey: string, target: WatchTarget) => + const finished = (groupKey: string, target: WatchTarget, detail: PullRequestDetail) => hold({ groupKey, target, next: null, - wake: { - text: `Pull request #${target.link.number} (${target.link.url}) was closed, so T3 Code stopped watching it. Call watch_pull_request if it reopens.`, - notification: { - source: { kind: "monitor" }, - outcome: "updated", - summary: `#${target.link.number}: closed, stopped watching`, - }, - }, - reason: "closed", + wake: pullRequestWatchEndMessage({ + number: target.link.number, + url: target.link.url, + detail, + }), + reason: detail.state === "merged" ? "merged" : "closed", }); - const endsWithoutRead = ({ thread, link }: WatchTarget): WatchEndReason | undefined => - link.snapshot?.state === "merged" - ? "merged" - : thread.settledOverride === "settled" || thread.settledAt !== null - ? "settled" - : thread.lineage.relationshipToParent === "subagent" - ? "subagent" - : undefined; + // A merged snapshot still gets one read, so the wake can name the merge commit. + const endsWithoutRead = ({ thread }: WatchTarget): WatchEndReason | undefined => + thread.settledOverride === "settled" || thread.settledAt !== null + ? "settled" + : thread.lineage.relationshipToParent === "subagent" + ? "subagent" + : undefined; const eachTarget = ( group: WatchGroup, @@ -652,11 +650,7 @@ export const make = Effect.gen(function* () { } if (detail.state !== "open") { lastReads.delete(group.key); - return yield* eachTarget(group, (target) => - detail.state === "closed" - ? closed(group.key, target) - : record(target, null).pipe(Effect.tap(() => ended(target, "merged"))), - ); + return yield* eachTarget(group, (target) => finished(group.key, target, detail)); } // Never advance the remark watermark past comments an incomplete read could have missed. diff --git a/apps/server/src/orchestration-v2/pullRequestWatch.test.ts b/apps/server/src/orchestration-v2/pullRequestWatch.test.ts index eb2ae2756d540..ad0a66e4b8860 100644 --- a/apps/server/src/orchestration-v2/pullRequestWatch.test.ts +++ b/apps/server/src/orchestration-v2/pullRequestWatch.test.ts @@ -8,6 +8,7 @@ import { assert, describe, it } from "@effect/vitest"; import { PULL_REQUEST_WATCH_WAKE_LIMIT, evaluatePullRequestWatch as evaluate, + pullRequestWatchEndMessage, pullRequestWatchMessage, } from "./pullRequestWatch.ts"; @@ -921,3 +922,42 @@ describe("evaluatePullRequestWatch", () => { assert.isTrue(unanswered(STARTED)); }); }); + +describe("pullRequestWatchEndMessage", () => { + const url = "https://git.cafe/pingdotgg/t3code/pull/7"; + const end = (overrides: Partial[0]["detail"]>) => + pullRequestWatchEndMessage({ + number: 7, + url, + detail: { state: "merged", baseBranch: "main", ...overrides }, + }); + const mergeCommitSha = "3b2f958603a51d8c42c4ede02eee7b8cb310b2d9"; + + it("says how the pull request ended, with the merge commit and merger the host reports", () => { + const merged = end({ + mergeCommitSha, + mergedBy: { login: "juliusmarminge", name: null, avatarUrl: null }, + }); + assert.equal( + merged.text, + `Pull request #7 (${url}) was merged into main by juliusmarminge as 3b2f958, so T3 Code stopped watching it.`, + ); + assert.deepEqual(merged.notification, { + source: { kind: "monitor" }, + outcome: "completed", + summary: "#7: merged, stopped watching", + }); + // GitCafe reports a merge made on GitHub without who made it. + assert.equal( + end({ mergeCommitSha }).text, + `Pull request #7 (${url}) was merged into main as 3b2f958, so T3 Code stopped watching it.`, + ); + + const closed = end({ state: "closed" }); + assert.equal( + closed.text, + `Pull request #7 (${url}) was closed without merging, so T3 Code stopped watching it. If it is reopened, call watch_pull_request to watch it again.`, + ); + assert.equal(closed.notification.summary, "#7: closed, stopped watching"); + }); +}); diff --git a/apps/server/src/orchestration-v2/pullRequestWatch.ts b/apps/server/src/orchestration-v2/pullRequestWatch.ts index e12d9a124df8a..adc694ca33b10 100644 --- a/apps/server/src/orchestration-v2/pullRequestWatch.ts +++ b/apps/server/src/orchestration-v2/pullRequestWatch.ts @@ -364,6 +364,33 @@ export function combinePullRequestWatchWakes( }; } +export function pullRequestWatchEndMessage(input: { + readonly number: number; + readonly url: string; + readonly detail: Pick; +}): { readonly text: string; readonly notification: OrchestrationV2Notification } { + const { detail } = input; + const merged = detail.state === "merged"; + const how = merged + ? [ + `was merged into ${detail.baseBranch}`, + detail.mergedBy ? ` by ${detail.mergedBy.login}` : "", + detail.mergeCommitSha ? ` as ${detail.mergeCommitSha.slice(0, 7)}` : "", + ].join("") + : "was closed without merging"; + return { + text: [ + `Pull request #${input.number} (${input.url}) ${how}, so T3 Code stopped watching it.`, + ...(merged ? [] : ["If it is reopened, call watch_pull_request to watch it again."]), + ].join(" "), + notification: { + source: { kind: "monitor" }, + outcome: merged ? "completed" : "updated", + summary: `#${input.number}: ${merged ? "merged" : "closed"}, stopped watching`, + }, + }; +} + export function pullRequestWatchMessage(input: { readonly number: number; readonly url: string; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 73f466c17542f..ffcdea078a862 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -3952,6 +3952,202 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + // regression: watched #26248 merged and its agent kept saying it was still watching + it.effect("wakes once when a watched pull request merges, naming the merge commit", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threadId = ThreadId.make("runtime-pull-request-watch-merged"); + const projectId = ProjectId.make("pr-watch-merged-project"); + yield* seedProject({ + projectId, + title: "Watch merged", + workspaceRoot: "/workspace/watch-merged", + defaultModelSelection: null, + createdAt: "2026-10-01T00:00:00.000Z", + }); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("pr-watch-merged-create"), + threadId, + projectId, + title: "Watch merged", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make("pr-watch-merged-start"), + threadId, + host: "github.com", + repository: "pingdotgg/t3code", + number: 15, + watching: true, + link: { url: "https://github.com/pingdotgg/t3code/pull/15", source: "agent" }, + }); + const reactor = yield* PullRequestWatchReactor.make.pipe( + Effect.provide( + Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(PullRequestService.PullRequestService)({ + detail: () => + Effect.succeed({ + ...watchedPullRequestDetail({ + projectId, + number: 15, + at: "2026-10-02T12:00:00.000Z", + }), + state: "merged" as const, + mergedAt: "2026-10-02T12:00:00.000Z", + mergeCommitSha: "3b2f958603a51d8c42c4ede02eee7b8cb310b2d9", + mergedBy: { login: "juliusmarminge", name: null, avatarUrl: null }, + }), + activity: () => + Effect.succeed({ + comments: [], + commentCount: 0, + commentsTruncated: false, + reviewThreads: [], + commits: [], + }), + }), + ), + ), + ); + yield* reactor.sweep; + yield* TestClock.adjust("10 minutes"); + yield* reactor.sweep; + + assert.isUndefined((yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch); + const wakes = (yield* orchestrator.getThreadRecords(threadId, ["messages"])).messages.filter( + (message) => message.notification != null, + ); + assert.deepEqual( + wakes.map((message) => message.notification?.summary), + ["#15: merged, stopped watching"], + ); + assert.include(wakes[0]!.text, "merged into main by juliusmarminge as 3b2f958"); + }), + ); + + it.effect("a pull request closed, reopened, and then merged wakes once for each end", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threadId = ThreadId.make("runtime-pull-request-watch-reopened"); + const projectId = ProjectId.make("pr-watch-reopened-project"); + yield* seedProject({ + projectId, + title: "Watch reopened", + workspaceRoot: "/workspace/watch-reopened", + defaultModelSelection: null, + createdAt: "2026-10-01T00:00:00.000Z", + }); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("pr-watch-reopened-create"), + threadId, + projectId, + title: "Watch reopened", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + const key = { host: "github.com", repository: "pingdotgg/t3code", number: 16 }; + const watch = (id: string) => + orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`pr-watch-reopened-${id}`), + threadId, + ...key, + watching: true, + link: { url: "https://github.com/pingdotgg/t3code/pull/16", source: "agent" }, + }); + yield* watch("start"); + + let state: PullRequestDetail["state"] = "closed"; + const host = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(PullRequestService.PullRequestService)({ + detail: () => + Effect.sync(() => ({ + ...watchedPullRequestDetail({ + projectId, + number: 16, + at: "2026-10-02T12:00:00.000Z", + }), + state, + checks: [{ name: "lint", status: "pending" as const, description: null, url: null }], + })), + activity: () => + Effect.succeed({ + comments: [], + commentCount: 0, + commentsTruncated: false, + reviewThreads: [], + commits: [], + }), + }), + ); + const summaries = Effect.map( + orchestrator.getThreadRecords(threadId, ["messages"]), + ({ messages }) => messages.flatMap((message) => message.notification?.summary ?? []), + ); + const watching = Effect.map( + orchestrator.getThreadShell(threadId), + (thread) => thread?.pullRequests?.[0]?.watch !== undefined, + ); + const reactor = yield* PullRequestWatchReactor.make.pipe(Effect.provide(host)); + yield* reactor.sweep; + assert.deepEqual(yield* summaries, ["#16: closed, stopped watching"]); + assert.isFalse(yield* watching); + + // Reopening does not restart the watch; the agent was told to watch again. + state = "open"; + yield* watch("again"); + yield* TestClock.adjust("10 minutes"); + yield* reactor.sweep; + assert.isTrue(yield* watching); + assert.deepEqual(yield* summaries, ["#16: closed, stopped watching"]); + + // The link's own sync can see the merge before the watch reads it. + state = "merged"; + yield* orchestrator.dispatch({ + type: "thread.pull-request-link.sync", + commandId: CommandId.make("pr-watch-reopened-synced"), + threadId, + ...key, + snapshot: { + state: "merged", + title: "Watched pull request", + headBranch: "feature", + baseBranch: "main", + isDraft: false, + updatedAt: null, + syncedAt: "2026-10-02T12:30:00.000Z", + mergedAt: "2026-10-02T12:30:00.000Z", + }, + stack: null, + }); + yield* TestClock.adjust("10 minutes"); + yield* reactor.sweep; + assert.deepEqual(yield* summaries, [ + "#16: closed, stopped watching", + "#16: merged, stopped watching", + ]); + assert.isFalse(yield* watching); + }), + ); + it.effect("ends a watch on a closed pull request whose activity cannot be read", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index 943fdc55d1d50..b2aebac03d7cd 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -1889,6 +1889,10 @@ export const make = Effect.gen(function* () { updatedAt: changeRequest.updatedAt, observedAt, mergedAt: changeRequest.mergedAt, + ...(changeRequest.mergeCommitSha + ? { mergeCommitSha: changeRequest.mergeCommitSha } + : {}), + ...(changeRequest.mergedBy ? { mergedBy: changeRequest.mergedBy } : {}), closedAt: changeRequest.closedAt, reviewers: changeRequest.reviewers, labels: changeRequest.labels, diff --git a/docs/user/source-control.md b/docs/user/source-control.md index 2ebd53ced8565..3a96039aaf0a1 100644 --- a/docs/user/source-control.md +++ b/docs/user/source-control.md @@ -331,8 +331,9 @@ check fails or waits for a maintainer's approval, the required checks pass, some conflict. On github.com it also wakes the agent, with GitHub's reason, when the merge queue drops the pull request without merging it. Threads in a project that watch the same pull request share one check. On GitHub, a check first asks whether anything changed and reads the pull request only when it did, which keeps -watching inside GitHub's rate limit. Comments from your own account do not wake it. Watching ends -when the pull request merges or closes, after 10 wakes in a row that bring only comments, after 8 +watching inside GitHub's rate limit. Comments from your own account do not wake it. When the pull +request merges or closes, the agent is woken once and watching ends; a reopened pull request needs a +new watch. Watching also ends after 10 wakes in a row that bring only comments, after 8 failed reads in a row, or when you press Stop on the thread. A rate limit only pauses watching. Settling or archiving a thread also ends all its watches. Unsettle the thread before starting a new watch. Watching a pull request again does not repeat check or conflict news its last diff --git a/packages/contracts/src/pullRequest.ts b/packages/contracts/src/pullRequest.ts index fb2c9b474268d..8760815fd6312 100644 --- a/packages/contracts/src/pullRequest.ts +++ b/packages/contracts/src/pullRequest.ts @@ -654,6 +654,8 @@ export const PullRequestDetail = Schema.Struct({ updatedAt: IsoDateTime, observedAt: Schema.optional(Schema.Finite), mergedAt: Schema.NullOr(IsoDateTime), + mergeCommitSha: Schema.optional(TrimmedNonEmptyString), + mergedBy: Schema.optional(PullRequestActor), closedAt: Schema.NullOr(IsoDateTime), reviewers: Schema.Array(PullRequestActor), labels: Schema.Array(PullRequestLabel), diff --git a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts index 3b25db88acecd..a2e433be69538 100644 --- a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts +++ b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts @@ -314,7 +314,7 @@ function pullRequestWatchTasks( pullRequests: ReadonlyArray | undefined, ): Array { return (pullRequests ?? []).flatMap((link) => - // The watch reactor ends a merged PR's watch without waking the agent, so it holds nothing. + // A merged PR's watch ends at the next sweep, whose wake starts its own run, so it holds nothing. link.watch === undefined || link.source === "stack-dismissed" || link.snapshot?.state === "merged" diff --git a/packages/source-control-core/src/server/PullRequestProvider.ts b/packages/source-control-core/src/server/PullRequestProvider.ts index 81d5e33d110f4..557efd77cdc67 100644 --- a/packages/source-control-core/src/server/PullRequestProvider.ts +++ b/packages/source-control-core/src/server/PullRequestProvider.ts @@ -203,6 +203,8 @@ export interface ProviderChangeRequestDetail extends ProviderChangeRequest { readonly body: string; readonly changedFiles: number; readonly mergedAt: string | null; + readonly mergeCommitSha?: string | null; + readonly mergedBy?: PullRequestActor | null; readonly closedAt: string | null; readonly reviewers: ReadonlyArray; readonly checks: ReadonlyArray; diff --git a/packages/source-control-forgejo/src/server/ForgejoPullRequestProvider.ts b/packages/source-control-forgejo/src/server/ForgejoPullRequestProvider.ts index 3a089e65c04bf..244c3a888feab 100644 --- a/packages/source-control-forgejo/src/server/ForgejoPullRequestProvider.ts +++ b/packages/source-control-forgejo/src/server/ForgejoPullRequestProvider.ts @@ -368,6 +368,8 @@ export const make = Effect.gen(function* () { return { ...forgejoChangeRequest(pr), headSha: pr.head.sha, + mergeCommitSha: pr.merge_commit_sha || null, + mergedBy: forgejoActor(pr.merged_by), body: pr.body ?? "", changedFiles: pr.changed_files ?? 0, reviewers: (pr.requested_reviewers ?? []).flatMap((user) => { diff --git a/packages/source-control-forgejo/src/server/ForgejoSourceControlProvider.test.ts b/packages/source-control-forgejo/src/server/ForgejoSourceControlProvider.test.ts index e68fbb50ec476..4a1e26f03a8cc 100644 --- a/packages/source-control-forgejo/src/server/ForgejoSourceControlProvider.test.ts +++ b/packages/source-control-forgejo/src/server/ForgejoSourceControlProvider.test.ts @@ -137,6 +137,7 @@ const requiredChecksInput = { const requiredChecksLayer = ( branch: Effect.Effect, statuses = requiredChecksStatuses, + pull: Readonly> = requiredChecksPullRequest, ) => Layer.mock(ForgejoCli.ForgejoCli)({ api: (input) => { @@ -145,7 +146,7 @@ const requiredChecksLayer = ( encodeJsonEffect(response).pipe(Effect.orDie, Effect.map(processOutput)); switch (path) { case "repos/acme/web/pulls/7": - return respond(requiredChecksPullRequest); + return respond(pull); case "repos/acme/web/statuses/head": return respond(new URLSearchParams(query).get("page") === "1" ? statuses : []); case "repos/acme/web/branches/release/1.x": @@ -420,6 +421,28 @@ it.effect.each([ }).pipe(Effect.provide(requiredChecksLayer(Effect.fail(cause)))); }); +it.effect("reads a merged pull request's merge commit and who merged it", () => + Effect.gen(function* () { + const provider = yield* ForgejoPullRequestProvider.make; + const detail = yield* provider.getChangeRequest(requiredChecksInput); + assert.strictEqual(detail.state, "merged"); + assert.strictEqual(detail.mergeCommitSha, "394a757942c5c3a9fb3cca31e348849e8c6a6dbe"); + assert.deepStrictEqual(detail.mergedBy, { login: "gusted", name: null, avatarUrl: null }); + }).pipe( + Effect.provide( + requiredChecksLayer(Effect.succeed({ name: "release/1.x" }), [], { + ...requiredChecksPullRequest, + state: "closed", + merged: true, + closed_at: "2026-10-27T00:08:58+01:00", + merged_at: "2026-10-27T00:08:58+01:00", + merge_commit_sha: "394a757942c5c3a9fb3cca31e348849e8c6a6dbe", + merged_by: { login: "gusted" }, + }), + ), + ), +); + it.effect("reads Forgejo jobs that did not run as finished checks", () => Effect.gen(function* () { const provider = yield* ForgejoPullRequestProvider.make; diff --git a/packages/source-control-forgejo/src/server/forgejoPullRequestJson.ts b/packages/source-control-forgejo/src/server/forgejoPullRequestJson.ts index 38ddae9ad40fa..d3c1fdbdbd199 100644 --- a/packages/source-control-forgejo/src/server/forgejoPullRequestJson.ts +++ b/packages/source-control-forgejo/src/server/forgejoPullRequestJson.ts @@ -60,6 +60,8 @@ export const ForgejoPullRequest = Schema.Struct({ updated_at: Schema.String, closed_at: Schema.NullOr(Schema.String), merged_at: Schema.NullOr(Schema.String), + merge_commit_sha: Schema.optional(Schema.NullOr(Schema.String)), + merged_by: Schema.optional(Schema.NullOr(ForgejoUser)), additions: Schema.optional(Schema.NullOr(Schema.Int)), deletions: Schema.optional(Schema.NullOr(Schema.Int)), changed_files: Schema.optional(Schema.NullOr(Schema.Int)), diff --git a/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.test.ts b/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.test.ts index cb2bf219c0b5e..4ddea55335bb2 100644 --- a/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.test.ts +++ b/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.test.ts @@ -1097,7 +1097,8 @@ describe("GitCafePullRequestProvider", () => { it.effect("reads a pull request GitHub merged as merged before the sync reports it", () => { const mergedAt = "2026-09-12T13:00:00Z"; - const server = detailWith({ providerMerge: { mergedAt } }); + const mergeCommitOid = "3b2f958603a51d8c42c4ede02eee7b8cb310b2d9"; + const server = detailWith({ providerMerge: { mergedAt, mergeCommitOid } }); return Effect.gen(function* () { const provider = yield* GitCafePullRequestProvider.make; const detail = yield* provider.getChangeRequest(target); @@ -1105,6 +1106,7 @@ describe("GitCafePullRequestProvider", () => { assert.strictEqual(detail.mergedAt, mergedAt); const preview = yield* provider.getChangeRequestPreview!(target); assert.strictEqual(preview.state, "merged"); + assert.strictEqual(detail.mergeCommitSha, mergeCommitOid); }).pipe(Effect.provide(server.layer)); }); diff --git a/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.ts b/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.ts index 314fc30902751..71ff3a317caa8 100644 --- a/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.ts +++ b/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.ts @@ -994,6 +994,7 @@ export const make = Effect.gen(function* () { changedFiles: changes?.changedFiles ?? 0, closedAt: pull.closedAt, mergedAt: pull.mergedAt, + mergeCommitSha: pull.mergeResultOid || pull.providerMerge?.mergeCommitOid || null, ...settledMerge(pull), reviewers: Json.toReviewers(listed, input.host), checks: checkList, diff --git a/packages/source-control-gitcafe/src/server/gitCafePullRequestJson.ts b/packages/source-control-gitcafe/src/server/gitCafePullRequestJson.ts index cc692637b82fe..a874a83a92c55 100644 --- a/packages/source-control-gitcafe/src/server/gitCafePullRequestJson.ts +++ b/packages/source-control-gitcafe/src/server/gitCafePullRequestJson.ts @@ -88,9 +88,17 @@ export const GitCafePullDetail = Schema.Struct({ mergeRoute: Schema.optional(Schema.Literals(["native", "provider", "unsupported"])), lockedAt: Schema.optional(Schema.NullOr(IsoDateTime)), viewerActorId: Schema.optional(Schema.NullOr(Schema.String)), + mergeResultOid: Schema.optional(Schema.NullOr(Schema.String)), // Set once GitHub has merged a pull request GitCafe routed there, which can be before the // sync brings the merged state back. - providerMerge: Schema.optional(Schema.NullOr(Schema.Struct({ mergedAt: IsoDateTime }))), + providerMerge: Schema.optional( + Schema.NullOr( + Schema.Struct({ + mergedAt: IsoDateTime, + mergeCommitOid: Schema.optional(Schema.NullOr(Schema.String)), + }), + ), + ), capabilities: Schema.Struct({ comment: Schema.Boolean, review: Schema.Boolean, diff --git a/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts b/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts index c83692d287fa1..a3c62e909c6bf 100644 --- a/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts +++ b/packages/source-control-github/src/server/GitHubPullRequestApi.test.ts @@ -76,6 +76,8 @@ const coreResponse = (pullRequest: Readonly> = {}) => ({ deletions: 0, changedFiles: 0, mergedAt: null, + mergeCommit: null, + mergedBy: null, closedAt: null, isCrossRepository: false, maintainerCanModify: true, diff --git a/packages/source-control-github/src/server/gitHubPullRequestJson.test.ts b/packages/source-control-github/src/server/gitHubPullRequestJson.test.ts index 5f084ac203fb6..37db822d34386 100644 --- a/packages/source-control-github/src/server/gitHubPullRequestJson.test.ts +++ b/packages/source-control-github/src/server/gitHubPullRequestJson.test.ts @@ -393,6 +393,8 @@ describe("pull request detail decoding", () => { createdAt: "2026-07-01T00:00:00Z", updatedAt: "2026-07-05T00:00:00Z", mergedAt: null, + mergeCommit: null, + mergedBy: null, closedAt: null, headRefName: "feat/detail", baseRefName: "main", @@ -551,6 +553,26 @@ describe("pull request detail decoding", () => { expect(detail.checksState).toBe("pending"); }); + it("reads a merged pull request's merge commit and who merged it", () => { + expect( + detailOf([], { + state: "MERGED", + mergedAt: "2026-07-05T00:00:00Z", + mergeCommit: { oid: "3b2f958603a51d8c42c4ede02eee7b8cb310b2d9" }, + mergedBy: { + __typename: "User", + login: "octocat", + avatarUrl: "https://avatars.githubusercontent.com/u/1", + name: "Octo Cat", + }, + }), + ).toMatchObject({ + state: "merged", + mergeCommitSha: "3b2f958603a51d8c42c4ede02eee7b8cb310b2d9", + mergedBy: { login: "octocat", name: "Octo Cat" }, + }); + }); + it("reads an auto-merge request and strategy, and its null as off", () => { expect(detailOf([], { autoMergeRequest: { mergeMethod: "SQUASH" } })).toMatchObject({ autoMergeEnabled: true, diff --git a/packages/source-control-github/src/server/gitHubPullRequestJson.ts b/packages/source-control-github/src/server/gitHubPullRequestJson.ts index 24cbd59260d62..6d3683d4de461 100644 --- a/packages/source-control-github/src/server/gitHubPullRequestJson.ts +++ b/packages/source-control-github/src/server/gitHubPullRequestJson.ts @@ -1031,6 +1031,8 @@ export interface GitHubPullRequestCore extends GitHubPullRequestDetail { readonly mergeQueueRemoval?: PullRequestMergeQueueRemoval; readonly maintainerCanModify: boolean; readonly canEditMaintainerEdits: boolean; + readonly mergeCommitSha?: string | null; + readonly mergedBy?: PullRequestActor | null; readonly reviewRequestTeams: ReadonlyArray; readonly viewerAccess: GitHubViewerAccess & GitHubRepositoryAccess; readonly comparison: GitHubBaseComparison | null; @@ -1094,6 +1096,8 @@ export function toPullRequestCore(result: PullRequestCore.Result): GitHubPullReq pr.headRepositoryOwner?.__typename === "User" && pr.headRepository !== null && pr.headRef !== null, + mergeCommitSha: trimmed(pr.mergeCommit?.oid), + mergedBy: toActor(pr.mergedBy), viewerAccess: { canWrite: toCanWrite(repository.viewerPermission), canDeleteBranch: toCanWrite(pr.headRepository?.viewerPermission) && pr.headRef !== null, diff --git a/packages/source-control-github/src/server/pullRequestReads.graphql b/packages/source-control-github/src/server/pullRequestReads.graphql index 88fbe80828e1e..2799b67525a27 100644 --- a/packages/source-control-github/src/server/pullRequestReads.graphql +++ b/packages/source-control-github/src/server/pullRequestReads.graphql @@ -83,6 +83,16 @@ query PullRequestCore( createdAt updatedAt mergedAt + mergeCommit { + oid + } + mergedBy { + login + avatarUrl + ... on User { + name + } + } closedAt headRefName baseRefName diff --git a/packages/source-control-github/src/server/pullRequestReads.graphql.ts b/packages/source-control-github/src/server/pullRequestReads.graphql.ts index a12334d8f98c3..ab5e211a4dec2 100644 --- a/packages/source-control-github/src/server/pullRequestReads.graphql.ts +++ b/packages/source-control-github/src/server/pullRequestReads.graphql.ts @@ -206,6 +206,35 @@ export declare namespace ActorAvatars { export type Result = typeof ActorAvatars.result.Type } +/** Represents a Git commit. */ +export const MergeCommitCommit = Schema.Struct({ + /** The Git object ID */ + oid: Shared.GitObjectID +}) +export interface MergeCommitCommit extends Schema.Struct.Type {} + +/** A user is an individual's account on GitHub that owns repositories and can make new content. */ +export const MergedByUser = Schema.Struct({ + __typename: Schema.Literal("User"), + /** The username used to login. */ + login: Schema.String, + /** A URL pointing to the user's public avatar. */ + avatarUrl: Shared.URI, + /** The user's public profile name. */ + name: Schema.NullOr(Schema.String) +}) +export interface MergedByUser extends Schema.Struct.Type {} + +/** Represents an object which can take actions on GitHub. Typically a User or Bot. */ +export const ActorOther = Schema.Struct({ + __typename: GraphQL.otherTypename()(["User"]), + /** The username of the actor. */ + login: Schema.String, + /** A URL pointing to the actor's public avatar. */ + avatarUrl: Shared.URI +}) +export interface ActorOther extends Schema.Struct.Type {} + /** Represents an owner of a Repository. */ export const RepositoryOwner = Schema.Struct({ __typename: GraphQL.otherTypename()([]), @@ -231,7 +260,7 @@ export const HeadRefRef = Schema.Struct({ export interface HeadRefRef extends Schema.Struct.Type {} /** A user is an individual's account on GitHub that owns repositories and can make new content. */ -export const PullRequestCoreRepositoryPullRequestAuthorUser = Schema.Struct({ +export const AuthorUser = Schema.Struct({ __typename: Schema.Literal("User"), /** The username used to login. */ login: Schema.String, @@ -242,17 +271,7 @@ export const PullRequestCoreRepositoryPullRequestAuthorUser = Schema.Struct({ /** The user's public profile name. */ name: Schema.NullOr(Schema.String) }) -export interface PullRequestCoreRepositoryPullRequestAuthorUser extends Schema.Struct.Type {} - -/** Represents an object which can take actions on GitHub. Typically a User or Bot. */ -export const ActorOther = Schema.Struct({ - __typename: GraphQL.otherTypename()(["User"]), - /** The username of the actor. */ - login: Schema.String, - /** A URL pointing to the actor's public avatar. */ - avatarUrl: Shared.URI -}) -export interface ActorOther extends Schema.Struct.Type {} +export interface AuthorUser extends Schema.Struct.Type {} /** Represents an auto-merge request for a pull request */ export const AutoMergeRequest = Schema.Struct({ @@ -459,6 +478,13 @@ export const PullRequestCoreRepositoryPullRequestPullRequest = Schema.Struct({ updatedAt: Shared.DateTime, /** The date and time that the pull request was merged. */ mergedAt: Schema.NullOr(Shared.DateTime), + /** The commit that was created when this pull request was merged. */ + mergeCommit: Schema.NullOr(MergeCommitCommit), + /** The actor who merged the pull request. */ + mergedBy: Schema.NullOr(Schema.Union([ + MergedByUser, + ActorOther + ])), /** Identifies the date and time when the object was closed. */ closedAt: Schema.NullOr(Shared.DateTime), /** Identifies the name of the head Ref associated with the pull request, even if the ref has been deleted. */ @@ -479,7 +505,7 @@ export const PullRequestCoreRepositoryPullRequestPullRequest = Schema.Struct({ headRef: Schema.NullOr(HeadRefRef), /** The actor who authored the comment. */ author: Schema.NullOr(Schema.Union([ - PullRequestCoreRepositoryPullRequestAuthorUser, + AuthorUser, ActorOther ])), /** Returns the auto-merge request object if one exists for this pull request. */ @@ -536,7 +562,7 @@ export const PullRequestCore = GraphQL.query("PullRequestCore", { { name: "mergeQueue", body: "Boolean!" } ], selections: [ - { name: "repository", body: ["(owner:$", "owner", " name:$", "name", "){mergeCommitAllowed squashMergeAllowed rebaseMergeAllowed autoMergeAllowed viewerPermission pullRequest(number:$", "number", "){number title url body state isDraft mergeable mergeStateStatus reviewDecision additions deletions changedFiles createdAt updatedAt mergedAt closedAt headRefName baseRefName headRefOid isCrossRepository maintainerCanModify headRepositoryOwner{__typename login}headRepository{nameWithOwner viewerPermission}headRef{id}author{__typename login avatarUrl...on User{id name}}autoMergeRequest{mergeMethod}mergeQueueRemovals:timelineItems(last:1 itemTypes:[REMOVED_FROM_MERGE_QUEUE_EVENT])@include(if:$", "mergeQueue", "){nodes{__typename...on RemovedFromMergeQueueEvent{id createdAt reason}}}viewerCanUpdate viewerDidAuthor viewerCanUpdateBranch baseRef{compare(headRef:$", "headRef", "){behindBy}}reviewRequests(first:$", "requestedReviewers", "){nodes{requestedReviewer{__typename...on User{login name}...on Bot{login}...on Team{slug combinedSlug teamName:name teamAvatarUrl:avatarUrl}}}}labels(first:100){nodes{name color description}}commits(last:1){nodes{commit{statusCheckRollup{contexts(first:100){nodes{__typename...", "CheckContext", "}pageInfo{hasNextPage endCursor}}}}}}}}"] } + { name: "repository", body: ["(owner:$", "owner", " name:$", "name", "){mergeCommitAllowed squashMergeAllowed rebaseMergeAllowed autoMergeAllowed viewerPermission pullRequest(number:$", "number", "){number title url body state isDraft mergeable mergeStateStatus reviewDecision additions deletions changedFiles createdAt updatedAt mergedAt mergeCommit{oid}mergedBy{__typename login avatarUrl...on User{name}}closedAt headRefName baseRefName headRefOid isCrossRepository maintainerCanModify headRepositoryOwner{__typename login}headRepository{nameWithOwner viewerPermission}headRef{id}author{__typename login avatarUrl...on User{id name}}autoMergeRequest{mergeMethod}mergeQueueRemovals:timelineItems(last:1 itemTypes:[REMOVED_FROM_MERGE_QUEUE_EVENT])@include(if:$", "mergeQueue", "){nodes{__typename...on RemovedFromMergeQueueEvent{id createdAt reason}}}viewerCanUpdate viewerDidAuthor viewerCanUpdateBranch baseRef{compare(headRef:$", "headRef", "){behindBy}}reviewRequests(first:$", "requestedReviewers", "){nodes{requestedReviewer{__typename...on User{login name}...on Bot{login}...on Team{slug combinedSlug teamName:name teamAvatarUrl:avatarUrl}}}}labels(first:100){nodes{name color description}}commits(last:1){nodes{commit{statusCheckRollup{contexts(first:100){nodes{__typename...", "CheckContext", "}pageInfo{hasNextPage endCursor}}}}}}}}"] } ], fragments: [ { name: "CheckContext", body: ["on StatusCheckRollupContext{__typename...on StatusContext{context state targetUrl createdAt description isRequired(pullRequestNumber:$", "number", ")@include(if:$", "required", ")}...on CheckRun{name status conclusion startedAt completedAt detailsUrl isRequired(pullRequestNumber:$", "number", ")@include(if:$", "required", ")checkSuite{workflowRun{workflow{name}}}}}"] } @@ -611,18 +637,6 @@ export declare namespace PullRequestCheckContexts { export type Result = typeof PullRequestCheckContexts.result.Type } -/** A user is an individual's account on GitHub that owns repositories and can make new content. */ -export const PullRequestPreviewRepositoryPullRequestAuthorUser = Schema.Struct({ - __typename: Schema.Literal("User"), - /** The username used to login. */ - login: Schema.String, - /** A URL pointing to the user's public avatar. */ - avatarUrl: Shared.URI, - /** The user's public profile name. */ - name: Schema.NullOr(Schema.String) -}) -export interface PullRequestPreviewRepositoryPullRequestAuthorUser extends Schema.Struct.Type {} - /** A repository pull request. */ export const PullRequestPreviewRepositoryPullRequestPullRequest = Schema.Struct({ /** Identifies the pull request number. */ @@ -639,7 +653,7 @@ export const PullRequestPreviewRepositoryPullRequestPullRequest = Schema.Struct( createdAt: Shared.DateTime, /** The actor who authored the comment. */ author: Schema.NullOr(Schema.Union([ - PullRequestPreviewRepositoryPullRequestAuthorUser, + MergedByUser, ActorOther ])) }) @@ -762,7 +776,7 @@ export const NodesIssueComment = Schema.Struct({ isMinimized: Schema.Boolean, /** The actor who authored the comment. */ author: Schema.NullOr(Schema.Union([ - PullRequestPreviewRepositoryPullRequestAuthorUser, + MergedByUser, ActorOther ])) }) @@ -795,7 +809,7 @@ export const NodesPullRequestReview = Schema.Struct({ isMinimized: Schema.Boolean, /** The actor who authored the comment. */ author: Schema.NullOr(Schema.Union([ - PullRequestPreviewRepositoryPullRequestAuthorUser, + MergedByUser, ActorOther ])) }) @@ -814,7 +828,7 @@ export interface PullRequestReviewConnection extends Schema.Struct.Type {} export const ReviewerCandidatesRepositoryPullRequestReviewRequestsNodesReviewRequest = Schema.Struct({ /** The reviewer that is requested. */ requestedReviewer: Schema.NullOr(Schema.Union([ - PullRequestPreviewRepositoryPullRequestAuthorUser, + MergedByUser, Team, NodesBot, RequestedReviewerOther From b976b007ddc196e1f631f3147249bbb9647f0958 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 12 Nov 2026 18:07:13 -0800 Subject: [PATCH 2/8] fix(server): an agent's own merge or close ends its watch without a wake The merge wake also fired when the thread's own agent had just merged the pull request with its host's CLI, costing a turn to tell it what it already knew. The watch reactor now follows the thread's merge and close commands: a pull request that ends while one runs, or since one ended after the last read that found it open, ends its watch quietly, and the command's end reads the thread's watches at once. A merge that lands later, as with `--auto`, still wakes. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestSyncReactor.ts | 10 +- .../PullRequestWatchReactor.ts | 94 +++++++++++- .../src/orchestration-v2/runtimeLayer.test.ts | 136 ++++++++++++++++++ .../sourceControl/sourceControlDefinitions.ts | 6 + docs/user/source-control.md | 4 +- 5 files changed, 240 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index b654364d3a7fb..e9b887d6c6e17 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -16,7 +16,10 @@ import { threadPullRequestKeysEqual, visibleThreadPullRequests, } from "../sourceControl/threadPullRequests.ts"; -import { sourceControlDefinitions } from "../sourceControl/sourceControlDefinitions.ts"; +import { + isChangeRequestCloseCommand, + sourceControlDefinitions, +} from "../sourceControl/sourceControlDefinitions.ts"; import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; @@ -48,11 +51,6 @@ const QUIET_SYNC_INTERVAL_MS = 5 * 60 * 1_000; // does merged ones. A reader that sees it open, or a requested refresh, still reads it at once. const CLOSED_FINAL_AFTER_MS = 24 * 60 * 60 * 1_000; -const isChangeRequestCloseCommand = (command: string) => - sourceControlDefinitions.definitions.some( - (definition) => definition.changeRequestCloseCommand?.test(command) === true, - ); - // Where a shell starts a command, past `NAME=value` assignments and wrappers like `timeout 60`. // A create command anywhere else, as in `echo "gh pr create"`, is text, not a command that runs. const COMMAND_START = diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index 6a584aa3958b1..7ce5254e68b31 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -1,5 +1,6 @@ import { CommandId, + isOrchestrationV2WorkActive, MessageId, type OrchestrationV2Notification, type PullRequestActivity, @@ -8,8 +9,10 @@ import { type PullRequestRef, type PullRequestThreadCommentsResult, type ThreadPullRequestLink, + type ThreadId, type ThreadPullRequestWatch, } from "@t3tools/contracts"; +import { isChangeRequestCloseCommand } from "../sourceControl/sourceControlDefinitions.ts"; import { normalizeThreadPullRequestKey, threadPullRequestKeyOf, @@ -38,6 +41,7 @@ import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import { forkParked } from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +import { isTerminalRunStatus } from "./ThreadManagementService.ts"; import { combinePullRequestWatchWakes, evaluatePullRequestWatch, @@ -287,6 +291,22 @@ export const make = Effect.gen(function* () { // A pass reads watches and records new ones, so passes must not overlap. const passes = Semaphore.makeUnsafe(1); const lastReads = new Map(); + /** When each group's last read that found its pull request open began. */ + const openAt = new Map(); + /** The merge and close commands each thread's agent ran with its host's CLI. */ + const agentEnds = new Map< + ThreadId, + { readonly running: Set; readonly finished: Set; endedAt: number } + >(); + // The agent needs no wake for its own merge or close. Only a command that ran since a read last + // found the pull request open can have ended it; a merge that lands later, as with `--auto`, is news. + const endedByAgent = (target: WatchTarget, openSince: number | undefined) => { + const ends = agentEnds.get(target.thread.id); + return ( + ends !== undefined && + (ends.running.size > 0 || ends.endedAt > (openSince ?? Date.parse(target.watch.startedAt))) + ); + }; const threadTails = new Map< string, { @@ -650,8 +670,16 @@ export const make = Effect.gen(function* () { } if (detail.state !== "open") { lastReads.delete(group.key); - return yield* eachTarget(group, (target) => finished(group.key, target, detail)); + const openSince = openAt.get(group.key); + openAt.delete(group.key); + const reason = detail.state === "merged" ? "merged" : "closed"; + return yield* eachTarget(group, (target) => + endedByAgent(target, openSince) + ? record(target, null).pipe(Effect.tap(() => ended(target, reason))) + : finished(group.key, target, detail), + ); } + openAt.set(group.key, now); // Never advance the remark watermark past comments an incomplete read could have missed. const remarks = Option.isSome(activity) @@ -749,9 +777,12 @@ export const make = Effect.gen(function* () { })); if (changed === undefined) { currentGroups = groups; - for (const cache of [readFailures, lastReads, threadTails]) { + for (const cache of [readFailures, lastReads, threadTails, openAt]) { for (const key of cache.keys()) if (!byPullRequest.has(key)) cache.delete(key); } + const watching = new Set(targets.map((target) => target.thread.id)); + for (const threadId of agentEnds.keys()) + if (!watching.has(threadId)) agentEnds.delete(threadId); } yield* Effect.forEach( ending, @@ -821,9 +852,68 @@ export const make = Effect.gen(function* () { Effect.withSpan("PullRequestWatchReactor.readOnChange"), ); + const readThreadWatches = (threadId: ThreadId) => + projections.getThreadsWithPullRequests(threadId).pipe( + Effect.flatMap((threads) => + Effect.forEach( + new Set( + threads.flatMap((thread) => + visibleThreadPullRequests(thread.pullRequests ?? []) + .filter((link) => link.watch !== undefined) + .map(threadPullRequestKeyOf), + ), + ), + readOnChange, + { discard: true }, + ), + ), + Effect.catchCause(logFailure("pull request watch read after a merge failed", { threadId })), + ); + const start: PullRequestWatchReactor["Service"]["start"] = Effect.fn( "PullRequestWatchReactor.start", )(function* () { + // Reading a thread's watches once its agent's merge or close ends lets them end at once, and + // without a wake, rather than at the next sweep. + const agentReads = yield* FiberMap.make(); + const commandEnded = Effect.fn("PullRequestWatchReactor.commandEnded")(function* ( + threadId: ThreadId, + ) { + const ends = agentEnds.get(threadId); + if (ends !== undefined) ends.endedAt = yield* Clock.currentTimeMillis; + yield* FiberMap.run(agentReads, threadId)(readThreadWatches(threadId)); + }); + yield* forkParked( + Stream.runForEach(engine.streamDomainEvents, (event) => { + if (event.type === "run.updated") { + const ends = agentEnds.get(event.threadId); + // A run can stop before its command reports that it ended. + if (ends === undefined || ends.running.size === 0) return Effect.void; + if (!isTerminalRunStatus(event.payload.status)) return Effect.void; + ends.running.clear(); + return commandEnded(event.threadId); + } + if (event.type !== "turn-item.updated") return Effect.void; + const item = event.payload; + if (item.type !== "command_execution" || !isChangeRequestCloseCommand(item.input)) { + return Effect.void; + } + const ends = agentEnds.get(event.threadId) ?? { + running: new Set(), + finished: new Set(), + endedAt: 0, + }; + agentEnds.set(event.threadId, ends); + if (ends.finished.has(item.id)) return Effect.void; + if (isOrchestrationV2WorkActive(item.status)) { + ends.running.add(item.id); + return Effect.void; + } + ends.running.delete(item.id); + ends.finished.add(item.id); + return commandEnded(event.threadId); + }).pipe(Effect.catchCause(logFailure("pull request watch command stream failed", {}))), + ); // One feed per pull request: the host keeps one subscription per pull request and account. const feeds = yield* FiberMap.make(); const follow = Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index ffcdea078a862..2d23839bd8a35 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -21,6 +21,7 @@ import { RuntimeRequestId, TurnItemId, type ModelSelection, + type OrchestrationV2DomainEvent, type OrchestrationV2Run, type OrchestrationV2ShellSnapshot, type OrchestrationV2RuntimeRequest, @@ -4035,6 +4036,141 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("ends a watch without a wake when the thread's own agent merged it", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threadId = ThreadId.make("runtime-pull-request-watch-own-merge"); + const projectId = ProjectId.make("pr-watch-own-merge-project"); + yield* seedProject({ + projectId, + title: "Watch own merge", + workspaceRoot: "/workspace/watch-own-merge", + defaultModelSelection: null, + createdAt: "2026-10-01T00:00:00.000Z", + }); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("pr-watch-own-merge-create"), + threadId, + projectId, + title: "Watch own merge", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + for (const number of [19, 20]) { + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`pr-watch-own-merge-start-${number}`), + threadId, + host: "github.com", + repository: "pingdotgg/t3code", + number, + watching: true, + link: { url: `https://github.com/pingdotgg/t3code/pull/${number}`, source: "agent" }, + }); + } + + const states = new Map([ + [19, "open"], + [20, "open"], + ]); + let fingerprints = 0; + const detailReads = yield* Queue.unbounded(); + const agentEvents = yield* Queue.unbounded(); + const host = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(PullRequestService.PullRequestService)({ + detail: (input) => + Queue.offer(detailReads, input.number).pipe( + Effect.as({ + ...watchedPullRequestDetail({ + projectId, + number: input.number, + at: "2026-10-02T12:00:00.000Z", + }), + state: states.get(input.number)!, + }), + ), + activity: () => + Effect.succeed({ + comments: [], + commentCount: 0, + commentsTruncated: false, + reviewThreads: [], + commits: [], + }), + // A fresh fingerprint every time, so every pass reads the detail. + watchFingerprint: () => + Effect.sync(() => ({ status: String(++fingerprints), remarks: "" })), + invalidate: () => Effect.void, + changes: () => Stream.never, + }), + Layer.succeed(Orchestrator.OrchestratorV2, { + ...orchestrator, + streamDomainEvents: Stream.fromQueue(agentEvents), + }), + ); + const runId = RunId.make(`run:${threadId}:1`); + const command = (status: "running" | "completed") => + Effect.map(DateTime.now, (at): OrchestrationV2DomainEvent => ({ + type: "turn-item.updated", + id: EventId.make(`event:${threadId}:merge-${status}`), + threadId, + runId, + occurredAt: at, + payload: { + id: TurnItemId.make(`item:${threadId}:merge`), + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status, + title: null, + startedAt: at, + completedAt: status === "completed" ? at : null, + updatedAt: at, + type: "command_execution", + input: "gh pr merge 19 --squash", + }, + })); + const reactor = yield* PullRequestWatchReactor.make.pipe(Effect.provide(host)); + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* reactor.start(); + // The first sweep finds both open. + yield* Queue.take(detailReads); + yield* Queue.take(detailReads); + + yield* Queue.offer(agentEvents, yield* command("running")); + states.set(19, "merged"); + yield* TestClock.adjust("1 second"); + // The command's end reads the thread's watches, so #19 ends before any sweep. + yield* Queue.offer(agentEvents, yield* command("completed")); + yield* Queue.take(detailReads); + yield* Queue.take(detailReads); + + // That read found #20 open, so someone else merging it later is news. + states.set(20, "merged"); + yield* reactor.sweep; + yield* worker.drain(); + const thread = yield* orchestrator.getThreadShell(threadId); + assert.isTrue(thread?.pullRequests?.every((link) => link.watch === undefined)); + const wakes = (yield* orchestrator.getThreadRecords(threadId, ["messages"])).messages.flatMap( + (message) => message.notification?.summary ?? [], + ); + assert.deepEqual(wakes, ["#20: merged, stopped watching"]); + }), + ); + it.effect("a pull request closed, reopened, and then merged wakes once for each end", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.ts b/apps/server/src/sourceControl/sourceControlDefinitions.ts index 8051ebfef6c6d..76970dab616d7 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.ts @@ -19,3 +19,9 @@ export const BUILT_IN_SOURCE_CONTROL_DRIVER_LIST = [ ] as const; export const sourceControlDefinitions = builtInSourceControlDefinitions; + +/** A host CLI command that merges or closes a change request, such as `gh pr merge`. */ +export const isChangeRequestCloseCommand = (command: string) => + sourceControlDefinitions.definitions.some( + (definition) => definition.changeRequestCloseCommand?.test(command) === true, + ); diff --git a/docs/user/source-control.md b/docs/user/source-control.md index 3a96039aaf0a1..f3844da1a5d9c 100644 --- a/docs/user/source-control.md +++ b/docs/user/source-control.md @@ -332,8 +332,8 @@ conflict. On github.com it also wakes the agent, with GitHub's reason, when the the pull request without merging it. Threads in a project that watch the same pull request share one check. On GitHub, a check first asks whether anything changed and reads the pull request only when it did, which keeps watching inside GitHub's rate limit. Comments from your own account do not wake it. When the pull -request merges or closes, the agent is woken once and watching ends; a reopened pull request needs a -new watch. Watching also ends after 10 wakes in a row that bring only comments, after 8 +request merges or closes, watching ends, and the agent is woken once unless it merged or closed the +pull request itself with the host's CLI. A reopened pull request needs a new watch. Watching also ends after 10 wakes in a row that bring only comments, after 8 failed reads in a row, or when you press Stop on the thread. A rate limit only pauses watching. Settling or archiving a thread also ends all its watches. Unsettle the thread before starting a new watch. Watching a pull request again does not repeat check or conflict news its last From f53e40cc88d9ced75eb5d3137af6cb9307d5fd15 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 12 Nov 2026 19:07:36 -0800 Subject: [PATCH 3/8] fix(server): an agent's merge silences only the pull requests its command named The own-merge rule was keyed on the thread and a time window, so a merge of one pull request could silence another one someone else closed, a failed merge command still counted, and the read at the command's end could skip a pull request whose fingerprint had not changed. A command now counts only while it runs and, if it succeeded, until a forced read of the thread's watches finishes, and one that names pull request numbers covers only those. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestWatchReactor.ts | 120 ++++---- .../src/orchestration-v2/runtimeLayer.test.ts | 257 +++++++++--------- .../sourceControlDefinitions.test.ts | 15 + .../sourceControl/sourceControlDefinitions.ts | 22 ++ 4 files changed, 228 insertions(+), 186 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index 7ce5254e68b31..ebf511da531c1 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -12,7 +12,10 @@ import { type ThreadId, type ThreadPullRequestWatch, } from "@t3tools/contracts"; -import { isChangeRequestCloseCommand } from "../sourceControl/sourceControlDefinitions.ts"; +import { + changeRequestsClosedBy, + isChangeRequestCloseCommand, +} from "../sourceControl/sourceControlDefinitions.ts"; import { normalizeThreadPullRequestKey, threadPullRequestKeyOf, @@ -291,22 +294,17 @@ export const make = Effect.gen(function* () { // A pass reads watches and records new ones, so passes must not overlap. const passes = Semaphore.makeUnsafe(1); const lastReads = new Map(); - /** When each group's last read that found its pull request open began. */ - const openAt = new Map(); - /** The merge and close commands each thread's agent ran with its host's CLI. */ - const agentEnds = new Map< - ThreadId, - { readonly running: Set; readonly finished: Set; endedAt: number } - >(); - // The agent needs no wake for its own merge or close. Only a command that ran since a read last - // found the pull request open can have ended it; a merge that lands later, as with `--auto`, is news. - const endedByAgent = (target: WatchTarget, openSince: number | undefined) => { - const ends = agentEnds.get(target.thread.id); - return ( - ends !== undefined && - (ends.running.size > 0 || ends.endedAt > (openSince ?? Date.parse(target.watch.startedAt))) + /** + * Each thread's merge and close commands, by item, with the pull requests each names. One counts + * while it runs and, if it succeeded, until the read at its end is done. + */ + const agentEnds = new Map | "any">>(); + // The agent needs no wake for a merge or close it made itself. Anything ending later, as a + // `--auto` merge does, is news. + const endedByAgent = (target: WatchTarget) => + [...(agentEnds.get(target.thread.id)?.values() ?? [])].some( + (named) => named === "any" || named.has(target.link.number), ); - }; const threadTails = new Map< string, { @@ -670,16 +668,13 @@ export const make = Effect.gen(function* () { } if (detail.state !== "open") { lastReads.delete(group.key); - const openSince = openAt.get(group.key); - openAt.delete(group.key); const reason = detail.state === "merged" ? "merged" : "closed"; return yield* eachTarget(group, (target) => - endedByAgent(target, openSince) + endedByAgent(target) ? record(target, null).pipe(Effect.tap(() => ended(target, reason))) : finished(group.key, target, detail), ); } - openAt.set(group.key, now); // Never advance the remark watermark past comments an incomplete read could have missed. const remarks = Option.isSome(activity) @@ -777,12 +772,9 @@ export const make = Effect.gen(function* () { })); if (changed === undefined) { currentGroups = groups; - for (const cache of [readFailures, lastReads, threadTails, openAt]) { + for (const cache of [readFailures, lastReads, threadTails]) { for (const key of cache.keys()) if (!byPullRequest.has(key)) cache.delete(key); } - const watching = new Set(targets.map((target) => target.thread.id)); - for (const threadId of agentEnds.keys()) - if (!watching.has(threadId)) agentEnds.delete(threadId); } yield* Effect.forEach( ending, @@ -852,18 +844,21 @@ export const make = Effect.gen(function* () { Effect.withSpan("PullRequestWatchReactor.readOnChange"), ); + // A full read, since an unchanged fingerprint or snapshot would otherwise skip the detail. const readThreadWatches = (threadId: ThreadId) => projections.getThreadsWithPullRequests(threadId).pipe( Effect.flatMap((threads) => Effect.forEach( - new Set( - threads.flatMap((thread) => - visibleThreadPullRequests(thread.pullRequests ?? []) - .filter((link) => link.watch !== undefined) - .map(threadPullRequestKeyOf), - ), + threads.flatMap((thread) => + visibleThreadPullRequests(thread.pullRequests ?? []) + .filter((link) => link.watch !== undefined) + .map((link) => [thread.projectId, threadPullRequestKeyOf(link)] as const), ), - readOnChange, + ([projectId, pullRequest]) => + Effect.suspend(() => { + lastReads.delete(`${projectId} ${pullRequest}`); + return readOnChange(pullRequest); + }), { discard: true }, ), ), @@ -873,45 +868,48 @@ export const make = Effect.gen(function* () { const start: PullRequestWatchReactor["Service"]["start"] = Effect.fn( "PullRequestWatchReactor.start", )(function* () { - // Reading a thread's watches once its agent's merge or close ends lets them end at once, and - // without a wake, rather than at the next sweep. - const agentReads = yield* FiberMap.make(); - const commandEnded = Effect.fn("PullRequestWatchReactor.commandEnded")(function* ( - threadId: ThreadId, - ) { - const ends = agentEnds.get(threadId); - if (ends !== undefined) ends.endedAt = yield* Clock.currentTimeMillis; - yield* FiberMap.run(agentReads, threadId)(readThreadWatches(threadId)); - }); + // Reading a thread's watches as its agent's merge or close ends lets them end at once. + const agentReads = yield* FiberMap.make(); + const finished = new Set(); yield* forkParked( Stream.runForEach(engine.streamDomainEvents, (event) => { if (event.type === "run.updated") { - const ends = agentEnds.get(event.threadId); // A run can stop before its command reports that it ended. - if (ends === undefined || ends.running.size === 0) return Effect.void; - if (!isTerminalRunStatus(event.payload.status)) return Effect.void; - ends.running.clear(); - return commandEnded(event.threadId); + const commands = agentEnds.get(event.threadId); + if (commands === undefined || !isTerminalRunStatus(event.payload.status)) { + return Effect.void; + } + for (const id of commands.keys()) if (!finished.has(id)) commands.delete(id); + if (commands.size === 0) agentEnds.delete(event.threadId); + return Effect.void; } if (event.type !== "turn-item.updated") return Effect.void; const item = event.payload; - if (item.type !== "command_execution" || !isChangeRequestCloseCommand(item.input)) { - return Effect.void; - } - const ends = agentEnds.get(event.threadId) ?? { - running: new Set(), - finished: new Set(), - endedAt: 0, - }; - agentEnds.set(event.threadId, ends); - if (ends.finished.has(item.id)) return Effect.void; - if (isOrchestrationV2WorkActive(item.status)) { - ends.running.add(item.id); + if ( + item.type !== "command_execution" || + finished.has(item.id) || + !isChangeRequestCloseCommand(item.input) + ) { return Effect.void; } - ends.running.delete(item.id); - ends.finished.add(item.id); - return commandEnded(event.threadId); + const commands = agentEnds.get(event.threadId) ?? new Map(); + agentEnds.set(event.threadId, commands); + commands.set(item.id, changeRequestsClosedBy(item.input)); + if (isOrchestrationV2WorkActive(item.status)) return Effect.void; + finished.add(item.id); + const done = Effect.sync(() => { + commands.delete(item.id); + if (commands.size === 0 && agentEnds.get(event.threadId) === commands) { + agentEnds.delete(event.threadId); + } + }); + return Effect.andThen( + item.status === "completed" ? Effect.void : done, + FiberMap.run( + agentReads, + item.id, + )(readThreadWatches(event.threadId).pipe(Effect.ensuring(done))), + ); }).pipe(Effect.catchCause(logFailure("pull request watch command stream failed", {}))), ); // One feed per pull request: the host keeps one subscription per pull request and account. diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 2d23839bd8a35..33c44b08f39c6 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4036,139 +4036,146 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); - it.effect("ends a watch without a wake when the thread's own agent merged it", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const threadId = ThreadId.make("runtime-pull-request-watch-own-merge"); - const projectId = ProjectId.make("pr-watch-own-merge-project"); - yield* seedProject({ - projectId, - title: "Watch own merge", - workspaceRoot: "/workspace/watch-own-merge", - defaultModelSelection: null, - createdAt: "2026-10-01T00:00:00.000Z", - }); - yield* orchestrator.dispatch({ - type: "thread.create", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make("pr-watch-own-merge-create"), - threadId, - projectId, - title: "Watch own merge", - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - }); - for (const number of [19, 20]) { + it.effect.each([ + // The agent's merge of #19 needs no wake, but its command did not close #20. + { status: "completed", wakes: ["#20: closed, stopped watching"] }, + // A merge command that failed ended nothing itself. + { + status: "failed", + wakes: ["#19: merged, stopped watching", "#20: closed, stopped watching"], + }, + ] as const)( + "ends a watch without a wake when the thread's own agent merged it ($status)", + ({ status: ended, wakes: expected }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threadId = ThreadId.make(`runtime-pull-request-watch-own-merge-${ended}`); + const projectId = ProjectId.make(`pr-watch-own-merge-${ended}-project`); + yield* seedProject({ + projectId, + title: "Watch own merge", + workspaceRoot: `/workspace/watch-own-merge-${ended}`, + defaultModelSelection: null, + createdAt: "2026-10-01T00:00:00.000Z", + }); yield* orchestrator.dispatch({ - type: "thread.pull-request.watch", - commandId: CommandId.make(`pr-watch-own-merge-start-${number}`), + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`pr-watch-own-merge-${ended}-create`), threadId, - host: "github.com", - repository: "pingdotgg/t3code", - number, - watching: true, - link: { url: `https://github.com/pingdotgg/t3code/pull/${number}`, source: "agent" }, + projectId, + title: "Watch own merge", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, }); - } + for (const number of [19, 20]) { + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`pr-watch-own-merge-${ended}-start-${number}`), + threadId, + host: "github.com", + repository: "pingdotgg/t3code", + number, + watching: true, + link: { url: `https://github.com/pingdotgg/t3code/pull/${number}`, source: "agent" }, + }); + } - const states = new Map([ - [19, "open"], - [20, "open"], - ]); - let fingerprints = 0; - const detailReads = yield* Queue.unbounded(); - const agentEvents = yield* Queue.unbounded(); - const host = Layer.mergeAll( - NodeServices.layer, - NodeCrypto.layer, - Layer.mock(PullRequestService.PullRequestService)({ - detail: (input) => - Queue.offer(detailReads, input.number).pipe( - Effect.as({ - ...watchedPullRequestDetail({ - projectId, - number: input.number, - at: "2026-10-02T12:00:00.000Z", + const states = new Map([ + [19, "open"], + [20, "open"], + ]); + const detailReads = yield* Queue.unbounded(); + const agentEvents = yield* Queue.unbounded(); + const host = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(PullRequestService.PullRequestService)({ + detail: (input) => + Queue.offer(detailReads, input.number).pipe( + Effect.as({ + ...watchedPullRequestDetail({ + projectId, + number: input.number, + at: "2026-10-02T12:00:00.000Z", + }), + state: states.get(input.number)!, }), - state: states.get(input.number)!, + ), + activity: () => + Effect.succeed({ + comments: [], + commentCount: 0, + commentsTruncated: false, + reviewThreads: [], + commits: [], }), - ), - activity: () => - Effect.succeed({ - comments: [], - commentCount: 0, - commentsTruncated: false, - reviewThreads: [], - commits: [], - }), - // A fresh fingerprint every time, so every pass reads the detail. - watchFingerprint: () => - Effect.sync(() => ({ status: String(++fingerprints), remarks: "" })), - invalidate: () => Effect.void, - changes: () => Stream.never, - }), - Layer.succeed(Orchestrator.OrchestratorV2, { - ...orchestrator, - streamDomainEvents: Stream.fromQueue(agentEvents), - }), - ); - const runId = RunId.make(`run:${threadId}:1`); - const command = (status: "running" | "completed") => - Effect.map(DateTime.now, (at): OrchestrationV2DomainEvent => ({ - type: "turn-item.updated", - id: EventId.make(`event:${threadId}:merge-${status}`), - threadId, - runId, - occurredAt: at, - payload: { - id: TurnItemId.make(`item:${threadId}:merge`), + // A fingerprint that never changes, so only a forced read sees an end. + watchFingerprint: () => Effect.succeed({ status: "steady", remarks: "" }), + invalidate: () => Effect.void, + changes: () => Stream.never, + }), + Layer.succeed(Orchestrator.OrchestratorV2, { + ...orchestrator, + streamDomainEvents: Stream.fromQueue(agentEvents), + }), + ); + const runId = RunId.make(`run:${threadId}:1`); + const command = (status: "running" | typeof ended) => + Effect.map(DateTime.now, (at): OrchestrationV2DomainEvent => ({ + type: "turn-item.updated", + id: EventId.make(`event:${threadId}:merge-${status}`), threadId, runId, - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status, - title: null, - startedAt: at, - completedAt: status === "completed" ? at : null, - updatedAt: at, - type: "command_execution", - input: "gh pr merge 19 --squash", - }, - })); - const reactor = yield* PullRequestWatchReactor.make.pipe(Effect.provide(host)); - const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - yield* reactor.start(); - // The first sweep finds both open. - yield* Queue.take(detailReads); - yield* Queue.take(detailReads); - - yield* Queue.offer(agentEvents, yield* command("running")); - states.set(19, "merged"); - yield* TestClock.adjust("1 second"); - // The command's end reads the thread's watches, so #19 ends before any sweep. - yield* Queue.offer(agentEvents, yield* command("completed")); - yield* Queue.take(detailReads); - yield* Queue.take(detailReads); - - // That read found #20 open, so someone else merging it later is news. - states.set(20, "merged"); - yield* reactor.sweep; - yield* worker.drain(); - const thread = yield* orchestrator.getThreadShell(threadId); - assert.isTrue(thread?.pullRequests?.every((link) => link.watch === undefined)); - const wakes = (yield* orchestrator.getThreadRecords(threadId, ["messages"])).messages.flatMap( - (message) => message.notification?.summary ?? [], - ); - assert.deepEqual(wakes, ["#20: merged, stopped watching"]); - }), + occurredAt: at, + payload: { + id: TurnItemId.make(`item:${threadId}:merge`), + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status, + title: null, + startedAt: at, + completedAt: status === "completed" ? at : null, + updatedAt: at, + type: "command_execution", + input: "gh pr merge 19 --squash", + }, + })); + const reactor = yield* PullRequestWatchReactor.make.pipe(Effect.provide(host)); + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* reactor.start(); + // The first sweep finds both open. + yield* Queue.take(detailReads); + yield* Queue.take(detailReads); + + // While the agent merges #19, someone else closes #20. + yield* Queue.offer(agentEvents, yield* command("running")); + states.set(19, "merged"); + states.set(20, "closed"); + // The command's end reads the thread's watches in full, before any sweep. + yield* Queue.offer(agentEvents, yield* command(ended)); + yield* Queue.take(detailReads); + yield* Queue.take(detailReads); + // A sweep waits for that read to finish. + yield* reactor.sweep; + yield* worker.drain(); + const thread = yield* orchestrator.getThreadShell(threadId); + assert.isTrue(thread?.pullRequests?.every((link) => link.watch === undefined)); + const wakes = (yield* orchestrator.getThreadRecords(threadId, [ + "messages", + ])).messages.flatMap((message) => message.notification?.summary ?? []); + assert.sameMembers(wakes, [...expected]); + }), ); it.effect("a pull request closed, reopened, and then merged wakes once for each end", () => diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.test.ts b/apps/server/src/sourceControl/sourceControlDefinitions.test.ts index 36996bdf3f370..3245fe9964392 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.test.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.test.ts @@ -2,6 +2,7 @@ import { expect, it } from "@effect/vitest"; import { BUILT_IN_SOURCE_CONTROL_DRIVER_LIST, + changeRequestsClosedBy, sourceControlDefinitions, } from "./sourceControlDefinitions.ts"; @@ -22,3 +23,17 @@ it.each([ ])("finds no host in %s, which git does not read as SSH", (remoteUrl) => { expect(sourceControlDefinitions.recognizeRemote(remoteUrl)).toBeNull(); }); + +it.each([ + ["gh pr merge 19 --squash", [19]], + ["gh pr merge --squash --match-head-commit 3b2f958 #19", [19]], + ["gh pr close https://github.com/o/r/pull/19 && gh pr merge 20", [19, 20]], + ["glab mr merge https://gitlab.com/o/r/-/merge_requests/7", [7]], + ["sleep 30; cafe pr merge 26271", [26271]], + ["gh pr merge --squash", "any"], + ["gh stack merge", "any"], + ["gh pr merge 19 && gh pr merge --auto", "any"], +] as const)("%s closes %j", (command, expected) => { + const closed = changeRequestsClosedBy(command); + expect(closed === "any" ? closed : [...closed]).toEqual(expected); +}); diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.ts b/apps/server/src/sourceControl/sourceControlDefinitions.ts index 76970dab616d7..3e5dcf1231abf 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.ts @@ -25,3 +25,25 @@ export const isChangeRequestCloseCommand = (command: string) => sourceControlDefinitions.definitions.some( (definition) => definition.changeRequestCloseCommand?.test(command) === true, ); + +const NAMED_CHANGE_REQUEST = /(?:^|\s)#?(\d+)(?=\s|$)|\/(?:pulls?|merge_requests)\/(\d+)/gu; + +/** + * The change request numbers a merge or close command names, or "any" when one names none, as a + * bare `gh pr merge` on the checked-out branch or `gh stack merge` does. + */ +export function changeRequestsClosedBy(command: string): ReadonlySet | "any" { + const named = new Set(); + for (const { changeRequestCloseCommand: close } of sourceControlDefinitions.definitions) { + if (close === undefined) continue; + for (const match of command.matchAll(new RegExp(close.source, `${close.flags}g`))) { + const args = command.slice(match.index + match[0].length).split(/[;&|\n]/u, 1)[0] ?? ""; + const numbers = [...args.matchAll(NAMED_CHANGE_REQUEST)].map(([, bare, url]) => + Number(bare ?? url), + ); + if (numbers.length === 0) return "any"; + for (const number of numbers) named.add(number); + } + } + return named.size === 0 ? "any" : named; +} From 68101f98adbea493329cb1ecca73e40978821310 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 12 Nov 2026 23:06:35 -0800 Subject: [PATCH 4/8] fix(server): an agent's merge read waits for any watch read in flight Reading a thread's watches when its agent's merge or close ends cleared the cached read before waiting for the pass in flight, which could then record its read from before the merge, so the agent's read skipped the host and the watch later woke the agent on its own merge. The cached read is now cleared once that pass is done. Azure's `az repos pr update --id 19 --status completed` named no pull request, because its close pattern runs through `--id`, so it silenced every watch on the thread. Numbers are now read from the start of the command, `--id=19` too. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestWatchReactor.ts | 8 ++--- .../src/orchestration-v2/runtimeLayer.test.ts | 33 +++++++++++++++---- .../sourceControlDefinitions.test.ts | 2 ++ .../sourceControl/sourceControlDefinitions.ts | 5 +-- 4 files changed, 36 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index ebf511da531c1..8a90109a67a99 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -855,10 +855,10 @@ export const make = Effect.gen(function* () { .map((link) => [thread.projectId, threadPullRequestKeyOf(link)] as const), ), ([projectId, pullRequest]) => - Effect.suspend(() => { - lastReads.delete(`${projectId} ${pullRequest}`); - return readOnChange(pullRequest); - }), + // After any pass in flight, which could record a read from before the merge. + passes + .withPermit(Effect.sync(() => lastReads.delete(`${projectId} ${pullRequest}`))) + .pipe(Effect.andThen(readOnChange(pullRequest))), { discard: true }, ), ), diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 33c44b08f39c6..95b7441220b67 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4090,21 +4090,25 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { [20, "open"], ]); const detailReads = yield* Queue.unbounded(); + const agentRead = yield* Deferred.make(); const agentEvents = yield* Queue.unbounded(); const host = Layer.mergeAll( NodeServices.layer, NodeCrypto.layer, Layer.mock(PullRequestService.PullRequestService)({ detail: (input) => - Queue.offer(detailReads, input.number).pipe( - Effect.as({ + Effect.sync(() => states.get(input.number)!).pipe( + Effect.tap(() => Queue.offer(detailReads, input.number)), + // Reads of an open pull request are slow, so the first sweep outlasts the merge. + Effect.tap((state) => (state === "open" ? Effect.sleep("1 second") : Effect.void)), + Effect.map((state) => ({ ...watchedPullRequestDetail({ projectId, number: input.number, at: "2026-10-02T12:00:00.000Z", }), - state: states.get(input.number)!, - }), + state, + })), ), activity: () => Effect.succeed({ @@ -4123,6 +4127,21 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { ...orchestrator, streamDomainEvents: Stream.fromQueue(agentEvents), }), + // Only the read at a command's end asks for one thread's watches. + Layer.effect( + ProjectionStore.ProjectionStoreV2, + Effect.map(ProjectionStore.ProjectionStoreV2, (store) => ({ + ...store, + getThreadsWithPullRequests: (thread?: ThreadId) => + store + .getThreadsWithPullRequests(thread) + .pipe( + Effect.tap(() => + thread === undefined ? Effect.void : Deferred.succeed(agentRead, undefined), + ), + ), + })), + ), ); const runId = RunId.make(`run:${threadId}:1`); const command = (status: "running" | typeof ended) => @@ -4154,7 +4173,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { const reactor = yield* PullRequestWatchReactor.make.pipe(Effect.provide(host)); const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; yield* reactor.start(); - // The first sweep finds both open. + // The first sweep has read both as open, and has not recorded that yet. yield* Queue.take(detailReads); yield* Queue.take(detailReads); @@ -4162,8 +4181,10 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { yield* Queue.offer(agentEvents, yield* command("running")); states.set(19, "merged"); states.set(20, "closed"); - // The command's end reads the thread's watches in full, before any sweep. + // The command's end reads the thread's watches in full, after that sweep. yield* Queue.offer(agentEvents, yield* command(ended)); + yield* Deferred.await(agentRead); + yield* TestClock.adjust("1 second"); yield* Queue.take(detailReads); yield* Queue.take(detailReads); // A sweep waits for that read to finish. diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.test.ts b/apps/server/src/sourceControl/sourceControlDefinitions.test.ts index 3245fe9964392..dfda1351ae44e 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.test.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.test.ts @@ -30,6 +30,8 @@ it.each([ ["gh pr close https://github.com/o/r/pull/19 && gh pr merge 20", [19, 20]], ["glab mr merge https://gitlab.com/o/r/-/merge_requests/7", [7]], ["sleep 30; cafe pr merge 26271", [26271]], + ["az repos pr update --id 19 --status completed", [19]], + ["az repos pr update --id=19 --status=abandoned", [19]], ["gh pr merge --squash", "any"], ["gh stack merge", "any"], ["gh pr merge 19 && gh pr merge --auto", "any"], diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.ts b/apps/server/src/sourceControl/sourceControlDefinitions.ts index 3e5dcf1231abf..90ab5b5385e90 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.ts @@ -26,7 +26,7 @@ export const isChangeRequestCloseCommand = (command: string) => (definition) => definition.changeRequestCloseCommand?.test(command) === true, ); -const NAMED_CHANGE_REQUEST = /(?:^|\s)#?(\d+)(?=\s|$)|\/(?:pulls?|merge_requests)\/(\d+)/gu; +const NAMED_CHANGE_REQUEST = /(?:^|[\s=])#?(\d+)(?=\s|$)|\/(?:pulls?|merge_requests)\/(\d+)/gu; /** * The change request numbers a merge or close command names, or "any" when one names none, as a @@ -37,7 +37,8 @@ export function changeRequestsClosedBy(command: string): ReadonlySet | " for (const { changeRequestCloseCommand: close } of sourceControlDefinitions.definitions) { if (close === undefined) continue; for (const match of command.matchAll(new RegExp(close.source, `${close.flags}g`))) { - const args = command.slice(match.index + match[0].length).split(/[;&|\n]/u, 1)[0] ?? ""; + // From the match's start, since Azure's match runs through `--id` to `--status`. + const args = command.slice(match.index).split(/[;&|\n]/u, 1)[0] ?? ""; const numbers = [...args.matchAll(NAMED_CHANGE_REQUEST)].map(([, bare, url]) => Number(bare ?? url), ); From 8399ab4bdba36cd1f309092d4eb32f1d950a9e5c Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 12 Nov 2026 23:13:59 -0800 Subject: [PATCH 5/8] fix(server): a quoted pull request number counts as named by a merge command `gh pr merge "19"`, and Codex's `/bin/zsh -lc "gh pr merge 19"`, named no pull request, so the merge silenced every watch on the thread instead of #19's. Co-Authored-By: Claude Opus 5.5 --- .../server/src/sourceControl/sourceControlDefinitions.test.ts | 2 ++ apps/server/src/sourceControl/sourceControlDefinitions.ts | 4 +++- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.test.ts b/apps/server/src/sourceControl/sourceControlDefinitions.test.ts index dfda1351ae44e..793c7bcbde7dd 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.test.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.test.ts @@ -26,6 +26,8 @@ it.each([ it.each([ ["gh pr merge 19 --squash", [19]], + ['gh pr merge "19" --squash', [19]], + ['/bin/zsh -lc "gh pr merge 19"', [19]], ["gh pr merge --squash --match-head-commit 3b2f958 #19", [19]], ["gh pr close https://github.com/o/r/pull/19 && gh pr merge 20", [19, 20]], ["glab mr merge https://gitlab.com/o/r/-/merge_requests/7", [7]], diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.ts b/apps/server/src/sourceControl/sourceControlDefinitions.ts index 90ab5b5385e90..8d5599c31727a 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.ts @@ -26,7 +26,9 @@ export const isChangeRequestCloseCommand = (command: string) => (definition) => definition.changeRequestCloseCommand?.test(command) === true, ); -const NAMED_CHANGE_REQUEST = /(?:^|[\s=])#?(\d+)(?=\s|$)|\/(?:pulls?|merge_requests)\/(\d+)/gu; +// Quotes included, since Codex runs `/bin/zsh -lc "gh pr merge 19"`. +const NAMED_CHANGE_REQUEST = + /(?:^|[\s=])["']?#?(\d+)["']?(?=\s|$)|\/(?:pulls?|merge_requests)\/(\d+)/gu; /** * The change request numbers a merge or close command names, or "any" when one names none, as a From 64e894e98fca2478e39513057ec2f2e84f067e6f Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 12 Nov 2026 23:36:15 -0800 Subject: [PATCH 6/8] fix(server): an agent's merge counts only when it exited zero, and only for whole-word numbers A merge command a provider reported as completed with a nonzero exit code still counted as the agent's own, so it could silence another account's merge or close. It now counts only when it exited zero, or, with no exit code, when its output did not indicate failure, the same rule the wire projection uses. The pull requests a command names are now read from its shell words, so a number inside `--subject "Release 20"`, a quoted URL in `--body`, or a comment after `#` no longer counts as a pull request the command closed. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestWatchReactor.ts | 8 ++++++- .../src/orchestration-v2/runtimeLayer.test.ts | 24 ++++++++++++------- .../sourceControlDefinitions.test.ts | 4 ++++ .../sourceControl/sourceControlDefinitions.ts | 18 +++++++++----- 4 files changed, 39 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index 8a90109a67a99..21e5327c68227 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -903,8 +903,14 @@ export const make = Effect.gen(function* () { agentEnds.delete(event.threadId); } }); + // Some providers report a command that exited nonzero as completed. + const succeeded = + item.status === "completed" && + (item.exitCode === undefined + ? item.outputIndicatesFailure !== true + : item.exitCode === 0); return Effect.andThen( - item.status === "completed" ? Effect.void : done, + succeeded ? Effect.void : done, FiberMap.run( agentReads, item.id, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 95b7441220b67..2d1c4b9cfa021 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4038,23 +4038,30 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect.each([ // The agent's merge of #19 needs no wake, but its command did not close #20. - { status: "completed", wakes: ["#20: closed, stopped watching"] }, + { name: "completed", status: "completed", wakes: ["#20: closed, stopped watching"] }, // A merge command that failed ended nothing itself. { + name: "failed", status: "failed", wakes: ["#19: merged, stopped watching", "#20: closed, stopped watching"], }, + { + name: "exit-1", + status: "completed", + exitCode: 1, + wakes: ["#19: merged, stopped watching", "#20: closed, stopped watching"], + }, ] as const)( - "ends a watch without a wake when the thread's own agent merged it ($status)", - ({ status: ended, wakes: expected }) => + "ends a watch without a wake when the thread's own agent merged it ($name)", + ({ name: key, status: ended, wakes: expected, ...result }) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; - const threadId = ThreadId.make(`runtime-pull-request-watch-own-merge-${ended}`); - const projectId = ProjectId.make(`pr-watch-own-merge-${ended}-project`); + const threadId = ThreadId.make(`runtime-pull-request-watch-own-merge-${key}`); + const projectId = ProjectId.make(`pr-watch-own-merge-${key}-project`); yield* seedProject({ projectId, title: "Watch own merge", - workspaceRoot: `/workspace/watch-own-merge-${ended}`, + workspaceRoot: `/workspace/watch-own-merge-${key}`, defaultModelSelection: null, createdAt: "2026-10-01T00:00:00.000Z", }); @@ -4062,7 +4069,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { type: "thread.create", createdBy: "user", creationSource: "web", - commandId: CommandId.make(`pr-watch-own-merge-${ended}-create`), + commandId: CommandId.make(`pr-watch-own-merge-${key}-create`), threadId, projectId, title: "Watch own merge", @@ -4075,7 +4082,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { for (const number of [19, 20]) { yield* orchestrator.dispatch({ type: "thread.pull-request.watch", - commandId: CommandId.make(`pr-watch-own-merge-${ended}-start-${number}`), + commandId: CommandId.make(`pr-watch-own-merge-${key}-start-${number}`), threadId, host: "github.com", repository: "pingdotgg/t3code", @@ -4168,6 +4175,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { updatedAt: at, type: "command_execution", input: "gh pr merge 19 --squash", + ...(status === "running" ? {} : result), }, })); const reactor = yield* PullRequestWatchReactor.make.pipe(Effect.provide(host)); diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.test.ts b/apps/server/src/sourceControl/sourceControlDefinitions.test.ts index 793c7bcbde7dd..85866fd71a792 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.test.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.test.ts @@ -28,6 +28,10 @@ it.each([ ["gh pr merge 19 --squash", [19]], ['gh pr merge "19" --squash', [19]], ['/bin/zsh -lc "gh pr merge 19"', [19]], + ['gh pr merge 19 --squash --subject "Release 20"', [19]], + [`/bin/zsh -lc "gh pr merge 19 --subject 'Release 20'"`, [19]], + ["gh pr merge 19 --body 'see https://github.com/o/r/pull/20'", [19]], + ["gh pr merge 19 # then 20", [19]], ["gh pr merge --squash --match-head-commit 3b2f958 #19", [19]], ["gh pr close https://github.com/o/r/pull/19 && gh pr merge 20", [19, 20]], ["glab mr merge https://gitlab.com/o/r/-/merge_requests/7", [7]], diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.ts b/apps/server/src/sourceControl/sourceControlDefinitions.ts index 8d5599c31727a..ee14a16b6384b 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.ts @@ -26,9 +26,11 @@ export const isChangeRequestCloseCommand = (command: string) => (definition) => definition.changeRequestCloseCommand?.test(command) === true, ); -// Quotes included, since Codex runs `/bin/zsh -lc "gh pr merge 19"`. -const NAMED_CHANGE_REQUEST = - /(?:^|[\s=])["']?#?(\d+)["']?(?=\s|$)|\/(?:pulls?|merge_requests)\/(\d+)/gu; +// A quoted or bare shell word. An unclosed quote, as Codex's `/bin/zsh -lc "gh pr merge 19"` +// leaves after the match, runs to the end. +const SHELL_WORD = /"((?:\\[\s\S]|[^"\\])*)"?|'([^']*)'?|((?:\\[\s\S]|[^\s"'\\])+)/gu; +// A whole word that names one, so a number inside `--subject "Release 20"` does not. +const NAMED_CHANGE_REQUEST = /^(?:--?[\w-]+=)?#?(\d+)$|^\S*\/(?:pulls?|merge_requests)\/(\d+)\/?$/u; /** * The change request numbers a merge or close command names, or "any" when one names none, as a @@ -41,9 +43,13 @@ export function changeRequestsClosedBy(command: string): ReadonlySet | " for (const match of command.matchAll(new RegExp(close.source, `${close.flags}g`))) { // From the match's start, since Azure's match runs through `--id` to `--status`. const args = command.slice(match.index).split(/[;&|\n]/u, 1)[0] ?? ""; - const numbers = [...args.matchAll(NAMED_CHANGE_REQUEST)].map(([, bare, url]) => - Number(bare ?? url), - ); + const numbers: Array = []; + for (const [, double, single, bare] of args.matchAll(SHELL_WORD)) { + // A comment ends the command. + if (bare !== undefined && /^#(?!\d+$)/u.test(bare)) break; + const [, number, url] = NAMED_CHANGE_REQUEST.exec(double ?? single ?? bare ?? "") ?? []; + if (number !== undefined || url !== undefined) numbers.push(Number(number ?? url)); + } if (numbers.length === 0) return "any"; for (const number of numbers) named.add(number); } From 0b3d1d8a806b4ebab9c97583fd7fccd8563647ca Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 12 Nov 2026 23:48:35 -0800 Subject: [PATCH 7/8] fix(server): a merge whose output shows it failed does not count as the agent's own The watch reactor read `outputIndicatesFailure`, which only the wire projection sets, so a command a provider reported as completed with no exit code and failure text in its output still counted as the agent's own merge. Both now use `commandIndicatesFailure`, which reads the exit code, or failing that, the output. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/PullRequestWatchReactor.ts | 10 +++------- apps/server/src/orchestration-v2/WireProjection.ts | 9 ++------- .../src/orchestration-v2/runtimeLayer.test.ts | 6 ++++++ packages/shared/src/toolOutput.ts | 13 +++++++++++++ 4 files changed, 24 insertions(+), 14 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index 21e5327c68227..03dc508dfe61a 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -12,6 +12,7 @@ import { type ThreadId, type ThreadPullRequestWatch, } from "@t3tools/contracts"; +import { commandIndicatesFailure } from "@t3tools/shared/toolOutput"; import { changeRequestsClosedBy, isChangeRequestCloseCommand, @@ -903,14 +904,9 @@ export const make = Effect.gen(function* () { agentEnds.delete(event.threadId); } }); - // Some providers report a command that exited nonzero as completed. - const succeeded = - item.status === "completed" && - (item.exitCode === undefined - ? item.outputIndicatesFailure !== true - : item.exitCode === 0); + // Some providers report a command that failed as completed. return Effect.andThen( - succeeded ? Effect.void : done, + item.status === "completed" && !commandIndicatesFailure(item) ? Effect.void : done, FiberMap.run( agentReads, item.id, diff --git a/apps/server/src/orchestration-v2/WireProjection.ts b/apps/server/src/orchestration-v2/WireProjection.ts index 7d4805103c372..f3652686870ed 100644 --- a/apps/server/src/orchestration-v2/WireProjection.ts +++ b/apps/server/src/orchestration-v2/WireProjection.ts @@ -16,10 +16,10 @@ import { dedupeTurnItemBodies, } from "@t3tools/shared/orchestrationV2DedupedBodies"; import { + commandIndicatesFailure, compactDynamicToolOutput, omitToolOutputImageData, toolOutputImages, - toolOutputIndicatesFailure, } from "@t3tools/shared/toolOutput"; // A checkpoint over vendored or generated trees can list 15k+ files; the card needs a prefix. @@ -166,12 +166,7 @@ export function projectTurnItemForWire(item: OrchestrationV2TurnItem): Orchestra const { output, ...projected } = item; // A successful search can print "ENOENT" or "exit code 2", so output text // decides only when the provider reported no exit code. - const failed = - item.exitCode === undefined - ? item.outputIndicatesFailure === true || - (output !== undefined && - toolOutputIndicatesFailure(output.slice(0, MAX_DETAIL_STRING_BYTES))) - : item.exitCode !== 0; + const failed = commandIndicatesFailure(item); return { ...projected, ...(failed ? { outputIndicatesFailure: true } : {}), diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 2d1c4b9cfa021..249bd8e086ba9 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4051,6 +4051,12 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { exitCode: 1, wakes: ["#19: merged, stopped watching", "#20: closed, stopped watching"], }, + { + name: "failure-output", + status: "completed", + output: "GraphQL: Pull request is not mergeable\n", + wakes: ["#19: merged, stopped watching", "#20: closed, stopped watching"], + }, ] as const)( "ends a watch without a wake when the thread's own agent merged it ($name)", ({ name: key, status: ended, wakes: expected, ...result }) => diff --git a/packages/shared/src/toolOutput.ts b/packages/shared/src/toolOutput.ts index f2773b0045e4c..65f46d0ef99d1 100644 --- a/packages/shared/src/toolOutput.ts +++ b/packages/shared/src/toolOutput.ts @@ -246,6 +246,19 @@ export function toolOutputIndicatesFailure(text: string): boolean { ); } +/** Whether a command failed. A reported exit code outranks text in its output. */ +export function commandIndicatesFailure(command: { + readonly exitCode?: number | undefined; + readonly outputIndicatesFailure?: boolean | undefined; + readonly output?: string | undefined; +}): boolean { + if (command.exitCode !== undefined) return command.exitCode !== 0; + return ( + command.outputIndicatesFailure === true || + (command.output !== undefined && toolOutputIndicatesFailure(command.output.slice(0, 32_768))) + ); +} + export interface ToolOutputImage { readonly mimeType: string; readonly data?: string; From 892e9412eab8e771b4de9ce5dcf5cbe8a80f0eb7 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 12 Nov 2026 23:52:25 -0800 Subject: [PATCH 8/8] fix(server): a merge command names only its first pull request Every host's merge and close takes one change request, so a later number or URL, as in `gh pr merge 19 --body https://github.com/o/r/pull/20`, is option text and no longer counts as a pull request the command closed. Co-Authored-By: Claude Opus 5.5 --- .../src/sourceControl/sourceControlDefinitions.test.ts | 1 + .../src/sourceControl/sourceControlDefinitions.ts | 10 ++++++---- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.test.ts b/apps/server/src/sourceControl/sourceControlDefinitions.test.ts index 85866fd71a792..457c3d5dee7ae 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.test.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.test.ts @@ -31,6 +31,7 @@ it.each([ ['gh pr merge 19 --squash --subject "Release 20"', [19]], [`/bin/zsh -lc "gh pr merge 19 --subject 'Release 20'"`, [19]], ["gh pr merge 19 --body 'see https://github.com/o/r/pull/20'", [19]], + ["gh pr merge 19 --body https://github.com/o/r/pull/20", [19]], ["gh pr merge 19 # then 20", [19]], ["gh pr merge --squash --match-head-commit 3b2f958 #19", [19]], ["gh pr close https://github.com/o/r/pull/19 && gh pr merge 20", [19, 20]], diff --git a/apps/server/src/sourceControl/sourceControlDefinitions.ts b/apps/server/src/sourceControl/sourceControlDefinitions.ts index ee14a16b6384b..670047ccae618 100644 --- a/apps/server/src/sourceControl/sourceControlDefinitions.ts +++ b/apps/server/src/sourceControl/sourceControlDefinitions.ts @@ -43,15 +43,17 @@ export function changeRequestsClosedBy(command: string): ReadonlySet | " for (const match of command.matchAll(new RegExp(close.source, `${close.flags}g`))) { // From the match's start, since Azure's match runs through `--id` to `--status`. const args = command.slice(match.index).split(/[;&|\n]/u, 1)[0] ?? ""; - const numbers: Array = []; + // Every host's merge and close takes one change request, so any later one is option text. + let target: string | undefined; for (const [, double, single, bare] of args.matchAll(SHELL_WORD)) { // A comment ends the command. if (bare !== undefined && /^#(?!\d+$)/u.test(bare)) break; const [, number, url] = NAMED_CHANGE_REQUEST.exec(double ?? single ?? bare ?? "") ?? []; - if (number !== undefined || url !== undefined) numbers.push(Number(number ?? url)); + target = number ?? url; + if (target !== undefined) break; } - if (numbers.length === 0) return "any"; - for (const number of numbers) named.add(number); + if (target === undefined) return "any"; + named.add(Number(target)); } } return named.size === 0 ? "any" : named;