fix(server): sync pull request draft status changes

apps/server/src/git/GitManager.test.ts

@@ -17,6 +17,7 @@ import * as PlatformError from "effect/PlatformError";
 import * as Result from "effect/Result";
 import * as Schema from "effect/Schema";
 import * as Scope from "effect/Scope";
+import * as Stream from "effect/Stream";
 import { ChildProcessSpawner } from "effect/process";
 import { expect } from "vite-plus/test";
 import type {
@@ -1076,6 +1077,62 @@ it.layer(layerGitManagerTest)("GitManager", (it) => {
     }),
   );
 
+  it.effect("branch PR lookup notifies linked threads when draft status changes", () =>
+    Effect.gen(function* () {
+      const repoDir = yield* makeTempDir("t3code-git-manager-");
+      yield* initRepo(repoDir);
+      yield* runGit(repoDir, ["checkout", "-b", "feature/draft-status"]);
+      const remoteDir = yield* createBareRemote();
+      yield* runGit(repoDir, ["remote", "add", "origin", remoteDir]);
+      yield* runGit(repoDir, ["push", "-u", "origin", "feature/draft-status"]);
+      const drafts = [false, true, true, false, false, true];
+      const pullRequest = {
+        number: 1,
+        title: "Draft status",
+        url: "https://github.com/pingdotgg/codething-mvp/pull/1",
+        baseRefName: "main",
+        headRefName: "feature/draft-status",
+        state: "OPEN",
+      };
+      const { manager } = yield* makeManager({
+        ghScenario: {
+          prListSequence: [
+            ...drafts.map((isDraft) => encodeCliJson([{ ...pullRequest, isDraft }])),
+            encodeCliJson([
+              {
+                ...pullRequest,
+                number: 2,
+                url: "https://github.com/pingdotgg/codething-mvp/pull/2",
+                state: "MERGED",
+              },
+            ]),
+          ],
+        },
+      });
+      const changes = yield* manager.subscribePullRequestStateChanges;
+      for (const isDraft of drafts) {
+        const pr = yield* manager.branchPullRequest(
+          { cwd: repoDir, branch: "feature/draft-status" },
+          { refresh: true },
+        );
+        expect(pr?.isDraft === true).toBe(isDraft);
+      }
+      // A different PR ends collection without waiting on a background stream consumer.
+      yield* manager.branchPullRequest(
+        { cwd: repoDir, branch: "feature/draft-status" },
+        { refresh: true },
+      );
+      const events = yield* Stream.runCollect(
+        changes.pipe(Stream.takeUntil((event) => event.number === 2, { excludeLast: true })),
+      );
+      expect(events).toEqual([
+        { host: "github.com", repository: "pingdotgg/codething-mvp", number: 1 },
+        { host: "github.com", repository: "pingdotgg/codething-mvp", number: 1 },
+        { host: "github.com", repository: "pingdotgg/codething-mvp", number: 1 },
+      ]);
+    }),
+  );
+
   it.effect("branch PR lookup uses a saved tracked branch without changing checkout", () =>
     Effect.gen(function* () {
       const repoDir = yield* makeTempDir("t3code-git-manager-");

apps/server/src/git/GitManager.ts

@@ -1192,21 +1192,27 @@ export const make = Effect.gen(function* () {
     return prLookupFailureTtl(streak);
   };
   const pullRequestStateChanges = yield* PubSub.sliding<ThreadPullRequestKey>(64);
-  const lookupStates = new Map<string, PullRequestInfo["state"]>();
+  const lookupStates = new Map<
+    string,
+    { readonly state: PullRequestInfo["state"]; readonly isDraft: boolean }
+  >();
   const noteLookupState = (pr: PullRequestInfo) =>
     Effect.suspend(() => {
       const parsed = parseChangeRequestUrl(pr.url);
       if (parsed === null || parsed.number !== pr.number) return Effect.void;
       const key = normalizeThreadPullRequestKey(parsed);
       const id = threadPullRequestKeyOf(key);
-      const previous = lookupStates.get(id) ?? "open";
+      const previous = lookupStates.get(id);
       lookupStates.delete(id);
       if (lookupStates.size >= PR_LOOKUP_CACHE_CAPACITY) {
         const oldest = lookupStates.keys().next().value;
         if (oldest !== undefined) lookupStates.delete(oldest);
       }
-      lookupStates.set(id, pr.state);
-      return previous === pr.state ? Effect.void : PubSub.publish(pullRequestStateChanges, key);
+      lookupStates.set(id, { state: pr.state, isDraft: pr.isDraft === true });
+      return (previous?.state ?? "open") === pr.state &&
+        (previous?.isDraft === true) === (pr.isDraft === true)
+        ? Effect.void
+        : PubSub.publish(pullRequestStateChanges, key);
     });
   const prLookupCache = yield* Cache.makeWith(
     (key: string) => {

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

@@ -913,6 +913,43 @@ describe("PullRequestSyncReactor", () => {
     ),
   );
 
+  it.effect.each([true, false])(
+    "syncs linked draft status %s as soon as a reader reports a change",
+    (isDraft) =>
+      Effect.gen(function* () {
+        yield* TestClock.setTime(Date.parse(NOW));
+        const draft = yield* Ref.make(!isDraft);
+        const invalidated = yield* Ref.make<ReadonlyArray<number>>([]);
+        const fixture = yield* makeHarness({
+          snapshot: makeSnapshot([
+            makeThread("agent", { pullRequests: [makeLink(7, { isDraft: !isDraft })] }),
+          ]),
+          summary: (input) =>
+            Ref.get(draft).pipe(Effect.map((isDraft) => makeSummary(input, { isDraft }))),
+          invalidate: ({ reference }) =>
+            Ref.update(invalidated, (numbers) => [...numbers, reference?.number ?? -1]),
+        });
+
+        yield* Effect.gen(function* () {
+          const reactor = yield* startAndSweep(fixture);
+          yield* Ref.set(draft, isDraft);
+          yield* Queue.offer(fixture.stateChanges, {
+            host: "github.com",
+            repository: "owner/repository",
+            number: 7,
+          });
+          yield* Queue.take(fixture.snapshotReads);
+          yield* reactor.drain;
+
+          assert.deepStrictEqual(yield* Ref.get(invalidated), [7]);
+          assert.deepStrictEqual(
+            (yield* Ref.get(fixture.syncCommands)).map((command) => command.snapshot.isDraft),
+            [isDraft],
+          );
+        }).pipe(Effect.provide(fixture.layer));
+      }),
+  );
+
   it.effect("preserves a refresh requested while an older host read is in flight", () =>
     Effect.scoped(
       Effect.gen(function* () {

apps/server/src/pullRequest/PullRequestService.test.ts

@@ -2680,6 +2680,74 @@ it.effect(
     }),
 );
 
+it.effect("notifies linked threads when detail changes draft status in either direction", () =>
+  Effect.gen(function* () {
+    let isDraft = true;
+    let updatedAt = "2026-07-02T00:00:00Z";
+    const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
+    const service = yield* makeService({
+      projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })],
+      providers: [
+        fakeProvider(SourceControlProviderKind.make("github"), {
+          getChangeRequest: () =>
+            Effect.sync(() => ({ ...hostedChangeRequest("body"), isDraft, updatedAt })),
+        }),
+      ],
+    });
+    const changes = yield* service.subscribeStateChanges;
+    const read = () => service.detail({ ...reference, allowStale: false });
+    yield* read();
+    yield* read();
+    isDraft = false;
+    yield* service.invalidate({ reference });
+    yield* read();
+    yield* read();
+    isDraft = true;
+    yield* service.invalidate({ reference });
+    yield* read();
+
+    isDraft = false;
+    updatedAt = "2026-07-01T00:00:00Z";
+    yield* service.invalidate({ reference });
+    yield* read();
+
+    // A different PR ends collection after all the synchronous notifications above.
+    yield* service.reportState({ reference: { ...reference, number: 2 }, state: "open" });
+    const events = yield* Stream.runCollect(
+      changes.pipe(Stream.takeUntil((event) => event.number === 2, { excludeLast: true })),
+    );
+    assert.deepStrictEqual(events, [
+      { host: "github.com", repository: "acme/web", number: 1 },
+      { host: "github.com", repository: "acme/web", number: 1 },
+      { host: "github.com", repository: "acme/web", number: 1 },
+    ]);
+  }),
+);
+
+it.effect("deduplicates routed PR reports without dropping draft changes", () =>
+  Effect.gen(function* () {
+    const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
+    const service = yield* makeService({
+      projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })],
+      providers: [fakeProvider(SourceControlProviderKind.make("github"))],
+    });
+    const changes = yield* service.subscribeStateChanges;
+    for (const isDraft of [true, true, false, false, true]) {
+      yield* service.reportState({ reference, state: "open", isDraft });
+    }
+    yield* service.reportState({ reference, state: "open" });
+    yield* service.reportState({ reference, state: "open" });
+    yield* service.reportState({ reference: { ...reference, number: 2 }, state: "open" });
+    const events = yield* Stream.runCollect(
+      changes.pipe(Stream.takeUntil((event) => event.number === 2, { excludeLast: true })),
+    );
+    assert.deepStrictEqual(
+      events.map((event) => event.number),
+      [1, 1, 1, 1],
+    );
+  }),
+);
+
 it.effect("reuses an observed merged state for strict settlement reads", () =>
   Effect.gen(function* () {
     const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };

apps/server/src/pullRequest/PullRequestService.ts

@@ -102,7 +102,10 @@ export interface PullRequestMergeEvent extends PullRequestRef {
   readonly mergedAt: string;
 }
 
-type PullRequestReading = Pick<PullRequestSummary, "state" | "updatedAt" | "observedAt">;
+type PullRequestReading = Pick<
+  PullRequestSummary,
+  "state" | "isDraft" | "updatedAt" | "observedAt"
+>;
 
 function supersedesReading(current: PullRequestReading | undefined, next: PullRequestReading) {
   if (current === undefined) return true;
@@ -2891,10 +2894,12 @@ export const make = Effect.gen(function* () {
       }
       detailStates.set(scope, {
         state: next.state,
+        isDraft: next.isDraft,
         updatedAt: next.updatedAt,
         observedAt: next.observedAt,
       });
-      return current?.state !== next.state && ref.host !== undefined
+      return (current?.state !== next.state || current?.isDraft !== next.isDraft) &&
+        ref.host !== undefined
         ? PubSub.publish(stateChanges, {
             host: ref.host,
             repository: ref.repository,
@@ -2906,22 +2911,34 @@ export const make = Effect.gen(function* () {
   // must never outrank a reading this environment made itself.
   const reportedStates = new Map<
     string,
-    { readonly state: PullRequestState; readonly atMs: number }
+    {
+      readonly state: PullRequestState;
+      readonly isDraft: boolean | undefined;
+      readonly atMs: number;
+    }
   >();
-  const reportState: PullRequestService["Service"]["reportState"] = ({ reference, state }) =>
+  const reportState: PullRequestService["Service"]["reportState"] = ({
+    reference,
+    state,
+    isDraft,
+  }) =>
     Effect.all([canonicalRef(reference), Clock.currentTimeMillis]).pipe(
       Effect.flatMap(([ref, nowMs]) =>
         Effect.suspend(() => {
           const scope = refScope(ref);
           const reported = reportedStates.get(scope);
-          if (reported?.state === state && nowMs - reported.atMs < REPORTED_STATE_DEDUPE_MS)
+          if (
+            reported?.state === state &&
+            reported.isDraft === isDraft &&
+            nowMs - reported.atMs < REPORTED_STATE_DEDUPE_MS
+          )
             return Effect.void;
           reportedStates.delete(scope);
           if (reportedStates.size >= REF_EPOCH_CAPACITY) {
             const oldest = reportedStates.keys().next().value;
             if (oldest !== undefined) reportedStates.delete(oldest);
           }
-          reportedStates.set(scope, { state, atMs: nowMs });
+          reportedStates.set(scope, { state, isDraft, atMs: nowMs });
           return ref.host === undefined
             ? Effect.void
             : PubSub.publish(stateChanges, {

packages/client-runtime/src/state/pullRequestRouting.ts

@@ -61,7 +61,9 @@ const writes = new Set<string>([
   WS_METHODS.pullRequestsSetLabels,
 ]);
 const isRef = Schema.is(PullRequestRef);
-const hasState = Schema.is(Schema.Struct({ state: PullRequestState }));
+const hasState = Schema.is(
+  Schema.Struct({ state: PullRequestState, isDraft: Schema.optional(Schema.Boolean) }),
+);
 const isInvalidation = Schema.is(PullRequestInvalidateInput);
 const readTimeout = (environmentId: EnvironmentId) =>
   Effect.timeoutOrElse({
@@ -223,14 +225,13 @@ const invalidateTarget = Effect.fn("PullRequestRouting.invalidateTarget")(functi
  */
 const reportState = (
   { allowStale: _allowStale, expectedAccountId: _expectedAccountId, ...reference }: PullRequestRef,
-  state: PullRequestState,
+  status: { readonly state: PullRequestState; readonly isDraft?: boolean | undefined },
 ) =>
-  request(WS_METHODS.pullRequestsReportState, { reference, state }).pipe(
-    Effect.timeoutOption("5 seconds"),
-    Effect.ignore,
-    Effect.forkDetach,
-    Effect.asVoid,
-  );
+  request(WS_METHODS.pullRequestsReportState, {
+    reference,
+    state: status.state,
+    ...(status.isDraft === undefined ? {} : { isDraft: status.isDraft }),
+  }).pipe(Effect.timeoutOption("5 seconds"), Effect.ignore, Effect.forkDetach, Effect.asVoid);
 
 /** Credentials stay on their environments. Only a verified host and account cross the wire. */
 export function createPullRequestRouter() {
@@ -378,7 +379,7 @@ export function createPullRequestRouter() {
       ).pipe(
         Effect.tap((result) =>
           id !== origin.target.environmentId && hasState(result)
-            ? reportState(ref, result.state)
+            ? reportState(ref, result)
             : Effect.void,
         ),
         Effect.tap(() =>