feat(server): opt in to compaction before PR waits

apps/server/src/mcp/toolkits/pullRequests/handlers.test.ts

@@ -253,6 +253,19 @@ describe("pull request toolkit handlers", () => {
     }),
   );
 
+  it.effect("passes explicit compaction opt-in to the watch service", () =>
+    Effect.gen(function* () {
+      const harness = yield* makeHarness();
+      yield* harness.call("watch_pull_request", {
+        url: "https://github.com/t3tools/t3code/pull/9",
+        compactBeforeWaiting: true,
+      });
+      expect(yield* Ref.get(harness.commands)).toMatchObject([
+        { type: "thread.pull-request.watch", compactBeforeWaiting: true },
+      ]);
+    }),
+  );
+
   it.effect("refuses to watch a merged pull request and stops an existing watch", () =>
     Effect.gen(function* () {
       const watch = {

apps/server/src/mcp/toolkits/pullRequests/handlers.ts

@@ -37,6 +37,7 @@ import {
   PullRequestListFailedError,
   PullRequestNotOpenError,
   type PullRequestTargetInput,
+  type WatchPullRequestInput,
   PullRequestWatchFailedError,
   PullRequestWatchFromSubagentError,
   PullRequestThreadNotFoundError,
@@ -222,7 +223,7 @@ const make = Effect.gen(function* () {
    * result reports the state the thread holds afterwards.
    */
   const setWatching = Effect.fn("PullRequestsToolkit.setWatching")(function* (
-    input: PullRequestTargetInput,
+    input: WatchPullRequestInput,
     watching: boolean,
   ) {
     const thread = yield* requireThread(PullRequestWatchFailedError, input.threadId);
@@ -250,6 +251,9 @@ const make = Effect.gen(function* () {
         repository: target.repository,
         number: target.number,
         watching,
+        ...(input.compactBeforeWaiting === undefined
+          ? {}
+          : { compactBeforeWaiting: input.compactBeforeWaiting }),
         ...(watching ? { link: { url: target.url, source: "agent" as const } } : {}),
       })
       .pipe(Effect.catchCause(dispatchFailure(PullRequestWatchFailedError)));

apps/server/src/mcp/toolkits/pullRequests/tools.ts

@@ -291,10 +291,21 @@ const ListThreadPullRequestsTool = Tool.make("list_thread_pull_requests", {
   .annotate(Tool.Idempotent, true)
   .annotate(Tool.OpenWorld, false);
 
+export const WatchPullRequestInput = Schema.Struct({
+  ...PullRequestTargetInput.fields,
+  compactBeforeWaiting: Schema.optional(
+    Schema.Boolean.annotate({
+      description:
+        "Opt in to one native compaction turn before a new watch waits. Only with explicit user authorization.",
+    }),
+  ),
+});
+export type WatchPullRequestInput = typeof WatchPullRequestInput.Type;
+
 const WatchPullRequestTool = Tool.make("watch_pull_request", {
   description:
-    "Have T3 Code watch an open pull request for this thread, linking it first if needed. T3 Code checks it every two minutes and wakes you with a message when a check fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict with its base. Use this to monitor or babysit a pull request instead of polling, sleeping, or running a watcher. Only comments posted after this call wake you, so handle the existing ones first, then end your turn. A wake is news, not a merge decision: check readiness yourself before merging. While T3 Code watches, the thread stays in the user's Working list, not their inbox. When you hand the work back to the user, call unwatch_pull_request first so the thread returns to their inbox. Watching ends when the pull request merges or closes, when its thread settles or is archived, when T3 Code fails to read it 8 times in a row (a host rate limit only delays it), when the user stops this thread, or when you call unwatch_pull_request. Unsettle the thread before starting a new watch. A subagent cannot watch: its parent thread owns the pull request.",
-  parameters: PullRequestTargetInput,
+    "Have T3 Code watch an open pull request for this thread, linking it first if needed. T3 Code checks it every two minutes and wakes you with a message when a check fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict with its base. Use this to monitor or babysit a pull request instead of polling, sleeping, or running a watcher. Only comments posted after this call wake you, so handle the existing ones first, then end your turn. A wake is news, not a merge decision: check readiness yourself before merging. While T3 Code watches, the thread stays in the user's Working list, not their inbox. When you hand the work back to the user, call unwatch_pull_request first so the thread returns to their inbox. Watching ends when the pull request merges or closes, when its thread settles or is archived, when T3 Code fails to read it 8 times in a row (a host rate limit only delays it), when the user stops this thread, or when you call unwatch_pull_request. Unsettle the thread before starting a new watch. A subagent cannot watch: its parent thread owns the pull request. Only when the user explicitly asks to compact before a long wait, set compactBeforeWaiting to true: starting a new watch then queues one /compact turn after your current turn, so end it with a clear account of the user's intent and pending work. If the call is rejected because the thread has queued work or pending input, watch without it.",
+  parameters: WatchPullRequestInput,
   success: WatchPullRequestResult,
   failure: PullRequestToolError,
   dependencies,

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -379,6 +379,9 @@ function wakeWorkStartedAt(
   return previous === undefined ? {} : { workStartedAt: orchestrationV2RunWorkStartedAt(previous) };
 }
 
+/** Message ids of the compaction turn a pull request watch queued before waiting. */
+const PULL_REQUEST_WATCH_COMPACTION_PREFIX = "pr-watch-compact:";
+
 /** A native /compact or /logout turn: provider maintenance, not agent work. */
 export function isNativeMaintenanceCommand(message: {
   readonly text: string;
@@ -1241,6 +1244,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     });
 
   const startNextQueuedRun = (threadId: ThreadId, options?: { readonly failedRunId?: RunId }) =>
+    startQueuedRunOnce(threadId, options).pipe(
+      Effect.flatMap((result) =>
+        result === "skipped" ? startQueuedRunOnce(threadId, options) : Effect.void,
+      ),
+    );
+
+  const startQueuedRunOnce = (threadId: ThreadId, options?: { readonly failedRunId?: RunId }) =>
     Effect.gen(function* () {
       // Every terminal run checks the queue. Only a deliverable queued run
       // needs the transcript for provider handoff and legacy import context.
@@ -1344,6 +1354,43 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         });
       }
 
+      // A question or approval that arrived after the watch queued its compaction drops the
+      // compaction, so the answer runs next instead of waiting behind a possibly cold summary.
+      if (
+        queuedMessage.id.startsWith(PULL_REQUEST_WATCH_COMPACTION_PREFIX) &&
+        // The user may have edited the queued compaction into real work.
+        queuedMessage.attachments.length === 0 &&
+        queuedMessage.text.trim().toLowerCase() === "/compact" &&
+        projection.runtimeRequests.some((request) => request.status === "pending")
+      ) {
+        const now = yield* DateTime.now;
+        const base = {
+          threadId,
+          runId: queuedRun.id,
+          nodeId: rootNode.id,
+          providerInstanceId: queuedRun.providerInstanceId,
+          occurredAt: now,
+        };
+        yield* writeSystemEvents([
+          {
+            ...base,
+            type: "run.updated",
+            payload: { ...queuedRun, status: "cancelled", queuePosition: null, completedAt: now },
+          },
+          {
+            ...base,
+            type: "run-attempt.updated",
+            payload: { ...attempt, status: "cancelled", completedAt: now },
+          },
+          {
+            ...base,
+            type: "node.updated",
+            payload: { ...rootNode, status: "cancelled", completedAt: now },
+          },
+        ]);
+        return "skipped" as const;
+      }
+
       const commandId = CommandId.make(`command:system:start-queued:${queuedRun.id}`);
       const now = yield* DateTime.now;
       const selectionChanged = !modelSelectionsEqual(
@@ -2284,6 +2331,67 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     });
   });
 
+  const dispatchPullRequestWatch = Effect.fn("orchestrationV2.dispatch.pullRequestWatch")(
+    function* (
+      command: Extract<OrchestrationV2Command, { readonly type: "thread.pull-request.watch" }>,
+      events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
+      effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
+    ) {
+      if (!command.watching || command.compactBeforeWaiting !== true) {
+        return yield* dispatchThreadMutation(command, events, effects);
+      }
+      const projection = yield* loadProjectionForCommand(command, [
+        "runs",
+        "runtimeRequests",
+        "messages",
+      ]);
+      // Repeating a watch call must not buy another summary. Stopping and starting again is
+      // a new opt-in; the queued maintenance turn remains visible and cancellable normally.
+      const existing = threadPullRequestsOf(projection.thread).find((link) =>
+        threadPullRequestKeysEqual(link, command),
+      );
+      if (existing?.watch !== undefined) {
+        return yield* dispatchThreadMutation(command, events, effects);
+      }
+      const active = projection.runs.find(isBlockingRun);
+      const latest = latestExecutedRun(projection.runs);
+      if (
+        projection.runtimeRequests.some((request) => request.status === "pending") ||
+        projection.runs.some((run) => run.status === "queued") ||
+        (active !== undefined && active.status !== "running") ||
+        (active === undefined && latest?.status !== "completed") ||
+        projection.messages.some(
+          (message) => message.id === active?.userMessageId && isNativeMaintenanceCommand(message),
+        )
+      ) {
+        return yield* new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause:
+            "Compacting before waiting requires a conversation with no pending input, approval, queued work, or unfinished maintenance. Watch without compaction instead.",
+        });
+      }
+      // Commit the watch and the bounded maintenance request together. The ordinary queue
+      // runs it only after this turn ends, and Stop/archiving retain their normal semantics.
+      yield* dispatchThreadMutation(command, events, effects);
+      yield* dispatchMessage(
+        {
+          type: "message.dispatch",
+          commandId: command.commandId,
+          threadId: command.threadId,
+          messageId: MessageId.make(`${PULL_REQUEST_WATCH_COMPACTION_PREFIX}${command.commandId}`),
+          text: "/compact",
+          attachments: [],
+          dispatchMode: { type: "queue_after_active" },
+          createdBy: "agent",
+          creationSource: "server",
+        },
+        events,
+        effects,
+      );
+    },
+  );
+
   // Checked under the thread lock: the watch or the thread can change while the host is read.
   // The watch is recorded first so the wake's own thread events carry it.
   const dispatchPullRequestWatchSync = Effect.fn("orchestrationV2.dispatch.pullRequestWatchSync")(
@@ -10208,7 +10316,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       case "thread.pull-request.link":
       case "thread.pull-request.unlink":
       case "thread.pull-request-link.sync":
-      case "thread.pull-request.watch":
       case "thread.pull-request.sync":
       case "thread.title.regeneration.complete":
       case "thread.runtime-mode.set":
@@ -10217,6 +10324,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       case "provider.switch":
         yield* dispatchThreadMutation(command, events, effects);
         break;
+      case "thread.pull-request.watch":
+        yield* dispatchPullRequestWatch(command, events, effects);
+        break;
       case "thread.pull-request-watch.sync":
         yield* dispatchPullRequestWatchSync(command, events, effects);
         break;