chore: address final review follow-ups

Author: vicary

docs/ReviewProtocol.md

@@ -10,18 +10,19 @@ request and review feedback needs to be handled systematically.
 - keep fixes narrow and scoped to the verified issue
 - resolve handled review threads and request a fresh review cycle
 
-## Required Latest-Page Query
+## Required Unresolved-Batch Query
 
-Use this command exactly as written for metadata-first traversal to the latest
-review-thread page, followed by a narrow detail fetch only for unresolved
-threads found on that last page. Do not rewrite, broaden, or replace it with an
-equivalent query.
+Use this command exactly as written for metadata-first traversal across
+review-thread pages until it collects the first 10 unresolved threads. The 10
+unresolved items may be sparse, non-contiguous, and spread across multiple
+pages. After that metadata pass, fetch narrow details only for that unresolved
+batch. Do not rewrite, broaden, or replace it with an equivalent query.
 
 If this command fails for any reason, stop and report the failure explicitly
 before taking any further review-handling action.
 
 ```bash
-deno eval 'const o="OWNER",r="REPO",n="PR_NUMBER",mq="query($o:String!,$r:String!,$n:Int!,$a:String){repository(owner:$o,name:$r){pullRequest(number:$n){reviewThreads(first:20,after:$a){pageInfo{hasNextPage endCursor}nodes{id isResolved isOutdated path}}}}}",dq="query($ids:[ID!]!){nodes(ids:$ids){... on PullRequestReviewThread{id path isResolved isOutdated comments(first:10){nodes{author{login}body url createdAt}}}}}";let a=null,t;for(;;){const c=new Deno.Command("gh",{args:["api","graphql","-f",`query=${mq}`,"-F",`o=${o}`,"-F",`r=${r}`,"-F",`n=${n}`,...(a?["-F",`a=${a}`]:[])]});const x=await c.output();if(!x.success){console.error(new TextDecoder().decode(x.stderr));Deno.exit(x.code)}t=JSON.parse(new TextDecoder().decode(x.stdout)).data.repository.pullRequest.reviewThreads;if(!t.pageInfo.hasNextPage)break;a=t.pageInfo.endCursor}const u=t.nodes.filter(x=>!x.isResolved),ids=u.map(x=>x.id);let d=[];if(ids.length){const c=new Deno.Command("gh",{args:["api","graphql","-f",`query=${dq}`,...ids.flatMap(id=>["-F",`ids[]=${id}`])]});const x=await c.output();if(!x.success){console.error(new TextDecoder().decode(x.stderr));Deno.exit(x.code)}d=JSON.parse(new TextDecoder().decode(x.stdout)).data.nodes.filter(Boolean)}console.log(JSON.stringify({pageInfo:t.pageInfo,unresolved:d.length?d:u}))'
+deno eval 'const o="OWNER",r="REPO",n="PR_NUMBER",maxUnresolved=10,mq="query($o:String!,$r:String!,$n:Int!,$a:String){repository(owner:$o,name:$r){pullRequest(number:$n){reviewThreads(first:20,after:$a){pageInfo{hasNextPage endCursor}nodes{id isResolved isOutdated path}}}}}",dq="query($ids:[ID!]!){nodes(ids:$ids){... on PullRequestReviewThread{id path isResolved isOutdated comments(first:10){nodes{author{login}body url createdAt}}}}}";let a=null,t={pageInfo:{hasNextPage:false,endCursor:null},nodes:[]},u=[];for(;;){const c=new Deno.Command("gh",{args:["api","graphql","-f",`query=${mq}`,"-F",`o=${o}`,"-F",`r=${r}`,"-F",`n=${n}`,...(a?["-F",`a=${a}`]:[])]});const x=await c.output();if(!x.success){console.error(new TextDecoder().decode(x.stderr));Deno.exit(x.code)}t=JSON.parse(new TextDecoder().decode(x.stdout)).data.repository.pullRequest.reviewThreads;for(const node of t.nodes){if(!node.isResolved)u.push(node);if(u.length===maxUnresolved)break}if(u.length===maxUnresolved||!t.pageInfo.hasNextPage)break;a=t.pageInfo.endCursor}const ids=u.slice(0,maxUnresolved).map(x=>x.id);let d=[];if(ids.length){const c=new Deno.Command("gh",{args:["api","graphql","-f",`query=${dq}`,...ids.flatMap(id=>["-F",`ids[]=${id}`])]});const x=await c.output();if(!x.success){console.error(new TextDecoder().decode(x.stderr));Deno.exit(x.code)}d=JSON.parse(new TextDecoder().decode(x.stdout)).data.nodes.filter(Boolean)}console.log(JSON.stringify({pageInfo:t.pageInfo,batchSize:ids.length,exhausted:!t.pageInfo.hasNextPage&&ids.length<maxUnresolved,unresolved:d.length?d:u.slice(0,maxUnresolved)}))'
 ```
 
 ## Workflow
@@ -35,17 +36,21 @@ deno eval 'const o="OWNER",r="REPO",n="PR_NUMBER",mq="query($o:String!,$r:String
    - Use GraphQL `reviewThreads` as the source of truth for unresolved state;
      REST review comments do not expose thread resolution and cannot be filtered
      to unresolved-only.
-   - Run the required latest-page query command exactly as written in
-     `Required Latest-Page Query`.
+   - Run the required unresolved-batch query command exactly as written in
+     `Required Unresolved-Batch Query`.
    - If the command fails, stop and report the failure explicitly.
    - Keep GraphQL payloads narrow: request small pages (`first: 20` or similar)
      and fetch only thread metadata first (`id`, `isResolved`, `isOutdated`,
      `path`, `pageInfo`) while paginating. Do not request full comment bodies
      for every thread in the first pass.
-   - Traverse metadata pages sequentially until you reach the latest page, then
-     filter unresolved threads locally from that latest page result.
-   - If unresolved items are found there, run a second narrow GraphQL query only
-     for those unresolved thread IDs to fetch review contents.
+   - Traverse metadata pages sequentially until you either collect 10 unresolved
+     threads or exhaust pagination.
+   - Expect unresolved threads to be sparse and split across multiple pages; do
+     not assume they are contiguous on one page.
+   - If unresolved items are found, run a second narrow GraphQL query only for
+     those unresolved thread IDs in the current batch to fetch review contents.
+   - Re-run the same query after resolving a batch so the protocol can process
+     the next 10 unresolved items in a pseudo while loop until none remain.
    - If you need latest-first lightweight browsing, use REST review comments as
      a secondary view (`/pulls/{number}/comments?sort=updated&direction=desc`),
      but do not use REST as the authoritative unresolved-thread source.

src/services/graphiti-async.test.ts

@@ -230,4 +230,82 @@ describe("GraphitiAsyncService", () => {
       globalThis.clearTimeout = originalClearTimeout;
     }
   });
+
+  it("does not start a second drain while a slow drain is still in flight", async () => {
+    const originalSetTimeout = globalThis.setTimeout;
+    const originalClearTimeout = globalThis.clearTimeout;
+
+    let recoveryHandler: (() => void) | undefined;
+    const clearedTimers: unknown[] = [];
+    const drainDeferred = deferred<{ status: "idle" }>();
+    let drainCalls = 0;
+
+    globalThis.setTimeout = ((handler: TimerHandler, timeout?: number) => {
+      if (timeout === 50) {
+        recoveryHandler = handler as () => void;
+      }
+      return { handler, timeout } as unknown as ReturnType<typeof setTimeout>;
+    }) as typeof setTimeout;
+    globalThis.clearTimeout = ((timer: unknown) => {
+      clearedTimers.push(timer);
+    }) as typeof clearTimeout;
+
+    try {
+      const service = new GraphitiAsyncService(
+        {
+          getEpisodes() {
+            return Promise.resolve([]);
+          },
+          searchMemoryFacts() {
+            return Promise.resolve([]);
+          },
+          searchNodesWithStatus() {
+            return Promise.resolve({ nodes: [], degraded: true });
+          },
+        } as never,
+        {
+          get() {
+            return Promise.resolve(null);
+          },
+          getMeta() {
+            return Promise.resolve(null);
+          },
+          rememberRefreshQuery() {
+            return Promise.resolve();
+          },
+          set() {
+            return Promise.resolve();
+          },
+        } as never,
+        {
+          drainGroup() {
+            drainCalls += 1;
+            return drainDeferred.promise;
+          },
+        } as never,
+        25,
+        50,
+      );
+
+      service.scheduleDrain("group-1");
+      await Promise.resolve();
+      service.scheduleDrain("group-1");
+
+      assertEquals(drainCalls, 1);
+      assert(recoveryHandler);
+
+      recoveryHandler();
+      await Promise.resolve();
+
+      assertEquals(drainCalls, 1);
+
+      drainDeferred.resolve({ status: "idle" });
+      await service.dispose();
+
+      assertEquals(drainCalls, 1);
+    } finally {
+      globalThis.setTimeout = originalSetTimeout;
+      globalThis.clearTimeout = originalClearTimeout;
+    }
+  });
 });

src/services/graphiti-async.ts

@@ -5,6 +5,7 @@ import { logger } from "./logger.ts";
 import type { RedisCacheService } from "./redis-cache.ts";
 
 export class GraphitiAsyncService {
+  private static readonly DEFAULT_DRAIN_RECOVERY_DELAY_MS = 30_000;
   private readonly drainInFlight = new Map<string, Promise<void>>();
   private readonly drainRetryTimers = new Map<
     string,
@@ -26,6 +27,8 @@ export class GraphitiAsyncService {
     private readonly cache: RedisCacheService,
     private readonly drain: BatchDrainService,
     private readonly drainRetryDelayMs = 1_000,
+    private readonly drainRecoveryDelayMs =
+      GraphitiAsyncService.DEFAULT_DRAIN_RECOVERY_DELAY_MS,
   ) {}
 
   async flushPendingGroups(groupIds: Iterable<string>): Promise<void> {
@@ -97,9 +100,11 @@ export class GraphitiAsyncService {
       if (!recovery || recovery.run !== run) return;
       this.drainRecoveryTimers.delete(groupId);
       if (this.drainInFlight.get(groupId) !== run) return;
-      this.drainInFlight.delete(groupId);
-      this.scheduleDrain(groupId);
-    }, this.drainRetryDelayMs);
+      logger.warn(
+        "Graphiti drain recovery timeout exceeded; leaving in-flight drain intact",
+        { groupId, timeoutMs: this.drainRecoveryDelayMs },
+      );
+    }, this.drainRecoveryDelayMs);
 
     this.drainRecoveryTimers.set(groupId, { run, timer });
   }