Preserve sync_state revision when poller bumps last_seen_at

haiku_rag_slim/haiku/rag/ingester/queue/repository.py

@@ -392,7 +392,9 @@ class SyncStateRepo:
         ingested: bool = False,
     ) -> None:
         """Insert-or-update the sync_state row. `ingested=True` stamps
-        last_ingested_at; otherwise only last_seen_at is bumped."""
+        last_ingested_at; otherwise only last_seen_at is bumped.
+        `revision=None` and `content_hash=None` leave any existing values
+        untouched."""
         now = _utcnow_iso()
         ingested_at = now if ingested else None
         async with self._lock:
@@ -403,8 +405,8 @@ class SyncStateRepo:
                 )
                 VALUES (?, ?, ?, ?, ?, ?)
                 ON CONFLICT(source_id, uri) DO UPDATE SET
-                    revision = excluded.revision,
-                    content_hash = excluded.content_hash,
+                    revision = COALESCE(excluded.revision, revision),
+                    content_hash = COALESCE(excluded.content_hash, content_hash),
                     last_seen_at = excluded.last_seen_at,
                     last_ingested_at = COALESCE(excluded.last_ingested_at, last_ingested_at)
                 """,

tests/ingester/test_pollers.py

@@ -202,6 +202,33 @@ async def test_skipped_sweep_records_pending_work_reason(fs_config, jobs, sync):
     assert poller.last_skip_reason is None
 
 
+@pytest.mark.asyncio
+async def test_dead_job_does_not_clear_sync_state_revision(fs_config, jobs, sync):
+    """When a job dies, the URI's previously-ingested revision must remain
+    in sync_state so subsequent sweeps still see the URI as known."""
+    await sync.upsert("src", "file:///a.md", revision="r1", content_hash="h1")
+
+    changed = _event("file:///a.md", revision="r2")
+    source = _StubSource("src", [[changed], [changed]])
+    poller = _periodic(source, fs_config, jobs, sync)
+
+    await poller._sweep_once()
+    queued = await jobs.list_jobs()
+    assert len(queued) == 1
+    assert queued[0].revision == "r2"
+
+    claimed = await jobs.claim_next("w")
+    assert claimed is not None
+    await jobs.mark_dead(claimed.id, "transient blew up")
+
+    row = await sync.get_row("src", "file:///a.md")
+    assert row is not None
+    assert row.revision == "r1"
+
+    await poller._sweep_once()
+    assert await sync.get_snapshot("src") == {"file:///a.md": "r1"}
+
+
 @pytest.mark.asyncio
 async def test_sweep_resumes_after_queue_drains(fs_config, jobs, sync):
     """Once the queue clears (success, dead, or cancel), sweeps resume."""

tests/ingester/test_queue.py

@@ -572,3 +572,27 @@ async def test_sync_state_snapshot_scoped_per_source(sync):
     await sync.upsert("s2", "u", revision="def", content_hash="m")
     assert await sync.get_snapshot("s1") == {"u": "abc"}
     assert await sync.get_snapshot("s2") == {"u": "def"}
+
+
+@pytest.mark.asyncio
+async def test_sync_state_upsert_preserves_revision_when_none(sync):
+    """upsert(revision=None) leaves an existing revision in place."""
+    await sync.upsert("s", "u", revision="v1", content_hash="hash-v1")
+    await sync.upsert("s", "u", revision=None, content_hash=None)
+    row = await sync.get_row("s", "u")
+    assert row is not None
+    assert row.revision == "v1"
+    assert row.content_hash == "hash-v1"
+    assert await sync.get_snapshot("s") == {"u": "v1"}
+
+
+@pytest.mark.asyncio
+async def test_sync_state_upsert_replaces_revision_when_provided(sync):
+    """upsert with a non-None revision overwrites the existing one."""
+    await sync.upsert("s", "u", revision="v1", content_hash="hash-v1")
+    await sync.upsert("s", "u", revision="v2", content_hash="hash-v2", ingested=True)
+    row = await sync.get_row("s", "u")
+    assert row is not None
+    assert row.revision == "v2"
+    assert row.content_hash == "hash-v2"
+    assert row.last_ingested_at is not None