fix: address latest review follow-ups

Author: vicary

src/services/batch-drain.test.ts

@@ -559,6 +559,37 @@ describe("batch drain", () => {
     assertEquals(await redis.getListLength(drainPendingKey("group-1")), 0);
   });
 
+  it("clears parsed but invalid retry state before retrying a batch", async () => {
+    const invalidStates = [
+      { attempts: -1, nextAttemptAt: 0 },
+      { attempts: 1, nextAttemptAt: "later" },
+    ];
+
+    for (const invalidState of invalidStates) {
+      const { redis, events, drain } = await createDeps();
+      const event = createSessionEvent("message", "user", {
+        summary: "recover invalid retry state",
+        body: "recover invalid retry state",
+      });
+      await events.recordEvent("session-1", "group-1", event);
+
+      const retryKey = drainRetryKey("group-1", `${event.id}:${event.id}`);
+      await redis.setString(retryKey, JSON.stringify(invalidState), 60);
+
+      let calls = 0;
+      const result = await drain.drainGroup("group-1", {
+        addMemory() {
+          calls += 1;
+        },
+      } as never);
+
+      assertEquals(result, { status: "success", drained: 1 });
+      assertEquals(calls, 1);
+      assertEquals(await redis.getString(retryKey), null);
+      assertEquals(await redis.getListLength(drainPendingKey("group-1")), 0);
+    }
+  });
+
   it("reports only successfully ingested events when a batch dead-letters mid-batch", async () => {
     const { redis, events, drain } = await createDeps({
       drain: { batchSize: 2 },

src/services/batch-drain.ts

@@ -23,6 +23,16 @@ export interface BatchDrainServiceOptions {
 
 type RetryState = { attempts: number; nextAttemptAt: number };
 
+const isValidRetryState = (value: unknown): value is RetryState => {
+  if (!value || typeof value !== "object") return false;
+  const state = value as Partial<RetryState>;
+  return typeof state.attempts === "number" &&
+    Number.isFinite(state.attempts) &&
+    state.attempts >= 0 &&
+    typeof state.nextAttemptAt === "number" &&
+    Number.isFinite(state.nextAttemptAt);
+};
+
 class DrainClaimLostError extends Error {
   constructor() {
     super("Drain claim lease lost during batch processing");
@@ -150,7 +160,14 @@ export class BatchDrainService {
     const raw = await this.redis.getString(key);
     if (!raw) return null;
     try {
-      return JSON.parse(raw) as RetryState;
+      const parsed = JSON.parse(raw);
+      if (isValidRetryState(parsed)) return parsed;
+      await this.redis.deleteKey(key);
+      logger.warn("Cleared invalid drain retry state", {
+        groupId,
+        batchKey,
+      });
+      return null;
     } catch {
       await this.redis.deleteKey(key);
       logger.warn("Cleared corrupted drain retry state", {

src/services/connection-manager.test.ts

@@ -924,7 +924,7 @@ describe("connection manager", () => {
           }),
         }),
       Error,
-      "Invalid URL",
+      'Invalid Graphiti endpoint: "not a valid url"',
     );
   });
 

src/services/connection-manager.ts

@@ -120,7 +120,18 @@ const validateEndpoint = (endpoint: string): string => {
   if (!normalized) {
     throw new Error("Graphiti endpoint must not be empty");
   }
-  new URL(normalized);
+
+  try {
+    new URL(normalized);
+  } catch (cause) {
+    throw new Error(
+      `Invalid Graphiti endpoint: ${JSON.stringify(normalized)}`,
+      {
+        cause,
+      },
+    );
+  }
+
   return normalized;
 };
 

src/services/graphiti-async.test.ts

@@ -248,6 +248,77 @@ describe("GraphitiAsyncService", () => {
     assertEquals(timers.clearedTimers.length, 1);
   });
 
+  it("preserves fact-only cache refreshes when node search degrades", async () => {
+    const cacheSets: Array<{
+      query: string;
+      nodes: Array<{ uuid: string; name: string; summary: string }>;
+      episodeSummaries?: string[];
+      nodeRefs: string[];
+    }> = [];
+
+    const service = new GraphitiAsyncService(
+      {
+        getEpisodes() {
+          return Promise.resolve([]);
+        },
+        searchMemoryFacts() {
+          return Promise.resolve([
+            {
+              fact: "fact:alpha",
+              source_node: { name: "Source" },
+              target_node: { name: "alpha" },
+            },
+          ]);
+        },
+        searchNodesWithStatus() {
+          return Promise.resolve({
+            nodes: [{ uuid: "node:alpha", name: "alpha", summary: "unused" }],
+            degraded: true,
+          });
+        },
+      } as never,
+      {
+        get() {
+          return Promise.resolve(null);
+        },
+        getMeta() {
+          return Promise.resolve({ lastQuery: "alpha" });
+        },
+        rememberRefreshQuery() {
+          return Promise.resolve();
+        },
+        set(
+          _groupId: string,
+          entry: {
+            query: string;
+            refreshedAt: number;
+            nodes: Array<{ uuid: string; name: string; summary: string }>;
+            episodeSummaries?: string[];
+            nodeRefs: string[];
+          },
+        ) {
+          cacheSets.push(entry);
+          return Promise.resolve();
+        },
+      } as never,
+      {
+        drainGroup: () => Promise.resolve({ status: "idle" as const }),
+      } as never,
+    );
+
+    service.scheduleCacheRefresh("group-1", "alpha");
+    await flushMicrotasks();
+    await service.dispose();
+
+    assertEquals(cacheSets.length, 1);
+    assertEquals(cacheSets[0]?.query, "alpha");
+    assertEquals(cacheSets[0]?.nodes, []);
+    assertEquals(cacheSets[0]?.episodeSummaries, [
+      "Source → alpha: fact:alpha",
+    ]);
+    assertEquals(cacheSets[0]?.nodeRefs, []);
+  });
+
   it("does not start a second drain while a slow drain is still in flight", async () => {
     const timers = createFakeTimers();
     const drainDeferred = deferred<{ status: "idle" }>();

src/services/graphiti-async.ts

@@ -186,7 +186,6 @@ export class GraphitiAsyncService {
           maxNodes: 12,
         }),
       ]);
-      if (result.degraded) return;
       if (this.stopped) return;
 
       const [meta, current] = await Promise.all([
@@ -202,7 +201,7 @@ export class GraphitiAsyncService {
       }
       if (this.stopped) return;
 
-      const { nodes } = result;
+      const nodes = result.degraded ? [] : result.nodes;
       await this.cache.set(groupId, {
         query: normalized,
         refreshedAt: Date.now(),

src/services/hot-tier-slice.test.ts

@@ -1450,7 +1450,7 @@ describe("hot-tier vertical slice", () => {
     }
   });
 
-  it("preserves warm cache entry when refresh degrades during node search", async () => {
+  it("stores fact-only refreshes with empty nodes when node search degrades", async () => {
     const redis = new RedisClient({
       endpoint: "redis://unused",
       runtimeFactory: () => new ReconnectingRedisRuntime({ available: true }),
@@ -1478,7 +1478,13 @@ describe("hot-tier vertical slice", () => {
           return Promise.resolve([]);
         },
         searchMemoryFacts() {
-          return Promise.resolve([]);
+          return Promise.resolve([
+            {
+              fact: "fact:outage",
+              source_node: { name: "WarmNode" },
+              target_node: { name: "OutageTopic" },
+            },
+          ]);
         },
         searchNodesWithStatus() {
           return Promise.resolve({ nodes: [], degraded: true });
@@ -1499,16 +1505,12 @@ describe("hot-tier vertical slice", () => {
     const cached = await redisCache.get("group-1");
     const meta = await redisCache.getMeta("group-1");
 
-    assertEquals(cached, {
-      query: "warm query",
-      refreshedAt: 111,
-      nodes: [{
-        uuid: "warm-node",
-        name: "WarmNode",
-        summary: "Existing warm cache entry",
-      }],
-      nodeRefs: ["warm-node"],
-    });
+    assertEquals(cached?.query, "outage query");
+    assertEquals(cached?.nodes, []);
+    assertEquals(cached?.nodeRefs, []);
+    assertEquals(cached?.episodeSummaries, [
+      "WarmNode → OutageTopic: fact:outage",
+    ]);
     assertEquals(meta?.lastQuery, "outage query");
   });