fix(import): fail closed on truncated previews and denials

Author: Coding-Dev-Tools

engraphis/obsidian_import.py

@@ -277,7 +277,7 @@ def preview(
         attachment_manifest: Optional[list[dict]] = None,
     ) -> dict:
         policy = self._policy(on_conflict)
-        vault, items = self._preview_manifest(
+        vault, items, manifest_complete = self._preview_manifest(
             scan, workspace_id=workspace_id, repo_id=repo_id,
             session_id=session_id, vault_id=vault_id, manifest=manifest,
             scope=scope, memory_type=memory_type, strict_root=strict_root,
@@ -286,11 +286,18 @@ def preview(
         plans, missing = self._plan(
             scan, identity, items, inspect_memories=manifest is None,
         )
+        # A bounded manifest is enough to plan visible notes, but not enough to claim
+        # that an unseen historical row was deleted. Match confirmed execution by
+        # surfacing those rows as deferred rather than actionable ``missing`` items.
+        pending_missing = missing if not manifest_complete else []
+        if pending_missing:
+            missing = []
         return self._report(
             plans, missing, scan, state="preview", vault_id=(vault or {}).get("id"),
             workspace_id=workspace_id, repo_id=repo_id, session_id=session_id,
             scope=scope, memory_type=memory_type, policy=policy,
             vault_label=vault_label, attachment_manifest=attachment_manifest,
+            pending_missing=pending_missing, manifest_complete=manifest_complete,
         )
 
     def import_scan(
@@ -426,6 +433,7 @@ def import_scan(
             workspace_id=workspace_id, repo_id=repo_id, session_id=session_id,
             scope=scope, memory_type=memory_type, policy=policy,
             vault_label=vault_label, attachment_manifest=attachment_manifest,
+            manifest_complete=manifest_complete,
         )
         if terminal_state == "completed" and report["counts"].get("conflict", 0):
             terminal_state = "partial"
@@ -476,7 +484,7 @@ def _preview_manifest(
         repo_id: Optional[str], session_id: Optional[str], vault_id: Optional[str],
         scope: Scope, memory_type: MemoryType, manifest: Optional[dict],
         strict_root: bool,
-    ) -> tuple[Optional[dict], list[dict]]:
+    ) -> tuple[Optional[dict], list[dict], bool]:
         vaults = (
             list((manifest or {}).get("vaults") or [])
             if manifest is not None else self.store.list_source_vaults(kind=self.SOURCE_KIND)
@@ -486,6 +494,7 @@ def _preview_manifest(
             if manifest is not None else []
         )
         vault: Optional[dict] = None
+        manifest_complete = True
         if vault_id:
             vault = (
                 self.store.get_source_vault(vault_id)
@@ -513,12 +522,12 @@ def _preview_manifest(
             if manifest is None:
                 # Page the manifest like import_scan does so previews on manifests
                 # larger than one list page plan against the full item set.
-                items, _ = self._all_source_items(vault_id=str(vault["id"]))
+                items, manifest_complete = self._all_source_items(vault_id=str(vault["id"]))
             else:
                 items = [row for row in items if row.get("vault_id") == vault.get("id")]
         else:
             items = []
-        return vault, items
+        return vault, items, manifest_complete
 
     def _resolve_or_register_vault(
         self, scan: _ImportScan, *, workspace_id: str,
@@ -1332,6 +1341,8 @@ def _report(
         repo_id: Optional[str], session_id: Optional[str], scope: Scope,
         memory_type: MemoryType, policy: str, vault_label: str,
         attachment_manifest: Optional[list[dict]],
+        pending_missing: Optional[list[dict]] = None,
+        manifest_complete: bool = True,
     ) -> dict:
         files = [self._preview_row(plan) for plan in plans]
         files.extend({
@@ -1346,11 +1357,16 @@ def _report(
             "relative_path": str(item.get("relative_path") or ""), "status": "missing",
             "action": "missing", "reason": "source_not_seen", "warnings": [],
         } for item in missing)
+        files.extend({
+            "relative_path": str(item.get("relative_path") or ""), "status": "pending",
+            "action": "pending", "reason": "missing_check_deferred", "warnings": [],
+        } for item in pending_missing or [])
         return self._report_payload(
             files, scan, state=state, vault_id=vault_id,
             workspace_id=workspace_id, repo_id=repo_id, session_id=session_id,
             scope=scope, memory_type=memory_type, policy=policy,
             vault_label=vault_label, attachment_manifest=attachment_manifest,
+            manifest_complete=manifest_complete,
         )
 
     def _final_report(
@@ -1360,6 +1376,7 @@ def _final_report(
         import_id: str, workspace_id: str, repo_id: Optional[str],
         session_id: Optional[str], scope: Scope, memory_type: MemoryType,
         policy: str, vault_label: str, attachment_manifest: Optional[list[dict]],
+        manifest_complete: bool = True,
     ) -> dict:
         files = list(outcomes)
         processed_paths = {
@@ -1396,6 +1413,7 @@ def _final_report(
             workspace_id=workspace_id, repo_id=repo_id, session_id=session_id,
             scope=scope, memory_type=memory_type, policy=policy,
             vault_label=vault_label, attachment_manifest=attachment_manifest,
+            manifest_complete=manifest_complete,
         )
         report.update({"job_id": job_id, "import_id": import_id})
         return report
@@ -1423,6 +1441,7 @@ def _report_payload(
         vault_id: Optional[str], workspace_id: Optional[str], repo_id: Optional[str],
         session_id: Optional[str], scope: Scope, memory_type: MemoryType,
         policy: str, vault_label: str, attachment_manifest: Optional[list[dict]],
+        manifest_complete: bool = True,
     ) -> dict:
         counts: dict[str, int] = {}
         for row in files:
@@ -1440,6 +1459,7 @@ def _report_payload(
             formats[name] = formats.get(name, 0) + 1
         return {
             "state": state, "status": state, "vault_id": vault_id,
+            "manifest_complete": bool(manifest_complete),
             "vault_label": str(vault_label or "")[:200],
             "source_id": vault_id,
             "source_label": str(vault_label or "")[:200],

engraphis/routes/v2_api.py

@@ -3126,10 +3126,16 @@ def _mark_authoritative_denial() -> None:
     global _authoritative_denial_at, _denied_state_digests
     with _ENTITLEMENT_REFRESH_LOCK:
         _authoritative_denial_at = time.time()
+        # Publish the fail-closed state before probing either persisted source. Those
+        # reads can block on a slow state directory; a concurrent license/bootstrap
+        # request must not keep serving the pre-denial grants while they are in flight.
+        # Keep the baseline empty until both probes complete so an active-looking record
+        # cannot be mistaken for a post-denial reconnect during the capture window.
+        _AUTHORITATIVE_DENIAL_PENDING.set()
+        _denied_state_digests = {}
         _denied_state_digests = {
             source: _persisted_state_digest(source) for source in ("session", "cloud")
         }
-        _AUTHORITATIVE_DENIAL_PENDING.set()
 
 
 def _clear_superseded_denial(known_source: str, observed_digest: Optional[str]) -> bool:

tests/test_hosted_plan_resolution.py

@@ -1134,6 +1134,32 @@ def _blocked_write():
     assert not worker.is_alive()
 
 
+def test_denial_guard_publishes_before_blocked_digest_probe(monkeypatch) -> None:
+    """Readers fail closed while the pre-persistence denial baseline is captured."""
+
+    _connect(monkeypatch, pinned_token=False)
+    entered = threading.Event()
+    release = threading.Event()
+
+    def _blocked_digest(_source):
+        entered.set()
+        assert v2_api._AUTHORITATIVE_DENIAL_PENDING.is_set()
+        assert v2_api._denied_state_digests == {}
+        assert release.wait(timeout=5.0)
+        return "before-denial"
+
+    monkeypatch.setattr(v2_api, "_persisted_state_digest", _blocked_digest)
+    worker = threading.Thread(target=v2_api._mark_authoritative_denial)
+    worker.start()
+    assert entered.wait(timeout=5.0)
+    try:
+        assert v2_api._AUTHORITATIVE_DENIAL_PENDING.is_set()
+    finally:
+        release.set()
+        worker.join(timeout=5.0)
+    assert not worker.is_alive()
+
+
 def test_a_transport_failure_is_not_mistaken_for_a_billing_denial(monkeypatch) -> None:
     """Only an authoritative 401/402/403 clears access; an outage must not."""
 

tests/test_obsidian_service.py

@@ -7,7 +7,7 @@
 import pytest
 
 from engraphis.service import MemoryService, ValidationError
-from engraphis.obsidian_import import scan_obsidian_upload
+from engraphis.obsidian_import import ObsidianImporter, scan_obsidian_upload
 
 
 _TERMINAL_STATES = {"completed", "partial", "failed", "cancelled"}
@@ -90,6 +90,40 @@ def test_preview_pages_the_full_manifest_like_execution():
         service.close()
 
 
+def test_preview_defers_missing_rows_when_manifest_is_truncated(monkeypatch):
+    service = _service()
+    try:
+        started, imported = _import(service, [("A.md", b"# A\n")])
+        assert imported["state"] == "completed"
+        vault_id = started["vault_id"]
+        service.store.upsert_source_import_item(
+            vault_id=vault_id,
+            source_key=hashlib.sha256(b"gone").hexdigest(),
+            relative_path="gone.md",
+        )
+        items = service.store.list_source_import_items(vault_id=vault_id)
+
+        def _truncated(_self, *, vault_id, states=None):
+            del vault_id, states
+            return items, False
+
+        monkeypatch.setattr(ObsidianImporter, "_all_source_items", _truncated)
+        preview = service.preview_obsidian_upload(
+            files=[("A.md", b"# A\n")], attachment_manifest=[],
+            workspace="alpha", vault_label="Team notes", vault_id=vault_id,
+        )
+
+        statuses = {
+            row["relative_path"]: row["status"] for row in preview["files"]
+        }
+        assert preview["manifest_complete"] is False
+        assert preview["counts"].get("missing", 0) == 0
+        assert preview["counts"]["pending"] == 1
+        assert statuses["gone.md"] == "pending"
+    finally:
+        service.close()
+
+
 def test_preview_is_write_free_and_service_enforces_confirmation_and_upload_guards(monkeypatch):
     service = _service()
     try: