fix(server): propagate linked pr changes and settle threads immediately

apps/server/src/orchestration/PullRequestSyncReactor.test.ts

@@ -1,9 +1,11 @@
 import {
+  EventId,
   ProjectId,
   ProviderInstanceId,
   PullRequestOperationError,
   ThreadId,
   type OrchestrationCommand,
+  type OrchestrationEvent,
   type OrchestrationProjectShell,
   type OrchestrationShellSnapshot,
   type OrchestrationThreadShell,
@@ -18,6 +20,7 @@ import * as Crypto from "effect/Crypto";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
+import * as PubSub from "effect/PubSub";
 import * as Queue from "effect/Queue";
 import * as Ref from "effect/Ref";
 import * as Stream from "effect/Stream";
@@ -31,6 +34,7 @@ import {
 } from "./Services/OrchestrationEngine.ts";
 import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts";
 import * as PullRequestSyncReactor from "./PullRequestSyncReactor.ts";
+import { resolveAutoSettlementAt } from "./ThreadSettlementPolicy.ts";
 
 const NOW = "2026-08-28T12:00:00.000Z";
 const PROJECT_ID = ProjectId.make("sync-project");
@@ -152,6 +156,7 @@ function makeSummary(
 }
 
 interface HarnessOptions {
+  readonly onDispatch?: (command: SyncCommand | LinkCommand) => Effect.Effect<void>;
   readonly invalidate?: PullRequestService["Service"]["invalidate"];
   readonly snapshot: OrchestrationShellSnapshot;
   readonly summary?: (
@@ -165,6 +170,7 @@ interface HarnessOptions {
 const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: HarnessOptions) {
   const activation = yield* Deferred.make<void>();
   const snapshots = yield* Ref.make(options.snapshot);
+  const events = yield* PubSub.unbounded<OrchestrationEvent>();
   const snapshotReads = yield* Queue.unbounded<void>();
   const syncCommands = yield* Ref.make<ReadonlyArray<SyncCommand>>([]);
   const linkCommands = yield* Ref.make<ReadonlyArray<LinkCommand>>([]);
@@ -188,11 +194,13 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
   const dispatch: OrchestrationEngineShape["dispatch"] = (command) => {
     if (command.type === "thread.pull-request-link.sync") {
       return Ref.update(syncCommands, (recorded) => [...recorded, command]).pipe(
+        Effect.andThen(options.onDispatch?.(command) ?? Effect.void),
         Effect.as({ sequence: 1 }),
       );
     }
     if (command.type === "thread.pull-request.link") {
       return Ref.update(linkCommands, (recorded) => [...recorded, command]).pipe(
+        Effect.andThen(options.onDispatch?.(command) ?? Effect.void),
         Effect.as({ sequence: 1 }),
       );
     }
@@ -213,13 +221,17 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
       readEvents: () => Stream.empty,
       dispatch,
       streamDomainEvents: Stream.empty,
+      subscribeDomainEvents: PubSub.subscribe(events).pipe(
+        Effect.map((subscription) => Stream.fromSubscription(subscription)),
+      ),
       latestSequence: Effect.succeed(0),
     }),
     Layer.succeed(ServerActivation, Deferred.await(activation)),
     Layer.succeed(Crypto.Crypto, testCrypto),
   );
 
   return {
+    events,
     activation,
     snapshots,
     snapshotReads,
@@ -274,6 +286,42 @@ function applySync(
 }
 
 describe("PullRequestSyncReactor", () => {
+  it.effect("syncs a newly linked merged PR without waiting for the periodic sweep", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        yield* TestClock.setTime(Date.parse(NOW));
+        const fixture = yield* makeHarness({
+          snapshot: makeSnapshot([makeThread("one")]),
+          summary: (input) =>
+            Effect.succeed(makeSummary(input, { state: "merged", mergedAt: NOW })),
+        });
+        yield* Effect.gen(function* () {
+          const reactor = yield* startAndSweep(fixture);
+          const link = makeLink(42);
+          yield* Ref.set(
+            fixture.snapshots,
+            makeSnapshot([makeThread("one", { pullRequests: [link] })]),
+          );
+          yield* PubSub.publish(fixture.events, {
+            type: "thread.pull-request-linked",
+            sequence: 2,
+            eventId: EventId.make("linked"),
+            aggregateKind: "thread",
+            aggregateId: ThreadId.make("one"),
+            occurredAt: NOW,
+            commandId: null,
+            causationEventId: null,
+            correlationId: null,
+            metadata: {},
+            payload: { threadId: ThreadId.make("one"), link, updatedAt: NOW },
+          });
+          yield* Queue.take(fixture.snapshotReads);
+          yield* reactor.drain;
+          assert.strictEqual((yield* Ref.get(fixture.syncCommands))[0]?.snapshot.state, "merged");
+        }).pipe(Effect.provide(fixture.layer));
+      }),
+    ),
+  );
   it.effect("retries a failed stack read after the summary becomes terminal", () =>
     Effect.scoped(
       Effect.gen(function* () {
@@ -303,6 +351,7 @@ describe("PullRequestSyncReactor", () => {
         yield* Effect.gen(function* () {
           const reactor = yield* startAndSweep(fixture);
           const commands = yield* Ref.get(fixture.syncCommands);
+          assert.deepStrictEqual(commands, []);
           yield* Ref.update(fixture.snapshots, (snapshot) => applySync(snapshot, commands));
           yield* sweepAgain(fixture, reactor);
           assert.strictEqual(attempts, 2);
@@ -315,6 +364,85 @@ describe("PullRequestSyncReactor", () => {
     ),
   );
 
+  it.effect("retries a failed sibling link before publishing a terminal snapshot", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        yield* TestClock.setTime(Date.parse(NOW));
+        let failSibling = true;
+        const fixture = yield* makeHarness({
+          snapshot: makeSnapshot([
+            makeThread("one", { pullRequests: [makeLink(7, { state: "closed" })] }),
+          ]),
+          summary: (input) =>
+            Effect.succeed(makeSummary(input, { state: "merged", mergedAt: NOW })),
+          stack: () =>
+            Effect.succeed({
+              id: "stack",
+              number: 7,
+              url: "https://github.com/owner/repository/stacks/7",
+              base: "main",
+              layers: [
+                { number: 7, headBranch: "feature", state: "merged" },
+                { number: 8, headBranch: "sibling", state: "open" },
+              ],
+            }),
+          onDispatch: (command) =>
+            command.type === "thread.pull-request.link" && failSibling
+              ? Effect.die("temporary link failure")
+              : Effect.void,
+        });
+        yield* Effect.gen(function* () {
+          const reactor = yield* startAndSweep(fixture);
+          assert.deepStrictEqual(yield* Ref.get(fixture.syncCommands), []);
+          failSibling = false;
+          yield* sweepAgain(fixture, reactor);
+          assert.strictEqual((yield* Ref.get(fixture.syncCommands))[0]?.snapshot.state, "merged");
+          assert.strictEqual((yield* Ref.get(fixture.linkCommands)).at(-1)?.number, 8);
+        }).pipe(Effect.provide(fixture.layer));
+      }),
+    ),
+  );
+
+  it.effect("concurrent stack reads persist a shared sibling once before syncing both roots", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const readsReady = yield* Deferred.make<void>();
+        let reads = 0;
+        let linked = false;
+        const fixture = yield* makeHarness({
+          snapshot: makeSnapshot([makeThread("one", { pullRequests: [makeLink(7), makeLink(8)] })]),
+          stack: () =>
+            Effect.gen(function* () {
+              if (++reads === 2) yield* Deferred.succeed(readsReady, undefined);
+              yield* Deferred.await(readsReady);
+              return {
+                id: "stack",
+                number: 7,
+                url: "https://github.com/owner/repository/stacks/7",
+                base: "main",
+                layers: [{ number: 9, headBranch: "sibling", state: "open" as const }],
+              };
+            }),
+          onDispatch: (command) =>
+            Effect.gen(function* () {
+              if (command.type === "thread.pull-request.link") {
+                yield* Effect.yieldNow;
+                assert.strictEqual(linked, false);
+                linked = true;
+              } else {
+                assert.strictEqual(linked, true);
+              }
+            }),
+        });
+        yield* Effect.gen(function* () {
+          yield* startAndSweep(fixture);
+          assert.strictEqual((yield* Ref.get(fixture.linkCommands)).length, 1);
+          assert.strictEqual((yield* Ref.get(fixture.syncCommands)).length, 2);
+        }).pipe(Effect.provide(fixture.layer));
+      }),
+    ),
+  );
+
   it.effect("explicit refresh reads a changed stack even when its PR summary is unchanged", () =>
     Effect.scoped(
       Effect.gen(function* () {
@@ -656,20 +784,38 @@ describe("PullRequestSyncReactor", () => {
           base: "main",
           layers: [
             { number: 41, headBranch: "layer-1", state: "merged" },
-            { number: 42, headBranch: "layer-2", state: "open" },
+            { number: 42, headBranch: "layer-2", state: "merged" },
             { number: 43, headBranch: "layer-3", state: "open" },
           ],
         };
+        let thread = makeThread("one", {
+          pullRequests: [
+            makeLink(42, { state: "open" }),
+            makeLink(41, { state: "merged" }, { source: "stack-dismissed" }),
+          ],
+        });
         const fixture = yield* makeHarness({
-          snapshot: makeSnapshot([
-            makeThread("one", {
-              pullRequests: [
-                makeLink(42),
-                makeLink(41, { state: "merged" }, { source: "stack-dismissed" }),
-              ],
-            }),
-          ]),
+          snapshot: makeSnapshot([thread]),
+          summary: (input) =>
+            Effect.succeed(makeSummary(input, { state: "merged", mergedAt: NOW })),
           stack: () => Effect.succeed(stack),
+          onDispatch: (command) =>
+            Effect.sync(() => {
+              thread =
+                command.type === "thread.pull-request.link"
+                  ? { ...thread, pullRequests: [...thread.pullRequests, makeLink(command.number)] }
+                  : applySync(makeSnapshot([thread]), [command]).threads[0]!;
+              // Every projected event may wake settlement, including the terminal root update.
+              assert.isNull(
+                resolveAutoSettlementAt({
+                  thread,
+                  pullRequest: null,
+                  now: NOW,
+                  autoSettleAfterDays: null,
+                  autoSettleOnMerge: true,
+                }),
+              );
+            }),
         });
 
         yield* Effect.gen(function* () {