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

CHANGELOG.md

@@ -108,6 +108,11 @@ All notable changes to Engraphis are documented here. Format loosely follows
 
 ### Fixed
 
+- Import previews now page the source manifest exactly like execution, so vaults whose manifest
+  outgrew one list page (10k identities) no longer show manifest-only files as silently absent
+  from the preview plan; beyond-boundary rows are reported as `missing` instead of dropped.
+  Manifest pages now use one read snapshot and de-duplicate identities that move across a
+  cursor while a concurrent import updates their path.
 - Importing more than 1,000 files through the dashboard no longer fails with "Internal Server
   Error": wizard upload routes parse multipart forms under the advertised 1,500-file ceiling
   instead of Starlette's hidden 1,000-part parser default, oversized batches return a clear 413,

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,22 +3841,66 @@ 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,
-    ) -> int:
+        missing_items: Iterable[Any] = (),
+    ) -> list[str]:
+        """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.
+        Returns the source_keys actually marked, so callers can finalize job history
+        and reports against reality instead of the plan.
+        """
         if self._source_vault_row(vault_id) is None:
-            return 0
+            return []
+        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),
                 )
-            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<?) "
-                "AND state NOT IN ('missing','conflict')",
-                (now_ts(), vault_id, float(seen_before)),
-            ).rowcount)
+            if planned:
+                marked: list[str] = []
+                stamp = now_ts()
+                for key, seen_at, seen_job in planned:
+                    if 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:
+                        marked.append(key)
+                return marked
+            marked_rows = [
+                str(row["source_key"]) for row in self.conn.execute(
+                    "SELECT source_key FROM source_imports WHERE vault_id=? "
+                    "AND (last_seen_at IS NULL OR last_seen_at<?) "
+                    "AND state NOT IN ('missing','conflict')",
+                    (vault_id, float(seen_before)),
+                ).fetchall()
+            ]
+            if not marked_rows:
+                return []
+            for start in range(0, len(marked_rows), 900):
+                chunk = marked_rows[start:start + 900]
+                placeholders = ",".join("?" for _ in chunk)
+                self.conn.execute(
+                    f"UPDATE source_imports SET state='missing', missing_at=? "
+                    f"WHERE vault_id=? AND source_key IN ({placeholders})",
+                    (now_ts(), vault_id, *chunk),
+                )
+            return marked_rows
 
     def get_source_import(self, import_id: str) -> Optional[dict]:
         row = self.conn.execute("SELECT * FROM source_imports WHERE id=?", (import_id,)).fetchone()