fix(server): optionally watch reviews from a shared account

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* (

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,

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

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 })),

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);

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,

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}`),

apps/server/src/provider/agentContextBudget.json

@@ -1,5 +1,5 @@
 {
   "claudeSystemAppendBytes": 5930,
   "codexAdditionalContextBytes": 7324,
-  "mcpToolListJsonBytes": 49806
+  "mcpToolListJsonBytes": 49845
 }