diff --git a/apps/server/src/mcp/toolkits/pullRequests/handlers.ts b/apps/server/src/mcp/toolkits/pullRequests/handlers.ts index 84e65ce877f69..b96f1ae6d973d 100644 --- a/apps/server/src/mcp/toolkits/pullRequests/handlers.ts +++ b/apps/server/src/mcp/toolkits/pullRequests/handlers.ts @@ -199,18 +199,27 @@ const make = Effect.gen(function* () { // The domain service owns the registration span. const setWatching = Effect.fnUntraced(function* ( - input: PullRequestTargetInput, + input: PullRequestTargetInput & { readonly includeOwnRemarks?: boolean | undefined }, watching: boolean, ) { const scope = yield* McpInvocationContext.requireMcpCapability("pull-requests"); const threadId = input.threadId ?? scope.thread?.threadId; if (threadId === undefined) return yield* new PullRequestThreadRequiredError(); - return yield* watches.setWatching({ threadId, target: input, watching }).pipe( - Effect.catchTags({ - PullRequestWatchThreadNotFoundError: (error) => - threadNotFound(engine, error.threadId, error.message), - }), - ); + return yield* watches + .setWatching({ + threadId, + target: input, + watching, + ...(input.includeOwnRemarks === undefined + ? {} + : { includeOwnRemarks: input.includeOwnRemarks }), + }) + .pipe( + Effect.catchTags({ + PullRequestWatchThreadNotFoundError: (error) => + threadNotFound(engine, error.threadId, error.message), + }), + ); }); const locateAll = Effect.fn("PullRequestsToolkit.locateAll")(function* ( diff --git a/apps/server/src/mcp/toolkits/pullRequests/tools.ts b/apps/server/src/mcp/toolkits/pullRequests/tools.ts index 376817db10079..f3cd78416e297 100644 --- a/apps/server/src/mcp/toolkits/pullRequests/tools.ts +++ b/apps/server/src/mcp/toolkits/pullRequests/tools.ts @@ -202,6 +202,7 @@ export type UnlinkPullRequestsResult = typeof UnlinkPullRequestsResult.Type; export const WatchPullRequestResult = Schema.Struct({ ...PullRequestIdentity, + includeOwnRemarks: Schema.Boolean, watching: Schema.Boolean, wasWatching: Schema.Boolean, }); @@ -327,7 +328,10 @@ const ListThreadPullRequestsTool = Tool.make("list_thread_pull_requests", { const WatchPullRequestTool = Tool.make("watch_pull_request", { description: "Watch an open PR for this thread, linking it first if needed. T3 wakes you on failed checks, passed required checks, new comments or reviews, or conflicts. Handle old comments first. Only later ones wake you. After this call, end your turn. The watch stays on after each wake, so do not call this again on a wake. A wake is not a merge decision. Check readiness yourself.", - parameters: PullRequestTargetInput, + parameters: Schema.Struct({ + ...PullRequestTargetInput.fields, + includeOwnRemarks: Schema.optional(Schema.Boolean), + }), success: WatchPullRequestResult, failure: PullRequestToolError, dependencies: watchDependencies, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c732e0ca67ccf..d823e63cd618c 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3851,7 +3851,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio pullRequestSnapshotEndsWatch(link?.snapshot ?? null, entry.startedAt), ); if ( - entries.some(({ entry, link }) => link?.watch?.startedAt !== entry.startedAt) || + entries.some( + ({ entry, link }) => + link?.watch?.startedAt !== entry.startedAt || + // The revision changes without invalidating wakes already queued for this watch. + (entry.remarksPolicyRevision ?? 0) !== (link.watch.remarksPolicyRevision ?? 0), + ) || (wake && inactive) || // One wake cannot drop the news of just some of its pull requests; the sender splits it. (entries.length > 1 && outdated.some(Boolean)) @@ -4902,7 +4907,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : linked; const existing = links.find(visible); if (existing === undefined) return thread; - const watch = + let watch = command.type === "thread.pull-request-watch.sync" ? existing.watch?.startedAt === command.startedAt ? syncedPullRequestWatch(existing.watch, command) @@ -4911,6 +4916,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? undefined : (existing.watch ?? { startedAt, + ...(command.includeOwnRemarks && { includeOwnRemarks: true }), headSha: null, failedChecks: [], passed: false, @@ -4945,6 +4951,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio remarkIds: [], wakes: 0, }); + if ( + command.type === "thread.pull-request.watch" && + watch !== undefined && + watch === existing.watch && + command.includeOwnRemarks !== undefined && + command.includeOwnRemarks !== (watch.includeOwnRemarks ?? false) + ) { + watch = { + ...watch, + includeOwnRemarks: command.includeOwnRemarks, + remarksPolicyRevision: (watch.remarksPolicyRevision ?? 0) + 1, + remarksThrough: DateTime.formatIso(now), + remarkIds: [], + wakes: 0, + }; + } if (watch === existing.watch && links === linked) return thread; const final = command.type === "thread.pull-request-watch.sync" && watch === undefined diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index ddfe74160459a..4d41c0157c92f 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -115,7 +115,8 @@ interface HeldWake { readonly reason?: WatchEndReason; } -const watchKey = ({ thread, watch }: WatchTarget) => `${thread.id} ${watch.startedAt}`; +const watchKey = ({ thread, watch }: WatchTarget) => + `${thread.id} ${watch.startedAt} ${watch.remarksPolicyRevision ?? 0}`; type WatchEndReason = | "merged" @@ -195,6 +196,11 @@ function planRead( const checksPendingBeforeOf = ({ watch }: WatchTarget) => watch.checksPending === undefined ? {} : { checksPendingBefore: watch.checksPending }; +const remarksPolicyRevisionOf = ({ watch }: WatchTarget) => + watch.remarksPolicyRevision === undefined + ? {} + : { remarksPolicyRevision: watch.remarksPolicyRevision }; + function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatch): boolean { return ( left.startedAt === right.startedAt && @@ -321,6 +327,7 @@ export const make = Effect.gen(function* () { ...identityOf(target.link), startedAt: target.watch.startedAt, watch: next, + ...remarksPolicyRevisionOf(target), ...checksPendingBeforeOf(target), ...(endedWatch === undefined ? {} : { endedWatch }), ...(wake === undefined @@ -463,12 +470,14 @@ export const make = Effect.gen(function* () { ...identityOf(first.target.link), startedAt: first.target.watch.startedAt, watch: first.next, + ...remarksPolicyRevisionOf(first.target), ...checksPendingBeforeOf(first.target), ...(first.endedWatch === undefined ? {} : { endedWatch: first.endedWatch }), others: rest.map((item) => ({ ...identityOf(item.target.link), startedAt: item.target.watch.startedAt, watch: item.next, + ...remarksPolicyRevisionOf(item.target), ...checksPendingBeforeOf(item.target), ...(item.endedWatch === undefined ? {} : { endedWatch: item.endedWatch }), ...(item.wake.includesRemarks === true ? { includesRemarks: true } : {}), @@ -596,8 +605,16 @@ export const make = Effect.gen(function* () { threadTails.delete(group.key); } const reference = referenceOf(group); - // Invalidate cached reads before using the newer fingerprint; clients receive no notification. - if (fingerprint !== null) yield* pullRequests.invalidate({ reference }); + const policyChanged = + last !== undefined && + group.targets.some( + (target) => + (target.watch.remarksPolicyRevision ?? 0) > 0 && !last.watches.has(watchKey(target)), + ); + // Refresh host reads for a changed policy or a fingerprint-driven read. + if (policyChanged || fingerprint !== null) { + yield* pullRequests.invalidate({ reference }); + } const [detailRead, activityRead] = yield* Effect.all( [ Effect.exit(pullRequests.detail({ ...reference, allowStale: false })), diff --git a/apps/server/src/orchestration-v2/pullRequestWatch.test.ts b/apps/server/src/orchestration-v2/pullRequestWatch.test.ts index eb2ae2756d540..044b69f350f58 100644 --- a/apps/server/src/orchestration-v2/pullRequestWatch.test.ts +++ b/apps/server/src/orchestration-v2/pullRequestWatch.test.ts @@ -643,6 +643,22 @@ describe("evaluatePullRequestWatch", () => { assert.deepEqual(evaluatePullRequestWatch(watching, detail(), [own]).changes, []); }); + it("can watch same-account reviews once while retaining the wake limit", () => { + const review = remark("AGENT-USER", "2026-10-02T12:06:00Z", "Please fix the race."); + const shared = { ...watch({ headSha: detail().headSha ?? null }), includeOwnRemarks: true }; + const first = evaluatePullRequestWatch(shared, detail(), [review]); + assert.deepEqual(first.changes, [{ kind: "remarks", remarks: [review] }]); + assert.isTrue(first.next.includeOwnRemarks); + assert.deepEqual(evaluatePullRequestWatch(first.next, detail(), [review]).changes, []); + const last = evaluatePullRequestWatch( + { ...shared, wakes: PULL_REQUEST_WATCH_WAKE_LIMIT - 1 }, + detail(), + [review], + ); + assert.isTrue(last.exhausted); + assert.deepEqual(evaluatePullRequestWatch(watch(), detail(), [review]).changes, []); + }); + it("does not treat a failed check read as a rerun", () => { const failed = detail({ checks: [check("lint", "failure")] }); const reported = evaluatePullRequestWatch(watch(), failed, noRemarks); diff --git a/apps/server/src/orchestration-v2/pullRequestWatch.ts b/apps/server/src/orchestration-v2/pullRequestWatch.ts index e12d9a124df8a..0d6eeb5c64632 100644 --- a/apps/server/src/orchestration-v2/pullRequestWatch.ts +++ b/apps/server/src/orchestration-v2/pullRequestWatch.ts @@ -184,7 +184,7 @@ export function evaluatePullRequestWatch( const at = Date.parse(activeAt(remark)); return ( (at > through || (at === through && !watch.remarkIds.includes(remark.id))) && - remark.author?.login.toLowerCase() !== own + (watch.includeOwnRemarks === true || remark.author?.login.toLowerCase() !== own) ); }); if (fresh.length > 0) changes.push({ kind: "remarks", remarks: fresh }); @@ -229,6 +229,12 @@ export function evaluatePullRequestWatch( changes, next: { startedAt: watch.startedAt, + ...(watch.remarksPolicyRevision === undefined + ? {} + : { remarksPolicyRevision: watch.remarksPolicyRevision }), + ...(watch.includeOwnRemarks === undefined + ? {} + : { includeOwnRemarks: watch.includeOwnRemarks }), headSha, failedChecks, passed, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index c7cc43ee5ddea..a3362759319c0 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -3578,6 +3578,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { threadId, ...key, watching: true, + includeOwnRemarks: true, link: { url, source: "agent" }, }); assert.equal( @@ -3587,6 +3588,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { const started = yield* watchOf; assert.isDefined(started); if (started === undefined) return; + assert.isTrue(started.includeOwnRemarks); yield* orchestrator.dispatch({ type: "thread.metadata.update", @@ -3609,6 +3611,81 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { assert.isTrue((yield* maintenance.rebuild).valid); assert.deepEqual(yield* watchOf, recorded); + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make("pr-watch-exclude-own"), + threadId, + ...key, + watching: true, + includeOwnRemarks: false, + }); + const changed = yield* watchOf; + assert.isFalse(changed?.includeOwnRemarks); + assert.equal(changed?.startedAt, started.startedAt); + assert.equal(changed?.remarksThrough, DateTime.formatIso(yield* DateTime.now)); + assert.deepEqual(changed?.failedChecks, recorded.failedChecks); + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make("pr-watch-preserve-option"), + threadId, + ...key, + watching: true, + }); + assert.deepEqual(yield* watchOf, changed); + const staleOption = yield* orchestrator + .dispatch({ + type: "thread.pull-request-watch.sync", + commandId: CommandId.make("pr-watch-old-option-sync"), + threadId, + ...key, + startedAt: recorded.startedAt, + watch: recorded, + }) + .pipe(Effect.flip); + assert.equal(staleOption._tag, "OrchestratorDispatchError"); + assert.deepEqual(yield* watchOf, changed); + const staleTerminal = yield* orchestrator + .dispatch({ + type: "thread.pull-request-watch.sync", + commandId: CommandId.make("pr-watch-old-option-terminal"), + threadId, + ...key, + startedAt: recorded.startedAt, + watch: null, + endedWatch: recorded, + }) + .pipe(Effect.result); + const afterTerminal = yield* watchOf; + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make("pr-watch-reinclude-own"), + threadId, + ...key, + watching: true, + includeOwnRemarks: true, + }); + const reset = yield* watchOf; + const staleRoundTrip = yield* orchestrator + .dispatch({ + type: "thread.pull-request-watch.sync", + commandId: CommandId.make("pr-watch-old-option-round-trip"), + threadId, + ...key, + startedAt: recorded.startedAt, + watch: recorded, + }) + .pipe(Effect.result); + assert.deepEqual( + [staleTerminal._tag === "Failure", staleRoundTrip._tag === "Failure"], + [true, true], + ); + assert.deepEqual(afterTerminal, changed); + assert.equal(reset?.startedAt, started.startedAt); + assert.equal(reset?.wakes, 0); + assert.deepEqual(yield* watchOf, reset); + assert.isTrue((yield* maintenance.rebuild).valid); + assert.deepEqual(yield* watchOf, reset); + yield* orchestrator.dispatch({ type: "thread.pull-request.watch", commandId: CommandId.make("pr-watch-stop"), @@ -5047,14 +5124,18 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }); let reads = 0; + let invalidations = 0; const host = Layer.mergeAll( NodeServices.layer, NodeCrypto.layer, Layer.mock(PullRequestService.PullRequestService)({ - // The host reports the same fingerprint throughout, so only the deadline can prompt a read. + // The host reports the same fingerprint throughout. watchFingerprint: () => Effect.succeed(fingerprinted ? { status: "quiet", remarks: "none" } : null), - invalidate: () => Effect.void, + invalidate: () => + Effect.sync(() => { + invalidations++; + }), detail: () => Effect.sync(() => { reads++; @@ -5100,11 +5181,24 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { // The notice clears the deadline, so the watch returns to its quiet cadence. const readsAfterNotice = reads; + const invalidationsAfterNotice = invalidations; yield* TestClock.adjust("2 minutes"); yield* sweep; assert.equal(reads, readsAfterNotice); assert.lengthOf(yield* wakes, 1); + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`pr-watch-due-change-policy-${fingerprinted}`), + threadId, + ...key, + watching: true, + includeOwnRemarks: true, + }); + yield* sweep; + assert.equal(reads, readsAfterNotice + 1); + assert.equal(invalidations, invalidationsAfterNotice + 1); + yield* orchestrator.dispatch({ type: "thread.pull-request.watch", commandId: CommandId.make(`pr-watch-due-end-${fingerprinted}`), diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 0a354c1e7d560..82c2b484669df 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5930, "codexAdditionalContextBytes": 7324, - "mcpToolListJsonBytes": 49806 + "mcpToolListJsonBytes": 49845 } diff --git a/apps/server/src/pullRequest/PullRequestWatchService.ts b/apps/server/src/pullRequest/PullRequestWatchService.ts index 6ed736075c018..c0d1e2c20f719 100644 --- a/apps/server/src/pullRequest/PullRequestWatchService.ts +++ b/apps/server/src/pullRequest/PullRequestWatchService.ts @@ -69,8 +69,13 @@ export class PullRequestWatchService extends Context.Service< readonly threadId: ThreadId; readonly target: PullRequestTarget; readonly watching: boolean; + readonly includeOwnRemarks?: boolean; }) => Effect.Effect< - ResolvedTarget & { readonly watching: boolean; readonly wasWatching: boolean }, + ResolvedTarget & { + readonly watching: boolean; + readonly wasWatching: boolean; + readonly includeOwnRemarks: boolean; + }, | PullRequestWatchFailedError | PullRequestNotOpenError | PullRequestWatchFromSubagentError @@ -104,6 +109,7 @@ const make = Effect.gen(function* () { readonly threadId: ThreadId; readonly target: PullRequestTarget; readonly watching: boolean; + readonly includeOwnRemarks?: boolean; }) { const { watching } = input; const thread = yield* requireThread(input.threadId); @@ -157,6 +163,9 @@ const make = Effect.gen(function* () { repository: target.repository, number: target.number, watching, + ...(watching && input.includeOwnRemarks !== undefined + ? { includeOwnRemarks: input.includeOwnRemarks } + : {}), ...(watching ? { link: { url: target.url, source: "agent" as const } } : {}), }) .pipe( @@ -167,9 +176,11 @@ const make = Effect.gen(function* () { ), ); const after = yield* requireThread(thread.id); + const watch = watchedLink(after)?.watch; return { ...target, - watching: watchedLink(after)?.watch !== undefined, + watching: watch !== undefined, + includeOwnRemarks: watch?.includeOwnRemarks === true, wasWatching: before?.watch !== undefined, }; }); diff --git a/docs/user/source-control.md b/docs/user/source-control.md index 41879e69a24cf..d45ecdb0b52c3 100644 --- a/docs/user/source-control.md +++ b/docs/user/source-control.md @@ -331,7 +331,10 @@ 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 +watching inside GitHub's rate limit. Comments from your own account do not wake it by default. +If multiple people or agents share that account, ask the agent to watch with `includeOwnRemarks: true` +to include their comments and reviews. Changing this option starts a new comment window; earlier +comments are not replayed. The same comment-only wake limit applies. Watching ends when the pull request merges or closes, 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 diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 69338865e55a6..544c9f51988c1 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3247,6 +3247,7 @@ export const OrchestrationV2Command = Schema.Union([ threadId: ThreadId, ...ThreadPullRequestKey.fields, watching: Schema.Boolean, + includeOwnRemarks: Schema.optional(Schema.Boolean), link: Schema.optional( Schema.Struct({ url: TrimmedNonEmptyString, @@ -3588,6 +3589,7 @@ const OrchestrationV2InternalCommand = Schema.Union([ ...ThreadPullRequestKey.fields, startedAt: IsoDateTime, watch: Schema.NullOr(ThreadPullRequestWatch), + remarksPolicyRevision: Schema.optional(NonNegativeInt), /** * The watch's `checksPending` when this read started. A push that clears it during the read * makes the read's answer describe the old head, so the cleared answer stays. @@ -3602,6 +3604,7 @@ const OrchestrationV2InternalCommand = Schema.Union([ ...ThreadPullRequestKey.fields, startedAt: IsoDateTime, watch: Schema.NullOr(ThreadPullRequestWatch), + remarksPolicyRevision: Schema.optional(NonNegativeInt), checksPendingBefore: Schema.optional(Schema.Boolean), endedWatch: Schema.optional(ThreadPullRequestWatch), includesRemarks: Schema.optional(Schema.Boolean), diff --git a/packages/contracts/src/threadPullRequest.ts b/packages/contracts/src/threadPullRequest.ts index 7775f0755ff0b..e9d82e467fff3 100644 --- a/packages/contracts/src/threadPullRequest.ts +++ b/packages/contracts/src/threadPullRequest.ts @@ -81,6 +81,9 @@ export type ThreadPullRequestKey = typeof ThreadPullRequestKey.Type; export const ThreadPullRequestWatch = Schema.Struct({ startedAt: IsoDateTime, + includeOwnRemarks: Schema.optional(Schema.Boolean), + /** Changes when remark filtering is reset without restarting the watch. */ + remarksPolicyRevision: Schema.optional(NonNegativeInt), headSha: Schema.NullOr(TrimmedNonEmptyString), failedChecks: Schema.Array(TrimmedNonEmptyString), passed: Schema.Boolean,