fix(ai): recover orphaned .draining file from SIGKILL edge case (#10153)

Gemini 3.1 Pro's Depth Floor §7.1 review on PR #10171 surfaced a legitimate
edge case: if the Node process is terminated via SIGKILL (OOM, hardware
failure) between `rename(queuePath, drainingPath)` and the try/catch restore,
the `.draining` file is orphaned and those queued edges wait for manual
recovery. The catch-block restore doesn't execute under signal-based
termination.

Fix: `LazyEdgeDrainer.recoverOrphanedDraining(queuePath, drainingPath)`
runs at the start of every `drainQueue` invocation. If a `.draining` file
exists (from a prior crashed run), its content is merged into the live
queue and the orphan deleted. Idempotent — subsequent recovery-on-boot
finds no orphan and no-ops.

Stats shape extended with `orphanRecovered: Boolean` so operators can
observe recovery events via the drain-cycle logs.

Test coverage: +1 Playwright test in LazyEdgeDrainer.spec covering the
orphan-recovery path (synthesizes a .draining file + fresh queue, asserts
both drain in a single run and the orphan is cleaned up). 8/8 green.

Author: tobiu

ai/daemons/services/LazyEdgeDrainer.mjs

@@ -83,13 +83,20 @@ class LazyEdgeDrainer extends Base {
         const
             queuePath    = filePath || aiConfig.lazyEdgesQueuePath,
             drainingPath = queuePath + '.draining',
-            stats        = {totalLines: 0, processed: 0, succeeded: 0, failed: 0, skippedMalformed: 0, queueRotated: false};
+            stats        = {totalLines: 0, processed: 0, succeeded: 0, failed: 0, skippedMalformed: 0, queueRotated: false, orphanRecovered: false};
 
         if (!queuePath) {
             logger.warn('[LazyEdgeDrainer] No queue path configured (aiConfig.lazyEdgesQueuePath); nothing to drain.');
             return stats;
         }
 
+        // Recovery-on-boot for orphaned `.draining` files — addresses the SIGKILL edge case
+        // where a prior drain crashed between `rename` and the `try`/`catch` restore path
+        // (the catch only runs on thrown exceptions; `SIGKILL` / `SIGSTOP` / process exit
+        // bypass it). Merging orphan content back into the live queue is idempotent and
+        // preserves the "failures retained for retry" contract.
+        stats.orphanRecovered = await this.recoverOrphanedDraining(queuePath, drainingPath);
+
         try {
             if (dryRun) {
                 // Dry-run: read from the live queue without rotating, so concurrent producers
@@ -142,6 +149,51 @@ class LazyEdgeDrainer extends Base {
         return stats;
     }
 
+    /**
+     * Recovers an orphaned `.draining` file from a prior drain cycle that crashed between
+     * `rename(queuePath, drainingPath)` and the catch-block restore. Such orphans occur under
+     * process termination signals that bypass `try`/`catch` (`SIGKILL`, OOM kills, hardware
+     * failure). Without recovery, the orphaned edges would remain on disk but never be
+     * retried until manual intervention.
+     *
+     * Resolution: merge the orphan's content into the live queue (prepended so orphaned
+     * edges are processed first, preserving FIFO semantics for the entries that have been
+     * waiting longest), then delete the orphan. The merge is idempotent — a subsequent
+     * recovery-on-boot finds no orphan and no-ops.
+     *
+     * @param {String} queuePath    The live queue path
+     * @param {String} drainingPath The `.draining` orphan path
+     * @returns {Promise<Boolean>} `true` if an orphan was recovered; `false` if none existed.
+     * @protected
+     */
+    async recoverOrphanedDraining(queuePath, drainingPath) {
+        if (!fs.existsSync(drainingPath)) {
+            return false;
+        }
+
+        try {
+            const orphanContent = await fs.promises.readFile(drainingPath, 'utf8');
+
+            if (orphanContent.length > 0) {
+                const existingContent = fs.existsSync(queuePath)
+                    ? await fs.promises.readFile(queuePath, 'utf8')
+                    : '';
+
+                // Normalize trailing newline on the orphan so the concat doesn't produce
+                // adjacent JSON objects on a single line.
+                const merged = orphanContent.endsWith('\n') ? orphanContent + existingContent : orphanContent + '\n' + existingContent;
+                await fs.promises.writeFile(queuePath, merged, 'utf8');
+            }
+
+            await fs.promises.unlink(drainingPath);
+            logger.warn(`[LazyEdgeDrainer] Recovered orphaned ${drainingPath} from prior run and merged into live queue.`);
+            return true;
+        } catch (e) {
+            logger.error(`[LazyEdgeDrainer] Failed to recover orphaned ${drainingPath}:`, e);
+            return false;
+        }
+    }
+
     /**
      * Processes the raw JSONL content — one edge per line. Parses, validates, attempts
      * resolution via `GraphService.linkNodesAsync`, and accumulates stats + failures.

test/playwright/unit/ai/daemons/services/LazyEdgeDrainer.spec.mjs

@@ -212,6 +212,36 @@ test.describe('Neo.ai.daemons.services.LazyEdgeDrainer', () => {
         expect(remaining).toContain('memory:dryrun-m');
     });
 
+    test('drainQueue recovers orphaned .draining file from a prior-run SIGKILL', async () => {
+        // Simulate a prior drain that was SIGKILL'd between rename and completion: the
+        // `.draining` file exists with queued edges but the live queue may or may not have
+        // fresh appends from a concurrent producer.
+        const drainingPath = queuePath + '.draining';
+        const orphanedEdge = {source: 'CONCEPT:orphan', target: 'memory:orphan', relationship: 'MENTIONED_IN'};
+        const freshEdge    = {source: 'CONCEPT:fresh',  target: 'memory:fresh',  relationship: 'DISCUSSED_IN'};
+
+        fs.writeFileSync(drainingPath, JSON.stringify(orphanedEdge) + '\n');
+        fs.writeFileSync(queuePath,    JSON.stringify(freshEdge)   + '\n');
+
+        const processedTargets = [];
+        GraphService.linkNodesAsync = async (source, target) => {
+            processedTargets.push(target);
+            return true;
+        };
+
+        const stats = await LazyEdgeDrainer.drainQueue({filePath: queuePath});
+
+        expect(stats.orphanRecovered).toBe(true);
+        // Both the orphaned and the fresh edge should have been drained in this single run.
+        expect(stats.succeeded).toBe(2);
+        expect(processedTargets).toContain('memory:orphan');
+        expect(processedTargets).toContain('memory:fresh');
+        // Orphan file removed after successful recovery.
+        expect(fs.existsSync(drainingPath)).toBe(false);
+        // Live queue fully drained (no failures to retain).
+        expect(fs.existsSync(queuePath)).toBe(false);
+    });
+
     test('drainQueue handles thrown exceptions from linkNodesAsync as failures', async () => {
         fs.writeFileSync(queuePath, JSON.stringify({source: 'a', target: 'memory:throw', relationship: 'RELATES_TO'}) + '\n');