fix: deterministic source-import missing detection and denial-guard supersession

engraphis/cloud_session.py

@@ -531,6 +531,66 @@ def saved_entitlement() -> dict:
         return {}
 
 
+def saved_entitlement_snapshot() -> tuple[dict, Optional[str]]:
+    """Read the session once; return its entitlement plus a digest of those exact bytes.
+
+    Binding the parse to the bytes it came from lets a caller prove where an answer
+    predates a denial without re-reading: a license read that parsed the pre-denial
+    session must never mistake the denial-persistence write landing mid-read for a
+    superseding reconnect. ``None`` means "could not determine" (unreadable state);
+    ``""`` means the file is absent.
+    """
+
+    try:
+        raw = read_private_text(
+            _session_path(), max_bytes=64 * 1024, allow_missing=True
+        )
+    except Exception:  # noqa: BLE001 — an unreadable session is simply "nothing known"
+        return {}, None
+    if not raw:
+        return {}, ""
+    digest = hashlib.sha256(raw.encode("utf-8", "surrogatepass")).hexdigest()
+    try:
+        value = json.loads(raw)
+    except (ValueError, RecursionError):
+        return {}, digest
+    if not isinstance(value, dict):
+        return {}, digest
+    declared = _declared_entitlement(value)
+    if not declared:
+        return {}, digest
+    try:
+        checked_at = float(value.get("entitlement_checked_at") or 0.0)
+    except (TypeError, ValueError, OverflowError):
+        checked_at = 0.0
+    declared["entitlement_checked_at"] = checked_at
+    declared["organization_id"] = str(value.get("organization_id") or "")
+    return declared, digest
+
+
+def saved_session_digest() -> Optional[str]:
+    """Return a ``sha256`` digest over the raw saved session bytes, or ``""`` if absent.
+
+    ``None`` means "could not determine" (unreadable or unsafe state file), which callers
+    must treat as "no evidence of change". This lets a caller detect that the session was
+    *rewritten* — a genuine reconnect always rotates the refresh credential, so the bytes
+    differ — without parsing the record or exposing any credential material. Wall-clock
+    timestamps cannot serve this role: two writes inside one coarse clock tick stamp equal
+    ``entitlement_checked_at`` values, so only content distinguishes a post-denial
+    reconnect from a pre-denial record.
+    """
+
+    try:
+        raw = read_private_text(
+            _session_path(), max_bytes=64 * 1024, allow_missing=True
+        )
+    except Exception:  # noqa: BLE001 — unreadable state must not crash a digest probe
+        return None
+    if raw is None:
+        return ""
+    return hashlib.sha256(raw.encode("utf-8", "surrogatepass")).hexdigest()
+
+
 def record_billing_denial() -> bool:
     """Mark the saved entitlement inactive after an authoritative billing denial.
 

engraphis/core/store.py

@@ -3733,17 +3733,28 @@ def get_source_import_item(self, *, vault_id: str, source_key: str) -> Optional[
         return dict(row) if row is not None else None
 
     def list_source_import_items(self, *, vault_id: str, states: Optional[list[str]] = None,
-                                 limit: int = 10_000) -> list[dict]:
+                                 limit: int = 10_000, after_path: str = "",
+                                 after_id: str = "") -> list[dict]:
+        """Page the manifest by ``(relative_path, id)`` cursor, not OFFSET.
+
+        OFFSET is applied to a live ``ORDER BY`` result: a concurrent rename or insert
+        shifts unread rows across the page boundary and they are silently skipped while
+        the pager believes it saw everything. A keyset cursor is immune — every row at
+        or after the cursor is returned exactly once regardless of concurrent writes.
+        """
         if self._source_vault_row(vault_id) is None:
             return []
         params: list[Any] = [vault_id]
         sql = "SELECT * FROM source_imports WHERE vault_id=?"
+        if after_path or after_id:
+            sql += " AND (relative_path>? OR (relative_path=? AND id>?))"
+            params.extend([str(after_path), str(after_path), str(after_id)])
         if states is not None:
             if not states:
                 return []
             sql += " AND state IN (" + ",".join("?" for _ in states) + ")"
             params.extend(str(state) for state in states)
-        sql += " ORDER BY relative_path LIMIT ?"
+        sql += " ORDER BY relative_path, id LIMIT ?"
         params.append(max(1, min(100_000, int(limit))))
         return [dict(row) for row in self.conn.execute(sql, params).fetchall()]
 
@@ -3830,16 +3841,44 @@ def rename_source_import_item(self, *, vault_id: str, source_key: str,
     def mark_source_import_items_missing(
         self, *, vault_id: str, seen_before: float,
         preserve_paths: Iterable[str] = (), commit: bool = True,
+        missing_items: Iterable[Any] = (),
     ) -> int:
+        """Mark planned rows missing, or fall back to the timestamp heuristic.
+
+        With ``missing_items`` (the planner's manifest rows), each key is updated
+        only while the row still matches its planned generation — ``last_seen_at``,
+        ``last_seen_job_id``, and a live state. A concurrent import that refreshed
+        or re-upserted the row after this run planned it therefore keeps its newer
+        state instead of being clobbered back to ``missing``, and per-key updates
+        stay clear of SQLite host-parameter limits no matter how many rows died.
+        """
         if self._source_vault_row(vault_id) is None:
             return 0
+        planned = [
+            (str(item["source_key"]), item.get("last_seen_at"),
+             item.get("last_seen_job_id"))
+            for item in missing_items if item.get("source_key")
+        ]
         with self._write_operation("source_missing", commit=commit):
             for relative_path in {str(path) for path in preserve_paths if str(path)}:
                 self.conn.execute(
                     "UPDATE source_imports SET last_seen_at=? WHERE vault_id=? "
                     "AND relative_path=? AND state NOT IN ('missing','conflict')",
                     (float(seen_before), vault_id, relative_path),
                 )
+            if planned:
+                updated = 0
+                stamp = now_ts()
+                for key, seen_at, seen_job in planned:
+                    updated += int(self.conn.execute(
+                        "UPDATE source_imports SET state='missing', missing_at=? "
+                        "WHERE vault_id=? AND source_key=? "
+                        "AND (last_seen_at IS NULL OR last_seen_at=?) "
+                        "AND (last_seen_job_id IS NULL OR last_seen_job_id=?) "
+                        "AND state NOT IN ('missing','conflict')",
+                        (stamp, vault_id, key, seen_at, seen_job),
+                    ).rowcount)
+                return updated
             return int(self.conn.execute(
                 "UPDATE source_imports SET state='missing', missing_at=? WHERE vault_id=? "
                 "AND (last_seen_at IS NULL OR last_seen_at<?) "

engraphis/obsidian_import.py

@@ -317,7 +317,7 @@ def import_scan(
         run_started = time.time()
         job_id = str(prepared["job_id"])
         import_id = str(prepared["import_id"])
-        items = self.store.list_source_import_items(vault_id=vault_id)
+        items, manifest_complete = self._all_source_items(vault_id=vault_id)
         plans, missing = self._plan(scan, vault_id, items, inspect_memories=True)
         for plan in plans:
             self.store.record_source_import_job_item(
@@ -348,7 +348,9 @@ def import_scan(
         finalized_missing: list[dict] = []
         pending_missing = list(missing)
         unreadable_directories = self._unreadable_directories(scan)
-        can_finalize_missing = scan.complete and not unreadable_directories
+        can_finalize_missing = (
+            scan.complete and not unreadable_directories and manifest_complete
+        )
         terminal_state = "completed"
         try:
             for index, plan in enumerate(plans, 1):
@@ -368,6 +370,7 @@ def import_scan(
                 self.store.mark_source_import_items_missing(
                     vault_id=vault_id, seen_before=run_started,
                     preserve_paths=self._rejected_paths(scan),
+                    missing_items=missing,
                 )
                 for item in missing:
                     self.store.record_source_import_job_item(
@@ -387,8 +390,11 @@ def import_scan(
                 )
                 self._persist_link_warnings(job_id, link_warnings)
             outcomes.extend(link_warnings)
-            if scan.rejected or not scan.complete or unreadable_directories or any(
-                row["status"] in {"error", "conflict", "rejected"} for row in outcomes
+            if (
+                scan.rejected or not scan.complete or unreadable_directories
+                or not manifest_complete
+                or any(row["status"] in {"error", "conflict", "rejected"}
+                       for row in outcomes)
             ):
                 terminal_state = "partial"
         except (KeyboardInterrupt, ObsidianImportCancelled):
@@ -947,15 +953,54 @@ def _metadata(
             },
         }
 
+    def _all_source_items(self, *, vault_id: str,
+                          states: Optional[list[str]] = None) -> tuple[list[dict], bool]:
+        """Page the full manifest by keyset cursor so nothing is skipped or miscounted.
+
+        ``list_source_import_items`` caps each page (default 10k rows); a manifest that
+        outgrew one page through repeated deletions and additions must still be planned
+        and reconciled in full. The ``(relative_path, id)`` cursor is immune to the
+        OFFSET failure mode, where a concurrent rename shifts an unread row across the
+        page boundary so it is silently skipped while the pager believes it saw
+        everything; a row renamed below the already-read range degrades into the
+        content-hash rename detection instead. Returns the rows and whether the whole
+        manifest was read: the 200k-row bound is a memory cap, and one extra row is
+        probed past it so a manifest of exactly that size is not misreported as
+        truncated.
+        """
+        items: list[dict] = []
+        page_size = 10_000
+        cursor_path = cursor_id = ""
+        for _ in range(20):  # bounded: at most 200k manifest rows per import run
+            page = self.store.list_source_import_items(
+                vault_id=vault_id, states=states, limit=page_size,
+                after_path=cursor_path, after_id=cursor_id,
+            )
+            items.extend(page)
+            if len(page) < page_size:
+                return items, True
+            cursor_path = str(page[-1].get("relative_path") or "")
+            cursor_id = str(page[-1].get("id") or "")
+        extra = self.store.list_source_import_items(
+            vault_id=vault_id, states=states, limit=1,
+            after_path=cursor_path, after_id=cursor_id,
+        )
+        return items, not extra
+
     def _reconcile_links(
         self, scan: _ImportScan, *, vault_id: str, job_id: Optional[str] = None,
         cancel_check: Optional[Callable[[], bool]] = None,
     ) -> list[dict]:
         """Resolve derived links in bounded, cancellable, replay-safe batches."""
-        items = self.store.list_source_import_items(
+        items, manifest_complete = self._all_source_items(
             vault_id=vault_id,
             states=["imported", "unchanged", "renamed", "skipped", "missing"],
         )
+        if not manifest_complete:
+            # A manifest beyond the paging bound is an incomplete view of the source;
+            # retiring derived edges against it could kill links whose targets were
+            # simply invisible. The run is already headed to partial upstream.
+            return []
         memory_by_path = {
             str(item["relative_path"]): str(item["memory_id"])
             for item in items if item.get("memory_id")