From f61c5fd5fd017b53a9ce8e6629a0b912db01555e Mon Sep 17 00:00:00 2001
From: Willow Lopez <100782273+Oxygen56@users.noreply.github.com>
Date: Mon, 15 Jun 2026 01:04:52 +0800
Subject: [PATCH] fix(pipeline): return cursor on zero-record L2 extraction for
 incremental processing (#130)

Fixes #98

Return { latestCursor: cursor } instead of void when VectorStore or
JSONL pipeline returns zero records. Add cold-start guard in pipeline
manager to initialize last_extraction_updated_time on first run.

Previously the L2 runner performed a full table scan on every invocation
because the cursor never advanced past the initial state.

Signed-off-by: Oxygen56 <1391083091@qq.com>
---
 src/utils/pipeline-factory.ts | 4 ++--
 src/utils/pipeline-manager.ts | 5 +++++
 2 files changed, 7 insertions(+), 2 deletions(-)

diff --git a/src/utils/pipeline-factory.ts b/src/utils/pipeline-factory.ts
index b0b3f1c..ce87801 100644
--- a/src/utils/pipeline-factory.ts
+++ b/src/utils/pipeline-factory.ts
@@ -463,7 +463,7 @@ export function createL2Runner(opts: {
         logger.debug?.(
           `${TAG} [L2] No new L1 records since cursor (session=${sessionKey}, updatedAfter=${cursor ?? "(full)"}), skipping scene extraction`,
         );
-        return { skipped: true };
+        return { skipped: true, latestCursor: cursor || undefined };
       }
 
       logger.debug?.(
@@ -494,7 +494,7 @@ export function createL2Runner(opts: {
 
       if (sessionRecords.length === 0) {
         logger.debug?.(`${TAG} [L2] No new L1 records found (JSONL fallback, session=${sessionKey}), skipping scene extraction`);
-        return;
+        return { latestCursor: cursor || undefined };
       }
 
       records = sessionRecords.map((r) => ({
diff --git a/src/utils/pipeline-manager.ts b/src/utils/pipeline-manager.ts
index 92ce95b..bccf293 100644
--- a/src/utils/pipeline-manager.ts
+++ b/src/utils/pipeline-manager.ts
@@ -923,6 +923,11 @@ export class MemoryPipelineManager {
     // Advance cursor using the record timestamp returned by the runner
     if (result?.latestCursor) {
       state.last_extraction_updated_time = result.latestCursor;
+    } else if (!state.last_extraction_updated_time) {
+      // Cold-start guard: if runner returned void (e.g. extraction failure) and
+      // last_extraction_updated_time is still empty, initialize it to now so
+      // the next L2 run doesn't do a full table scan.
+      state.last_extraction_updated_time = new Date().toISOString();
     }
 
     await this.persistStates();