Add PR monitoring: server-driven babysit mode for agent threads

apps/server/src/mcp/McpHttpServer.ts

@@ -22,6 +22,8 @@ import {
   PreviewSnapshotToolkit,
   PreviewStandardToolkit,
 } from "./toolkits/preview/tools.ts";
+import { MonitorToolkitHandlersLive } from "./toolkits/monitor/handlers.ts";
+import { MonitorToolkit } from "./toolkits/monitor/tools.ts";
 
 const unauthorized = HttpServerResponse.jsonUnsafe(
   {
@@ -208,10 +210,17 @@ export const PreviewToolkitRegistrationLive = Layer.mergeAll(
   PreviewSnapshotRegistrationLive,
 );
 
+export const MonitorToolkitRegistrationLive = McpServer.toolkit(MonitorToolkit).pipe(
+  Layer.provide(MonitorToolkitHandlersLive),
+);
+
 const McpTransportLive = McpServer.layerHttp({
   name: "T3 Code",
   version: packageJson.version,
   path: "/mcp",
 }).pipe(Layer.provide(McpAuthMiddlewareLive));
 
-export const layer = PreviewToolkitRegistrationLive.pipe(Layer.provideMerge(McpTransportLive));
+export const layer = Layer.mergeAll(
+  PreviewToolkitRegistrationLive,
+  MonitorToolkitRegistrationLive,
+).pipe(Layer.provideMerge(McpTransportLive));

apps/server/src/mcp/McpInvocationContext.ts

@@ -7,7 +7,7 @@ import {
 import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
 
-export type McpCapability = "preview";
+export type McpCapability = "preview" | "monitor";
 
 export interface McpInvocationScope {
   readonly environmentId: EnvironmentId;
@@ -25,7 +25,7 @@ export class McpInvocationContext extends Context.Service<
 >()("t3/mcp/McpInvocationContext") {}
 
 export const requireMcpCapability = Effect.fn("mcp.requireCapability")(function* (
-  capability: McpCapability,
+  capability: "preview",
 ) {
   const invocation = yield* McpInvocationContext;
   if (!invocation.capabilities.has(capability)) {

apps/server/src/mcp/McpSessionRegistry.ts

@@ -114,7 +114,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* (
         threadId: ThreadId.make(request.threadId),
         providerSessionId,
         providerInstanceId: ProviderInstanceId.make(request.providerInstanceId),
-        capabilities: new Set(["preview"]),
+        capabilities: new Set(["preview", "monitor"]),
         issuedAt,
         expiresAt,
       };

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

@@ -0,0 +1,120 @@
+import { CommandId } from "@t3tools/contracts";
+import * as Crypto from "effect/Crypto";
+import * as DateTime from "effect/DateTime";
+import * as Effect from "effect/Effect";
+import * as Option from "effect/Option";
+
+import * as McpInvocationContext from "../../McpInvocationContext.ts";
+import * as MonitorRegistry from "../../../monitor/MonitorRegistry.ts";
+import { cursorFromSnapshot } from "../../../monitor/monitorDiff.ts";
+import { OrchestrationEngineService } from "../../../orchestration/Services/OrchestrationEngine.ts";
+import { ProjectionSnapshotQuery } from "../../../orchestration/Services/ProjectionSnapshotQuery.ts";
+import { getPullRequestMonitorSnapshot } from "../../../sourceControl/gitHubPullRequestMonitor.ts";
+import { MonitorStartError, MonitorToolkit } from "./tools.ts";
+
+const monitorStart = Effect.fn("MonitorToolkit.monitorStart")(function* ({
+  prNumber,
+}: {
+  readonly prNumber: number;
+}) {
+  const invocation = yield* McpInvocationContext.McpInvocationContext;
+  if (!invocation.capabilities.has("monitor")) {
+    return yield* new MonitorStartError({ message: "PR monitoring is not available." });
+  }
+  const registry = yield* MonitorRegistry.MonitorRegistry;
+  const existing = yield* registry.get(invocation.threadId);
+  if (Option.isSome(existing)) {
+    if (existing.value.prNumber !== prNumber) {
+      return yield* new MonitorStartError({
+        message: `This thread is already monitoring PR #${existing.value.prNumber}. Ask the user before switching to a different pull request.`,
+      });
+    }
+    return {
+      prNumber,
+      status: "monitoring" as const,
+      warning: null,
+      message: `PR #${prNumber} is already being monitored.`,
+    };
+  }
+
+  const snapshots = yield* ProjectionSnapshotQuery;
+  const readModel = yield* snapshots.getCommandReadModel();
+  const thread = readModel.threads.find((candidate) => candidate.id === invocation.threadId);
+  const project = thread
+    ? readModel.projects.find((candidate) => candidate.id === thread.projectId)
+    : undefined;
+  if (!thread || !project) {
+    return yield* new MonitorStartError({ message: "The invoking thread no longer exists." });
+  }
+  const repoCwd = thread.worktreePath ?? project.workspaceRoot;
+  const snapshot = yield* getPullRequestMonitorSnapshot({
+    cwd: repoCwd,
+    pullRequestNumber: prNumber,
+  }).pipe(Effect.mapError((error) => new MonitorStartError({ message: error.message })));
+  if (snapshot.state !== "open") {
+    return yield* new MonitorStartError({
+      message: `PR #${prNumber} is ${snapshot.state}; only open pull requests can be monitored.`,
+    });
+  }
+
+  const createdAt = DateTime.formatIso(yield* DateTime.now);
+  const crypto = yield* Crypto.Crypto;
+  const commandId = CommandId.make(yield* crypto.randomUUIDv4);
+  const warning = snapshot.draft
+    ? `PR #${prNumber} is a draft. Monitoring started, but review bots may not run until it is marked ready for review.`
+    : null;
+  const generation = yield* registry.nextGeneration;
+  const won = yield* registry.registerIfAbsent({
+    threadId: invocation.threadId,
+    prNumber,
+    generation,
+    startedAt: createdAt,
+    cursor: cursorFromSnapshot(snapshot),
+    wakeCount: 0,
+    repoCwd,
+  });
+  if (!won) {
+    // A concurrent monitor_start beat us between the existence check and the
+    // registration write. Surface it rather than clobbering the winner.
+    return yield* new MonitorStartError({
+      message: "A monitor was just started for this thread; check its status before retrying.",
+    });
+  }
+  yield* OrchestrationEngineService.pipe(
+    Effect.flatMap((engine) =>
+      engine.dispatch({
+        type: "thread.monitor.start",
+        commandId,
+        threadId: invocation.threadId,
+        prNumber,
+        blockersSummary: warning ?? "",
+        headSha: snapshot.headSha,
+        createdAt,
+      }),
+    ),
+    Effect.mapError((error) => new MonitorStartError({ message: String(error) })),
+    // Roll back only the registration this call installed — a raced winner's
+    // registration must survive our failure.
+    Effect.onError(() => registry.removeGeneration(invocation.threadId, generation)),
+  );
+  return {
+    prNumber,
+    status: "monitoring" as const,
+    warning,
+    message: warning ?? `Started monitoring PR #${prNumber}.`,
+  };
+});
+
+export const MonitorToolkitHandlersLive = MonitorToolkit.toLayer({
+  monitor_start: (input) =>
+    monitorStart(input).pipe(
+      Effect.catch((error) =>
+        typeof error === "object" &&
+        error !== null &&
+        "_tag" in error &&
+        error._tag === "MonitorStartError"
+          ? error
+          : new MonitorStartError({ message: String(error) }),
+      ),
+    ),
+});

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

@@ -0,0 +1,41 @@
+import * as Schema from "effect/Schema";
+import * as Crypto from "effect/Crypto";
+import { Tool, Toolkit } from "effect/unstable/ai";
+
+import * as McpInvocationContext from "../../McpInvocationContext.ts";
+import * as MonitorRegistry from "../../../monitor/MonitorRegistry.ts";
+import { OrchestrationEngineService } from "../../../orchestration/Services/OrchestrationEngine.ts";
+import { ProjectionSnapshotQuery } from "../../../orchestration/Services/ProjectionSnapshotQuery.ts";
+import * as GitHubCli from "../../../sourceControl/GitHubCli.ts";
+
+export class MonitorStartError extends Schema.TaggedErrorClass<MonitorStartError>()(
+  "MonitorStartError",
+  { message: Schema.String },
+) {}
+
+export const MonitorStartTool = Tool.make("monitor_start", {
+  description:
+    "Start monitoring a pull request for this thread. Call this after creating or identifying the PR when the user has asked you to monitor, babysit, or watch it for review feedback and CI results. The server will watch the PR and wake this thread when there is something to address. Never create the PR as a draft — review bots do not run on drafts.",
+  parameters: Schema.Struct({ prNumber: Schema.Int.check(Schema.isGreaterThan(0)) }),
+  success: Schema.Struct({
+    prNumber: Schema.Number,
+    status: Schema.Literal("monitoring"),
+    warning: Schema.NullOr(Schema.String),
+    message: Schema.String,
+  }),
+  failure: MonitorStartError,
+  dependencies: [
+    McpInvocationContext.McpInvocationContext,
+    MonitorRegistry.MonitorRegistry,
+    OrchestrationEngineService,
+    ProjectionSnapshotQuery,
+    GitHubCli.GitHubCli,
+    Crypto.Crypto,
+  ],
+})
+  .annotate(Tool.Title, "Start PR monitoring")
+  .annotate(Tool.Idempotent, true)
+  .annotate(Tool.Destructive, false)
+  .annotate(Tool.OpenWorld, true);
+
+export const MonitorToolkit = Toolkit.make(MonitorStartTool);

apps/server/src/monitor/MonitorRegistry.ts

@@ -0,0 +1,175 @@
+import { CommandId, type ThreadId } from "@t3tools/contracts";
+import * as Context from "effect/Context";
+import * as DateTime from "effect/DateTime";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
+
+import type { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts";
+import type { PullRequestMonitorCursor } from "./monitorDiff.ts";
+
+export interface MonitorRegistration {
+  readonly threadId: ThreadId;
+  readonly prNumber: number;
+  readonly generation: number;
+  readonly startedAt: string;
+  readonly cursor: PullRequestMonitorCursor;
+  readonly wakeCount: number;
+  readonly repoCwd: string;
+}
+
+export interface MonitorRegistryShape {
+  /** Registers only when the thread has no active registration; returns
+      whether this call won. Losing concurrent monitor_start calls must not
+      clobber the winner's cursor/generation. */
+  readonly registerIfAbsent: (registration: MonitorRegistration) => Effect.Effect<boolean>;
+  /** Removes only the exact generation the caller installed, so a failed
+      start can roll back without erasing a racing winner's registration. */
+  readonly removeGeneration: (threadId: ThreadId, generation: number) => Effect.Effect<void>;
+  readonly get: (threadId: ThreadId) => Effect.Effect<Option.Option<MonitorRegistration>>;
+  readonly updateCursor: (
+    threadId: ThreadId,
+    cursor: PullRequestMonitorCursor,
+    expectedGeneration?: number,
+  ) => Effect.Effect<void>;
+  readonly incrementWake: (
+    threadId: ThreadId,
+    expectedGeneration?: number,
+  ) => Effect.Effect<number>;
+  readonly setWakeCount: (
+    threadId: ThreadId,
+    wakeCount: number,
+    expectedGeneration?: number,
+  ) => Effect.Effect<void>;
+  readonly remove: (
+    threadId: ThreadId,
+    expectedGeneration?: number,
+  ) => Effect.Effect<Option.Option<MonitorRegistration>>;
+  readonly listActive: Effect.Effect<ReadonlyArray<MonitorRegistration>>;
+  readonly nextGeneration: Effect.Effect<number>;
+}
+
+export class MonitorRegistry extends Context.Service<MonitorRegistry, MonitorRegistryShape>()(
+  "t3/monitor/MonitorRegistry",
+) {}
+
+// Module-global store, mirroring McpProviderSession: the registry is
+// constructed both in the runtime core (for the poller) and inside the MCP
+// routes layer (for monitor_start), and those two layer instances must see
+// the same registrations. All operations are synchronous, so plain-Map
+// mutation inside Effect.sync is atomic per call.
+const registrations = new Map<ThreadId, MonitorRegistration>();
+let generation = 0;
+
+const make: MonitorRegistryShape = {
+  registerIfAbsent: (registration) =>
+    Effect.sync(() => {
+      if (registrations.has(registration.threadId)) return false;
+      registrations.set(registration.threadId, registration);
+      return true;
+    }),
+  removeGeneration: (threadId, generation) =>
+    Effect.sync(() => {
+      const current = registrations.get(threadId);
+      if (current !== undefined && current.generation === generation) {
+        registrations.delete(threadId);
+      }
+    }),
+  get: (threadId) => Effect.sync(() => Option.fromNullishOr(registrations.get(threadId))),
+  updateCursor: (threadId, cursor, expectedGeneration) =>
+    Effect.sync(() => {
+      const current = registrations.get(threadId);
+      if (
+        current !== undefined &&
+        (expectedGeneration === undefined || current.generation === expectedGeneration)
+      ) {
+        registrations.set(threadId, { ...current, cursor });
+      }
+    }),
+  incrementWake: (threadId, expectedGeneration) =>
+    Effect.sync(() => {
+      const current = registrations.get(threadId);
+      if (
+        current === undefined ||
+        (expectedGeneration !== undefined && current.generation !== expectedGeneration)
+      ) {
+        return current?.wakeCount ?? 0;
+      }
+      const wakeCount = (current?.wakeCount ?? 0) + 1;
+      registrations.set(threadId, { ...current, wakeCount });
+      return wakeCount;
+    }),
+  setWakeCount: (threadId, wakeCount, expectedGeneration) =>
+    Effect.sync(() => {
+      const current = registrations.get(threadId);
+      if (
+        current !== undefined &&
+        (expectedGeneration === undefined || current.generation === expectedGeneration)
+      ) {
+        registrations.set(threadId, { ...current, wakeCount });
+      }
+    }),
+  remove: (threadId, expectedGeneration) =>
+    Effect.sync(() => {
+      const current = registrations.get(threadId);
+      if (
+        current === undefined ||
+        (expectedGeneration !== undefined && current.generation !== expectedGeneration)
+      ) {
+        return Option.none();
+      }
+      registrations.delete(threadId);
+      return Option.fromNullishOr(current);
+    }),
+  listActive: Effect.sync(() => [...registrations.values()]),
+  nextGeneration: Effect.sync(() => ++generation),
+};
+
+// The engine used for teardown dispatch is bound by PullRequestMonitor when
+// it starts (it is the only construction site that has the engine and runs in
+// the runtime core). Until then teardown leaves registrations intact because
+// it cannot durably project the corresponding monitor end.
+let activeEngine: OrchestrationEngineService["Service"] | undefined;
+
+export const bindEngine = (engine: OrchestrationEngineService["Service"]): void => {
+  activeEngine = engine;
+};
+
+export const layer = Layer.effect(
+  MonitorRegistry,
+  Effect.acquireRelease(Effect.succeed(MonitorRegistry.of(make)), () =>
+    Effect.gen(function* () {
+      for (const registration of [...registrations.values()]) {
+        yield* endActiveMonitorForSession(registration.threadId);
+      }
+    }),
+  ),
+);
+
+export const endActiveMonitorForSession = (threadId: ThreadId): Effect.Effect<void> =>
+  Effect.gen(function* () {
+    const registration = yield* make.get(threadId);
+    if (Option.isNone(registration) || activeEngine === undefined) return;
+    const endedAt = DateTime.formatIso(yield* DateTime.now);
+    const commandId = CommandId.make(`monitor-session-ended:${threadId}:${endedAt}`);
+    yield* activeEngine.dispatch({
+      type: "thread.monitor.end",
+      commandId,
+      threadId,
+      reason: "session-ended",
+      blockersSummary: "",
+      endedAt,
+    });
+    yield* make.remove(threadId, registration.value.generation);
+  }).pipe(
+    Effect.catch((error) => Effect.logWarning("monitor teardown failed", { threadId, error })),
+  );
+
+export const __testing = {
+  make,
+  reset: (): void => {
+    registrations.clear();
+    generation = 0;
+    activeEngine = undefined;
+  },
+};