chore: address latest review follow-ups

Author: vicary

docs/ReviewProtocol.md

@@ -10,6 +10,20 @@ 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
+
+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.
+
+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}))'
+```
+
 ## Workflow
 
 1. Detect the active PR.
@@ -21,22 +35,20 @@ request and review feedback needs to be handled systematically.
    - 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`.
+   - 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`). Do not request full comment bodies for every thread in
-     the first pass.
-   - Filter unresolved threads locally from that metadata page set, then run a
-     second narrow GraphQL query only for the unresolved thread IDs you still
-     need details for.
+     `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.
    - 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.
-   - Minified unresolved-only page pattern:
-     `gh api graphql -f query='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}}}}}' -F o=OWNER -F r=REPO -F n=PR_NUMBER > /tmp/review-page.json && python3 - <<'PY'
-import json
-with open('/tmp/review-page.json') as f:t=json.load(f)['data']['repository']['pullRequest']['reviewThreads']
-print(json.dumps({'pageInfo':t['pageInfo'],'unresolved':[x for x in t['nodes'] if not x['isResolved']]},separators=(',',':')))
-PY`
    - Normalize unresolved items into discrete review claims with:
      - thread id
      - file/path

src/config.test.ts

@@ -14,6 +14,7 @@ function makeAdapter(options?: {
   searchResult?: unknown | null;
   loadResult?: unknown | null;
   searchError?: Error;
+  loadError?: Error;
 }): ConfigExplorerAdapter {
   return {
     search() {
@@ -23,6 +24,7 @@ function makeAdapter(options?: {
         : { config: options.searchResult };
     },
     load() {
+      if (options?.loadError) throw options.loadError;
       return options?.loadResult == null
         ? null
         : { config: options.loadResult };
@@ -243,6 +245,23 @@ describe("config", () => {
     assertEquals(config.redis.batchSize, 20);
   });
 
+  it("fails open to defaults when the legacy config file cannot be loaded", () => {
+    using _homedir = stub(os, "homedir", () => "/users/tester");
+    setConfigExplorerAdapterForTesting(() =>
+      makeAdapter({
+        loadError: new Error("legacy load failed"),
+      })
+    );
+
+    const config = loadConfig();
+
+    assertEquals(config.graphiti.endpoint, "http://localhost:8000/mcp");
+    assertEquals(config.graphiti.groupIdPrefix, "opencode");
+    assertEquals(config.graphiti.driftThreshold, 0.5);
+    assertEquals(config.redis.endpoint, "redis://localhost:6379");
+    assertEquals(config.redis.batchSize, 20);
+  });
+
   it("fails open to defaults when config discovery initialization fails", () => {
     setConfigExplorerAdapterForTesting(() => {
       throw new Error("cosmiconfig unavailable");

src/config.ts

@@ -305,10 +305,11 @@ const loadLegacyConfig = (
   );
 };
 
-const isDiscoveryInfrastructureFailure = (error: unknown): boolean =>
+const isRecoverableConfigLoadFailure = (error: unknown): boolean =>
   error instanceof ConfigLoadError &&
   (error.code === "config-discovery-init" ||
-    error.code === "config-discovery-search");
+    error.code === "config-discovery-search" ||
+    error.code === "config-file-load");
 
 export function loadConfig(directory?: string): GraphitiConfig {
   try {
@@ -320,7 +321,7 @@ export function loadConfig(directory?: string): GraphitiConfig {
   } catch (error) {
     if (
       !(error instanceof ConfigLoadError) ||
-      !isDiscoveryInfrastructureFailure(error)
+      !isRecoverableConfigLoadFailure(error)
     ) {
       throw error;
     }

src/services/batch-drain.test.ts

@@ -335,6 +335,41 @@ describe("batch drain", () => {
     }
   });
 
+  it("serializes claim heartbeat refreshes so they never overlap", async () => {
+    const { events, drain } = await createDeps({
+      events: { claimLockTtlSeconds: 2 },
+      drain: { batchSize: 1, claimHeartbeatIntervalMs: 250 },
+    });
+    const event = createSessionEvent("message", "user", {
+      summary: "long running",
+      body: "long running",
+    });
+    await events.recordEvent("session-1", "group-1", event);
+
+    const originalRefreshClaimLease = events.refreshClaimLease.bind(events);
+    let inFlight = 0;
+    let maxInFlight = 0;
+    let refreshCalls = 0;
+    events.refreshClaimLease = async (...args) => {
+      refreshCalls += 1;
+      inFlight += 1;
+      maxInFlight = Math.max(maxInFlight, inFlight);
+      await new Promise((resolve) => setTimeout(resolve, 300));
+      inFlight -= 1;
+      return await originalRefreshClaimLease(...args);
+    };
+
+    const result = await drain.drainGroup("group-1", {
+      async addMemory() {
+        await new Promise((resolve) => setTimeout(resolve, 650));
+      },
+    } as never);
+
+    assertEquals(result, { status: "success", drained: 1 });
+    assertEquals(refreshCalls >= 3, true);
+    assertEquals(maxInFlight, 1);
+  });
+
   it("limits batches using serialized Graphiti episode bodies", async () => {
     const first = createSessionEvent("message", "user", {
       summary: "first",

src/services/batch-drain.ts

@@ -191,22 +191,41 @@ export class BatchDrainService {
     }
 
     let lostClaim = false;
-    const refreshClaimOwnership = async (): Promise<boolean> => {
-      if (lostClaim) return false;
-      try {
-        const refreshed = await this.events.refreshClaimLease(
-          groupId,
-          claimed.claimToken,
-          claimed.lockTtlSeconds,
-        );
-        if (!refreshed) lostClaim = true;
-      } catch {
-        lostClaim = true;
-      }
-      return !lostClaim;
+    let claimRefreshChain: Promise<void> = Promise.resolve();
+    let heartbeatTimer: number | null = null;
+    let refreshClaimHeartbeatRunning = false;
+    const refreshClaimOwnership = (): Promise<boolean> => {
+      const refreshTask = claimRefreshChain.then(async () => {
+        if (lostClaim) return false;
+        try {
+          const refreshed = await this.events.refreshClaimLease(
+            groupId,
+            claimed.claimToken,
+            claimed.lockTtlSeconds,
+          );
+          if (!refreshed) lostClaim = true;
+        } catch {
+          lostClaim = true;
+        }
+        return !lostClaim;
+      });
+      claimRefreshChain = refreshTask.then(() => undefined, () => undefined);
+      return refreshTask;
     };
     const refreshClaimHeartbeat = async (): Promise<void> => {
-      await refreshClaimOwnership();
+      if (refreshClaimHeartbeatRunning) return;
+      refreshClaimHeartbeatRunning = true;
+      try {
+        await refreshClaimOwnership();
+      } finally {
+        refreshClaimHeartbeatRunning = false;
+        if (!lostClaim) {
+          heartbeatTimer = setTimeout(
+            refreshClaimHeartbeat,
+            this.getClaimHeartbeatIntervalMs(claimed.lockTtlSeconds),
+          ) as unknown as number;
+        }
+      }
     };
     const confirmClaimOwnership = (): Promise<boolean> =>
       refreshClaimOwnership();
@@ -215,9 +234,10 @@ export class BatchDrainService {
         throw new DrainClaimLostError();
       }
     };
-    const heartbeatInterval = setInterval(() => {
-      void refreshClaimHeartbeat();
-    }, this.getClaimHeartbeatIntervalMs(claimed.lockTtlSeconds));
+    heartbeatTimer = setTimeout(
+      refreshClaimHeartbeat,
+      this.getClaimHeartbeatIntervalMs(claimed.lockTtlSeconds),
+    ) as unknown as number;
     let checkpointedCount = 0;
 
     try {
@@ -300,7 +320,8 @@ export class BatchDrainService {
       logger.warn("Drain batch failed; will retry later", { groupId, err });
       return { status: "retry", drained: 0 };
     } finally {
-      clearInterval(heartbeatInterval);
+      if (heartbeatTimer !== null) clearTimeout(heartbeatTimer);
+      await claimRefreshChain;
     }
   }
 }

src/services/graphiti-mcp.test.ts

@@ -2,6 +2,11 @@ import { assertEquals, assertRejects } from "jsr:@std/assert@^1.0.0";
 import { describe, it } from "jsr:@std/testing@^1.0.0/bdd";
 import { GraphitiOfflineError } from "./connection-manager.ts";
 import { GraphitiMcpClient } from "./graphiti-mcp.ts";
+import {
+  setOpenCodeClient,
+  setSuppressConsoleWarningsDuringTestsOverride,
+  setWarningTaskScheduler,
+} from "./opencode-warning.ts";
 
 describe("GraphitiMcpClient", () => {
   it("connect rejects explicitly after stop", async () => {
@@ -57,4 +62,52 @@ describe("GraphitiMcpClient", () => {
       degraded: true,
     });
   });
+
+  it("reports searchNodesWithStatus availability warnings with the correct operation name", async () => {
+    const scheduledTasks: Array<() => void> = [];
+    const appLogCalls: unknown[] = [];
+    setSuppressConsoleWarningsDuringTestsOverride(true);
+    setWarningTaskScheduler((callback) => {
+      scheduledTasks.push(callback);
+    });
+    setOpenCodeClient({
+      app: {
+        log(input: unknown) {
+          appLogCalls.push(input);
+        },
+      },
+    });
+
+    try {
+      const client = new GraphitiMcpClient({
+        start() {},
+        stop() {
+          return Promise.resolve();
+        },
+        ready() {
+          return Promise.resolve(true);
+        },
+        callTool() {
+          return Promise.reject(new GraphitiOfflineError("offline", "offline"));
+        },
+      });
+
+      assertEquals(await client.searchNodesWithStatus({ query: "test" }), {
+        nodes: [],
+        degraded: true,
+      });
+      assertEquals(scheduledTasks.length, 1);
+      assertEquals(appLogCalls.length, 0);
+      for (const task of scheduledTasks) task();
+      assertEquals(
+        (appLogCalls[0] as { body: { extra: { operation: string } } }).body
+          .extra.operation,
+        "searchNodesWithStatus",
+      );
+    } finally {
+      setOpenCodeClient(undefined);
+      setWarningTaskScheduler(undefined);
+      setSuppressConsoleWarningsDuringTestsOverride(undefined);
+    }
+  });
 });

src/services/graphiti-mcp.ts

@@ -214,7 +214,7 @@ export class GraphitiMcpClient {
         notifyGraphitiAvailabilityIssue(
           "Graphiti unavailable; continuing without memory nodes.",
           {
-            operation: "searchNodes",
+            operation: "searchNodesWithStatus",
             err,
           },
         );