feat: watch a linked pull request and follow up on checks, reviews, and conflicts

apps/server/integration/OrchestrationEngineHarness.integration.ts

@@ -71,6 +71,7 @@ import * as PullRequestSyncReactor from "../src/orchestration/PullRequestSyncRea
 import * as ThreadPullRequestReactor from "../src/orchestration/ThreadPullRequestReactor.ts";
 import { ThreadTabSettlementReactor } from "../src/threadTabs/settlement.ts";
 import { UsageLimitResumeReactor } from "../src/orchestration/UsageLimitResumeReactor.ts";
+import * as PullRequestWatchReactor from "../src/pullRequestWatch/reactor.ts";
 import { OrchestrationReactor } from "../src/orchestration/Services/OrchestrationReactor.ts";
 import { ProjectionSnapshotQuery } from "../src/orchestration/Services/ProjectionSnapshotQuery.ts";
 import {
@@ -430,6 +431,13 @@ export const makeOrchestrationIntegrationHarness = (
           requestSync: () => Effect.void,
         }),
       ),
+      Layer.provideMerge(
+        Layer.succeed(PullRequestWatchReactor.PullRequestWatchReactor, {
+          start: () => Effect.void,
+          drain: Effect.void,
+          evaluate: () => Effect.void,
+        }),
+      ),
       Layer.provideMerge(
         Layer.succeed(AgentAwarenessRelay.AgentAwarenessRelay, {
           publishThread: () => Effect.void,

apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts

@@ -18,6 +18,7 @@ import * as AgentAwarenessRelay from "../../relay/AgentAwarenessRelay.ts";
 import { StorageCleanup } from "../../storageCleanup.ts";
 import { ThreadTabSettlementReactor } from "../../threadTabs/settlement.ts";
 import { UsageLimitResumeReactor } from "../UsageLimitResumeReactor.ts";
+import * as PullRequestWatchReactor from "../../pullRequestWatch/reactor.ts";
 
 describe("OrchestrationReactor", () => {
   let runtime: ManagedRuntime.ManagedRuntime<OrchestrationReactor, never> | null = null;
@@ -124,6 +125,16 @@ describe("OrchestrationReactor", () => {
             requestSync: () => Effect.void,
           }),
         ),
+        Layer.provideMerge(
+          Layer.succeed(PullRequestWatchReactor.PullRequestWatchReactor, {
+            start: () => {
+              started.push("pull-request-watch-reactor");
+              return Effect.void;
+            },
+            drain: Effect.void,
+            evaluate: () => Effect.void,
+          }),
+        ),
         Layer.provideMerge(
           Layer.succeed(AgentAwarenessRelay.AgentAwarenessRelay, {
             publishThread: () => Effect.void,
@@ -151,6 +162,7 @@ describe("OrchestrationReactor", () => {
       "thread-tab-settlement-reactor",
       "usage-limit-resume-reactor",
       "pull-request-sync-reactor",
+      "pull-request-watch-reactor",
       "agent-awareness-relay",
       "storage-cleanup",
     ]);

apps/server/src/orchestration/Layers/OrchestrationReactor.ts

@@ -16,6 +16,7 @@ import * as AgentAwarenessRelay from "../../relay/AgentAwarenessRelay.ts";
 import * as StorageCleanup from "../../storageCleanup.ts";
 import * as ThreadTabSettlementReactor from "../../threadTabs/settlement.ts";
 import * as UsageLimitResumeReactor from "../UsageLimitResumeReactor.ts";
+import * as PullRequestWatchReactor from "../../pullRequestWatch/reactor.ts";
 
 export const makeOrchestrationReactor = Effect.gen(function* () {
   const providerRuntimeIngestion = yield* ProviderRuntimeIngestionService;
@@ -26,6 +27,7 @@ export const makeOrchestrationReactor = Effect.gen(function* () {
   const threadTabSettlementReactor = yield* ThreadTabSettlementReactor.ThreadTabSettlementReactor;
   const usageLimitResumeReactor = yield* UsageLimitResumeReactor.UsageLimitResumeReactor;
   const pullRequestSyncReactor = yield* PullRequestSyncReactor.PullRequestSyncReactor;
+  const pullRequestWatchReactor = yield* PullRequestWatchReactor.PullRequestWatchReactor;
   const threadPullRequestReactor = yield* ThreadPullRequestReactor.ThreadPullRequestReactor;
   const agentAwarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay;
   const storageCleanup = yield* StorageCleanup.StorageCleanup;
@@ -40,6 +42,7 @@ export const makeOrchestrationReactor = Effect.gen(function* () {
     yield* threadTabSettlementReactor.start();
     yield* usageLimitResumeReactor.start();
     yield* pullRequestSyncReactor.start();
+    yield* pullRequestWatchReactor.start();
     yield* agentAwarenessRelay.start();
     yield* storageCleanup.start();
   });

apps/server/src/persistence/Layers/Sqlite.ts

@@ -8,6 +8,7 @@ import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
 import { runMigrations } from "../Migrations.ts";
 import { ensureThreadTabsSchema } from "../../threadTabs/schema.ts";
 import { ensureLinearThreadLinksSchema } from "../../linear/threadLinksSchema.ts";
+import { ensurePullRequestWatchSchema } from "../../pullRequestWatch/store.ts";
 import { ServerConfig } from "../../config.ts";
 
 // Size the -wal file is cut back to on the first commit after a WAL reset.
@@ -26,6 +27,7 @@ const setup = Layer.effectDiscard(
     yield* runMigrations();
     yield* ensureThreadTabsSchema();
     yield* ensureLinearThreadLinksSchema();
+    yield* ensurePullRequestWatchSchema();
   }),
 );
 

apps/server/src/pullRequestWatch/http.ts

@@ -0,0 +1,113 @@
+import {
+  AuthOrchestrationOperateScope,
+  AuthOrchestrationReadScope,
+  EnvironmentHttpApi,
+} from "@t3tools/contracts";
+import {
+  threadPullRequestKeysEqual,
+  visibleThreadPullRequests,
+} from "@t3tools/shared/threadPullRequests";
+import * as DateTime from "effect/DateTime";
+import * as Effect from "effect/Effect";
+import * as Option from "effect/Option";
+import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder";
+
+import {
+  annotateEnvironmentRequest,
+  failEnvironmentInternal,
+  failEnvironmentInvalidRequest,
+  failEnvironmentNotFound,
+  requireEnvironmentScope,
+} from "../auth/http.ts";
+import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts";
+import { PullRequestWatchReactor } from "./reactor.ts";
+import {
+  deletePullRequestWatches,
+  listPullRequestWatches,
+  upsertPullRequestWatch,
+} from "./store.ts";
+
+export const pullRequestWatchesHttpApiLayer = HttpApiBuilder.group(
+  EnvironmentHttpApi,
+  "pullRequestWatches",
+  Effect.fnUntraced(function* (handlers) {
+    const snapshots = yield* ProjectionSnapshotQuery;
+    const reactor = yield* PullRequestWatchReactor;
+
+    return handlers
+      .handle(
+        "list",
+        Effect.fn("environment.pullRequestWatches.list")(function* (args) {
+          yield* annotateEnvironmentRequest(args.endpoint.name);
+          yield* requireEnvironmentScope(AuthOrchestrationReadScope);
+          return yield* listPullRequestWatches(args.params.threadId).pipe(
+            Effect.catch((cause) => failEnvironmentInternal("internal_error", cause)),
+          );
+        }),
+      )
+      .handle(
+        "set",
+        Effect.fn("environment.pullRequestWatches.set")(function* (args) {
+          yield* annotateEnvironmentRequest(args.endpoint.name);
+          yield* requireEnvironmentScope(AuthOrchestrationOperateScope);
+          const { threadId } = args.params;
+          const { action, ...key } = args.payload;
+          const shell = yield* snapshots
+            .getThreadShellById(threadId)
+            .pipe(Effect.catch((cause) => failEnvironmentInternal("internal_error", cause)));
+          if (Option.isNone(shell)) return yield* failEnvironmentNotFound("thread_not_found");
+          const linked = visibleThreadPullRequests(shell.value.pullRequests).some((link) =>
+            threadPullRequestKeysEqual(link, key),
+          );
+          if (!linked && action !== "stop") {
+            return yield* failEnvironmentInvalidRequest("invalid_command");
+          }
+
+          const updatedAt = DateTime.formatIso(yield* DateTime.now);
+          yield* Effect.gen(function* () {
+            const existing = (yield* listPullRequestWatches(threadId)).find((watch) =>
+              threadPullRequestKeysEqual(watch, key),
+            );
+            switch (action) {
+              case "watch":
+                if (existing === undefined) {
+                  yield* upsertPullRequestWatch({
+                    threadId,
+                    ...key,
+                    status: "active",
+                    attemptsUsed: 0,
+                    handled: [],
+                    updatedAt,
+                  });
+                }
+                return;
+              case "pause":
+                if (existing !== undefined) {
+                  yield* upsertPullRequestWatch({ ...existing, status: "paused", updatedAt });
+                }
+                return;
+              case "resume":
+                // A fresh budget, so resuming is also how a spent watch is allowed to try again.
+                if (existing !== undefined) {
+                  yield* upsertPullRequestWatch({
+                    ...existing,
+                    status: "active",
+                    attemptsUsed: 0,
+                    updatedAt,
+                  });
+                }
+                return;
+              case "stop":
+                yield* deletePullRequestWatches(threadId, key);
+                return;
+            }
+          }).pipe(Effect.catch((cause) => failEnvironmentInternal("internal_error", cause)));
+
+          yield* reactor.evaluate(threadId);
+          return yield* listPullRequestWatches(threadId).pipe(
+            Effect.catch((cause) => failEnvironmentInternal("internal_error", cause)),
+          );
+        }),
+      );
+  }),
+);