diff --git a/CHANGELOG.md b/CHANGELOG.md index 13657664..ee9bcaa9 100644 --- a/CHANGELOG.md +++ b/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, diff --git a/engraphis/cloud_session.py b/engraphis/cloud_session.py index 5c2dd3e8..4bdb7f80 100644 --- a/engraphis/cloud_session.py +++ b/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. diff --git a/engraphis/core/store.py b/engraphis/core/store.py index d95075a9..57f44ed6 100644 --- a/engraphis/core/store.py +++ b/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,9 +3841,26 @@ 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( @@ -3840,12 +3868,39 @@ def mark_source_import_items_missing( "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 Optional[dict]: row = self.conn.execute("SELECT * FROM source_imports WHERE id=?", (import_id,)).fetchone() diff --git a/engraphis/obsidian_import.py b/engraphis/obsidian_import.py index a006ece7..30d6e859 100644 --- a/engraphis/obsidian_import.py +++ b/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( @@ -317,7 +324,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 +355,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): @@ -365,30 +374,54 @@ def import_scan( progress(dict(outcome)) self._check_cancel(job_id, cancel_check) if can_finalize_missing: - self.store.mark_source_import_items_missing( + # Set membership: O(1) average per lookup, replacing the previous + # quadratic list scan over up to 200k items — a complexity fix, + # not a timing-sensitive comparison. + marked_keys = set(self.store.mark_source_import_items_missing( vault_id=vault_id, seen_before=run_started, preserve_paths=self._rejected_paths(scan), - ) - for item in missing: + missing_items=missing, + )) + finalized = [ + item for item in missing + if str(item.get("source_key") or "") in marked_keys + ] + for item in finalized: self.store.record_source_import_job_item( job_id=job_id, source_id=item.get("id"), relative_path=str(item.get("relative_path") or "(missing)"), planned_action="missing", result_state="missing", ) - finalized_missing = missing + for item in missing: + if str(item.get("source_key") or "") in marked_keys: + continue + # The generation guard left this row live: a concurrent import + # refreshed it after this run planned it missing. The job history + # must not claim a live source was removed. + self.store.record_source_import_job_item( + job_id=job_id, source_id=item.get("id"), + relative_path=str(item.get("relative_path") or "(missing)"), + planned_action="missing", result_state="skipped", + ) + finalized_missing = finalized pending_missing = [] # Link reconciliation is safe only for a complete view of the source. # An incomplete scan must not retire a valid edge merely because its target # was hidden by a transient filesystem or scan-budget failure. link_warnings: list[dict] = [] + links_manifest_complete = True if can_finalize_missing: - link_warnings = self._reconcile_links( + link_warnings, links_manifest_complete = self._reconcile_links( scan, vault_id=vault_id, job_id=job_id, cancel_check=cancel_check, ) self._persist_link_warnings(job_id, link_warnings) + manifest_complete = manifest_complete and links_manifest_complete 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): @@ -402,6 +435,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" @@ -452,7 +486,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) @@ -462,6 +496,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) @@ -487,12 +522,14 @@ def _preview_manifest( scope=scope, memory_type=memory_type, strict_root=strict_root, ) if manifest is None: - items = self.store.list_source_import_items(vault_id=str(vault["id"])) + # Page the manifest like import_scan does so previews on manifests + # larger than one list page plan against the full item set. + 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, @@ -947,15 +984,85 @@ 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 avoids 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. + Store-backed reads hold one SQLite snapshot for all pages; identity de-duplication + is also defensive for injected readers that expose a row after a forward rename. + 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] = [] + positions: dict[str, int] = {} + page_size = 10_000 + cursor_path = cursor_id = "" + conn = getattr(self.store, "conn", None) + snapshot_started = bool( + conn is not None + and hasattr(conn, "transaction_owned_by_current_thread") + and not conn.transaction_owned_by_current_thread() + ) + try: + if snapshot_started: + conn.execute("BEGIN") + result: Optional[tuple[list[dict], bool]] = None + 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, + ) + for row in page: + identity = str(row.get("id") or row.get("source_key") or "") + position = positions.get(identity) if identity else None + if position is None: + if identity: + positions[identity] = len(items) + items.append(row) + else: + # A reader that cannot hold a snapshot may observe one source + # row again after a concurrent rename moves it forward. Keep the + # newest observation rather than planning that identity twice. + items[position] = row + if len(page) < page_size: + result = (items, True) + break + cursor_path = str(page[-1].get("relative_path") or "") + cursor_id = str(page[-1].get("id") or "") + if result is None: + extra = self.store.list_source_import_items( + vault_id=vault_id, states=states, limit=1, + after_path=cursor_path, after_id=cursor_id, + ) + result = (items, not extra) + if snapshot_started and conn.transaction_owned_by_current_thread(): + conn.commit() + return result + except BaseException: + if snapshot_started and conn.transaction_owned_by_current_thread(): + conn.rollback() + raise + def _reconcile_links( self, scan: _ImportScan, *, vault_id: str, job_id: Optional[str] = None, cancel_check: Optional[Callable[[], bool]] = None, - ) -> list[dict]: + ) -> tuple[list[dict], bool]: """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 [], False memory_by_path = { str(item["relative_path"]): str(item["memory_id"]) for item in items if item.get("memory_id") @@ -1123,7 +1230,7 @@ def retire_unsupported_links() -> None: self.store.conn.rollback() raise flush() - return warnings + return warnings, True def _persist_link_warnings(self, job_id: str, warnings: list[dict]) -> None: """Attach reconciliation warnings to the durable polling rows.""" @@ -1236,6 +1343,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({ @@ -1250,11 +1359,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( @@ -1264,6 +1378,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 = { @@ -1300,6 +1415,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 @@ -1327,6 +1443,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: @@ -1344,6 +1461,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], diff --git a/engraphis/routes/v2_api.py b/engraphis/routes/v2_api.py index f12c0ac0..2b5fae42 100644 --- a/engraphis/routes/v2_api.py +++ b/engraphis/routes/v2_api.py @@ -2674,6 +2674,7 @@ def entitled_features(plan: str) -> list: # state directory is temporarily unwritable. A newer active authoritative record clears it. _AUTHORITATIVE_DENIAL_PENDING = threading.Event() _authoritative_denial_at = 0.0 +_denied_state_digests: dict = {} #: Same opt-out vocabulary as ``ENGRAPHIS_UPDATE_CHECK`` (see engraphis/update_check.py). _FALSY_SETTINGS = {"0", "false", "no", "off", "disable", "disabled"} @@ -2873,27 +2874,27 @@ def _normalized_features(values: object, plan: str) -> list: return sorted(granted & allowed & set(_FEATURE_LABELS)) -def _session_entitlement() -> dict: - """Return the entitlement the control plane put on this client's own session. +def _session_entitlement_snapshot() -> tuple[dict, Optional[str]]: + """Read the session once; return its entitlement and the digest of those bytes. - Shaped exactly like ``_read_entitlement_cache`` so both persisted answers feed the - resolver identically and only their precedence differs. Reads state only — no network — - and never raises: this is on the ``/api/bootstrap`` boot path. + The digest travels with the parse: a caller comparing against the denial baseline + must judge the exact bytes ``known`` was parsed from, not whatever the file contains + by the time the comparison runs. """ try: from engraphis import cloud_session - reader = getattr(cloud_session, "saved_entitlement", None) - declared = reader() if reader is not None else None + reader = getattr(cloud_session, "saved_entitlement_snapshot", None) + declared, digest = reader() if reader is not None else ({}, None) if not isinstance(declared, dict) or not declared: - return {} + return {}, digest # A deployment pinned to ``ENGRAPHIS_CLOUD_ORGANIZATION_ID`` may be pointed at a # different organization than the saved session was registered for. Refuse to # relabel one customer's plan with another's, exactly as the entitlements read # refuses a mis-routed answer. pinned = os.environ.get("ENGRAPHIS_CLOUD_ORGANIZATION_ID", "").strip() if pinned and pinned != str(declared.get("organization_id") or ""): - return {} + return {}, digest plan = _normalized_plan(declared.get("plan")) active = bool(declared.get("cloud_access_active")) resolved = { @@ -2909,36 +2910,50 @@ def _session_entitlement() -> dict: "fetched_at": float(declared.get("entitlement_checked_at") or 0.0), } resolved.update(_trial_facts(declared)) - return resolved + return resolved, digest except Exception: # noqa: BLE001 - a badge must never break /bootstrap - return {} + return {}, None -def _read_entitlement_cache() -> dict: - """Return the last cached ``GET /v1/entitlements`` answer, or ``{}``. Never raises. +def _session_entitlement() -> dict: + """Return the entitlement the control plane put on this client's own session. - Secondary to ``_session_entitlement``: this file exists only for a control plane that - does not yet return the entitlement on registration and refresh. + Shaped exactly like ``_read_entitlement_cache`` so both persisted answers feed the + resolver identically and only their precedence differs. Reads state only — no network — + and never raises: this is on the ``/api/bootstrap`` boot path. + """ + + return _session_entitlement_snapshot()[0] + + +def _read_entitlement_cache_snapshot() -> tuple[dict, Optional[str]]: + """Read the cache once; return its entitlement and the digest of those bytes. + + Same binding rule as ``_session_entitlement_snapshot``: the supersession check must + judge the bytes ``known`` was parsed from, so the denial-persistence cache rewrite + landing mid-read cannot pose as a newer active answer. """ path = _entitlement_cache_path() if path is None: - return {} + return {}, "" + digest: Optional[str] try: from engraphis.private_state import read_private_text raw = read_private_text( path, max_bytes=_ENTITLEMENT_MAX_RESPONSE_BYTES, allow_missing=True ) except Exception: # noqa: BLE001 - an unreadable cache is just "nothing known yet" - return {} + return {}, None if not raw: - return {} + return {}, "" + digest = hashlib.sha256(raw.encode("utf-8", "surrogatepass")).hexdigest() try: value = json.loads(raw) except (ValueError, RecursionError): - return {} + return {}, digest if not isinstance(value, dict) or value.get("schema") != _ENTITLEMENT_CACHE_SCHEMA: - return {} + return {}, digest # Validate rather than coerce the plan: a corrupt value must be *discarded* so the # caller falls through to its own inference. Coercing it would quietly downgrade a # connected paying customer to the free local core on a damaged file. @@ -2946,7 +2961,7 @@ def _read_entitlement_cache() -> dict: if not isinstance(stored_plan, str) or stored_plan.strip().lower() not in ( "pro", "team", "local", "free" ): - return {} + return {}, digest try: fetched_at = float(value.get("fetched_at") or 0.0) except (TypeError, ValueError, OverflowError): @@ -2968,7 +2983,7 @@ def _read_entitlement_cache() -> dict: organization_id = str(value.get("organization_id") or "") current = _configured_organization_id() if not current or organization_id != current: - return {} + return {}, digest plan = _normalized_plan(stored_plan) # Mirror _session_entitlement: an inactive entitlement publishes no features. Today # hosted_plan_summary re-zeroes them downstream, but any future consumer reading @@ -2985,7 +3000,17 @@ def _read_entitlement_cache() -> dict: # so a cache written by an older build simply has none of them and reads back as "not a # trial" rather than as a corrupt entry. resolved.update(_trial_facts(value)) - return resolved + return resolved, digest + + +def _read_entitlement_cache() -> dict: + """Return the last cached ``GET /v1/entitlements`` answer, or ``{}``. Never raises. + + Secondary to ``_session_entitlement``: this file exists only for a control plane that + does not yet return the entitlement on registration and refresh. + """ + + return _read_entitlement_cache_snapshot()[0] def _write_entitlement_cache(entitlement: dict) -> bool: @@ -3063,23 +3088,82 @@ def _deny_entitlement_cache() -> bool: return False +def _persisted_state_digest(source: str) -> Optional[str]: + """Digest the persisted bytes backing one entitlement source, for change detection. + + ``None`` means "could not determine" and must be treated as "unchanged" — a probe + failure must never look like the superseding rewrite it cannot prove. Wall-clock + timestamps cannot order a reconnect against a denial: two writes inside one coarse + clock tick stamp equal ``entitlement_checked_at``/``fetched_at`` values, so only + content distinguishes a post-denial reconnect from a pre-denial record. + """ + + try: + if source == "session": + from engraphis import cloud_session + + return cloud_session.saved_session_digest() + if source == "cloud": + path = _entitlement_cache_path() + if path is None: + return "" + from engraphis.private_state import read_private_text + + raw = read_private_text( + path, max_bytes=_ENTITLEMENT_MAX_RESPONSE_BYTES, allow_missing=True + ) + if raw is None: + return "" + return hashlib.sha256(raw.encode("utf-8", "surrogatepass")).hexdigest() + except Exception: # noqa: BLE001 - an unreadable state file must fail closed + return None + return None + + def _mark_authoritative_denial() -> None: """Make an authoritative cloud denial visible before persistence starts.""" - global _authoritative_denial_at + global _authoritative_denial_at, _denied_state_digests + denial_at = time.time() with _ENTITLEMENT_REFRESH_LOCK: - _authoritative_denial_at = time.time() + _authoritative_denial_at = denial_at + # Publish the fail-closed state before probing either potentially slow or + # unavailable persisted source. Readers must not observe paid access while a + # network-mounted state directory is blocking the digest capture below. + _denied_state_digests = {"session": None, "cloud": None} _AUTHORITATIVE_DENIAL_PENDING.set() - - -def _clear_superseded_denial(checked_at: float) -> bool: - """Clear the process guard only for a newer active authoritative answer.""" + digests = { + source: _persisted_state_digest(source) for source in ("session", "cloud") + } + with _ENTITLEMENT_REFRESH_LOCK: + # A later denial owns the newer baseline; never let an older slow probe replace it. + if _authoritative_denial_at == denial_at: + _denied_state_digests = digests + + +def _clear_superseded_denial(known_source: str, observed_digest: Optional[str]) -> bool: + """Clear the process guard only for an answer parsed from post-denial bytes. + + ``observed_digest`` fingerprints the exact bytes ``known`` was parsed from, captured + in the same read — never a fresh stat of a file that may have changed since. A + genuine reconnect rewrites the session (the control plane rotates the refresh + credential) or the entitlement cache with a fresh answer, so its digest differs from + the baseline the denial captured. Equal digests mean the active-looking record + predates the denial — however equal their wall-clock stamps are, and even when the + denial's own persistence write landed between the parse and this check — and must + not resurrect grants the control plane just refused. Unknown digests on either side + (unreadable state at denial or parse time) can never prove supersession, so the + guard sticks — fail-closed — until the next denial cycle or process restart. + """ global _authoritative_denial_at with _ENTITLEMENT_REFRESH_LOCK: + baseline = _denied_state_digests.get(known_source) if ( _AUTHORITATIVE_DENIAL_PENDING.is_set() - and checked_at > _authoritative_denial_at + and baseline is not None + and observed_digest is not None + and observed_digest != baseline ): _AUTHORITATIVE_DENIAL_PENDING.clear() _authoritative_denial_at = 0.0 @@ -3359,10 +3443,10 @@ def _plan_entitlement() -> dict: if _AUTHORITATIVE_DENIAL_PENDING.is_set(): # Reads are safe here: only access is overridden. Keeping the last known paid plan # lets the UI direct a lapsed Team customer to billing without restoring any grant. - known = _session_entitlement() + known, observed_digest = _session_entitlement_snapshot() known_source = "session" if not known: - known = _read_entitlement_cache() + known, observed_digest = _read_entitlement_cache_snapshot() known_source = "cloud" try: known_checked_at = float(known.get("fetched_at") or 0.0) @@ -3371,7 +3455,7 @@ def _plan_entitlement() -> dict: if ( known and bool(known.get("cloud_access_active")) - and _clear_superseded_denial(known_checked_at) + and _clear_superseded_denial(known_source, observed_digest) ): return _resolved_entitlement(known, source=known_source) with _ENTITLEMENT_REFRESH_LOCK: diff --git a/tests/test_consolidate.py b/tests/test_consolidate.py index ccaf8732..aef77f94 100644 --- a/tests/test_consolidate.py +++ b/tests/test_consolidate.py @@ -1715,6 +1715,10 @@ def test_safety_repair_cursor_eventually_reaches_rows_beyond_limit(monkeypatch): mtype=MemoryType.SEMANTIC, resolve_conflicts=False, ) + # ULIDs sort by creation millisecond; rows created within the same millisecond + # order randomly inside it. Force the derived row to sort strictly after every + # noise row so the paging limit genuinely defers it to a later repair sweep. + time.sleep(0.005) derived_id = eng.remember( "Derived summary.", workspace_id=workspace_id, @@ -1969,8 +1973,11 @@ def test_archive_preserves_vector_for_historical_recall(): ) eng.store.conn.commit() + # valid_from defaults to creation time; a same-tick archived_at would make the + # historical as_of midpoint land exactly on valid_to and the half-open temporal + # predicate would exclude the row. Nudge the archive stamp well into the future + # so the midpoint sits strictly inside [valid_from, valid_to) even on coarse clocks. archived_at = time.time() + 3_600 - report = consolidate(eng, workspace_id=wid, now=archived_at) assert [row["id"] for row in report["archived"]] == [stale] diff --git a/tests/test_hosted_plan_resolution.py b/tests/test_hosted_plan_resolution.py index 352509a6..b139a153 100644 --- a/tests/test_hosted_plan_resolution.py +++ b/tests/test_hosted_plan_resolution.py @@ -1065,6 +1065,42 @@ def test_newer_active_session_clears_the_process_denial_guard(monkeypatch) -> No assert not v2_api._AUTHORITATIVE_DENIAL_PENDING.is_set() +def test_byte_identical_post_denial_session_never_supersedes( + monkeypatch, tmp_path, +) -> None: + """Replaying the exact pre-denial session bytes cannot resurrect grants. + + Supersession is decided by a content digest, never by timestamps: under + coarse clocks a restored active-looking record ties the denial stamp, and + only *different* bytes parsed after the denial prove a genuine reconnect. + """ + + monkeypatch.setenv("ENGRAPHIS_STATE_DIR", str(tmp_path)) + _connect(monkeypatch, pinned_token=False) + response = { + "refresh_credential": "engr_rt_stale_active", + "organization_id": ORGANIZATION, + "token_subject": "member", + } + response.update(_registration_entitlement("team")) + cloud_session.save_bootstrap(response, control_url=CONTROL_URL) + session_file = tmp_path / "cloud_session.json" + stale_active_bytes = session_file.read_bytes() + + monkeypatch.setenv("ENGRAPHIS_CLOUD_ENTITLEMENT_REFRESH", "0") + v2_api._mark_authoritative_denial() + # A stale writer restores the pre-denial record verbatim -- indistinguishable + # from the original by every clock-based signal, and therefore by none. + session_file.write_bytes(stale_active_bytes) + + payload = v2_api.get_license() + + assert payload["plan"] == "team" + assert payload["cloud_access_active"] is False + assert payload["features"] == [] + assert v2_api._AUTHORITATIVE_DENIAL_PENDING.is_set() + + def test_denial_guard_precedes_a_blocked_persistence_write(monkeypatch) -> None: """Readers fail closed while the durable denial write is still blocked.""" @@ -1098,6 +1134,65 @@ 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 == {"session": None, "cloud": None} + 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_denial_guard_reads_do_not_wait_for_digest_probe(monkeypatch) -> None: + """A slow baseline probe cannot block a fail-closed license response.""" + + _connect(monkeypatch, pinned_token=False) + _serve(monkeypatch, _FakeControlPlane( + _entitlement_dto("team"), + registration=_registration_entitlement("team"), + )) + assert _settled_license(monkeypatch)["cloud_access_active"] is True + monkeypatch.setenv("ENGRAPHIS_CLOUD_ENTITLEMENT_REFRESH", "0") + entered = threading.Event() + release = threading.Event() + original = v2_api._persisted_state_digest + + def _blocked_digest(source): + entered.set() + assert release.wait(timeout=5.0) + return original(source) + + monkeypatch.setattr(v2_api, "_persisted_state_digest", _blocked_digest) + worker = threading.Thread(target=v2_api._record_authoritative_denial) + worker.start() + assert entered.wait(timeout=5.0) + try: + payload = v2_api.get_license() + assert payload["cloud_access_active"] is False + assert payload["features"] == [] + 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.""" diff --git a/tests/test_obsidian_import_schema.py b/tests/test_obsidian_import_schema.py index 5ec1ec29..99a1504c 100644 --- a/tests/test_obsidian_import_schema.py +++ b/tests/test_obsidian_import_schema.py @@ -48,7 +48,7 @@ def test_source_manifest_is_scoped_idempotent_and_marks_missing(): importer_version="1", seen_at=10, ) assert item_id.startswith("src_") - assert store.mark_source_import_items_missing(vault_id=vault_id, seen_before=11) == 1 + assert len(store.mark_source_import_items_missing(vault_id=vault_id, seen_before=11)) == 1 item = store.get_source_import_item(vault_id=vault_id, source_key="b" * 64) assert item["state"] == "missing" store.upsert_source_import_item( diff --git a/tests/test_obsidian_importer.py b/tests/test_obsidian_importer.py index d89ba8c1..cbc604fd 100644 --- a/tests/test_obsidian_importer.py +++ b/tests/test_obsidian_importer.py @@ -626,6 +626,47 @@ def __iter__(self): assert items.iterations <= 3 +def test_manifest_cursor_deduplicates_a_row_renamed_forward_between_pages(): + class MovingItems: + def __init__(self): + self.rows = [ + { + "id": f"src_{index:05d}", + "source_key": f"{index + 1:064x}", + "relative_path": f"p{index:05d}.md", + } + for index in range(20_001) + ] + self.calls = 0 + + def list_source_import_items(self, *, vault_id, states=None, limit=10_000, + after_path="", after_id=""): + del vault_id, states + self.calls += 1 + if self.calls == 2: + # A non-snapshot reader can observe this row again after it moves + # beyond the first page's cursor. The planner must keep one identity. + self.rows[0]["relative_path"] = "z00000.md" + rows = sorted(self.rows, key=lambda row: (row["relative_path"], row["id"])) + if after_path: + rows = [ + row for row in rows + if row["relative_path"] > after_path + or (row["relative_path"] == after_path and row["id"] > after_id) + ] + return [dict(row) for row in rows[:limit]] + + moving = MovingItems() + importer = ObsidianImporter() + importer.store = moving + items, complete = importer._all_source_items(vault_id="vault") + + assert complete is True + assert len(items) == len({row["id"] for row in items}) == 20_001 + moved = next(row for row in items if row["id"] == "src_00000") + assert moved["relative_path"] == "z00000.md" + + def test_conflict_new_branch_and_atomic_note_failure(tmp_path: Path, monkeypatch): vault = tmp_path / "Vault" vault.mkdir() diff --git a/tests/test_obsidian_service.py b/tests/test_obsidian_service.py index bfcdc810..d11ec80d 100644 --- a/tests/test_obsidian_service.py +++ b/tests/test_obsidian_service.py @@ -1,12 +1,13 @@ """Real-service coverage for the owner-only Obsidian import facade.""" from __future__ import annotations +import hashlib import time 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"} @@ -43,6 +44,110 @@ def _import(service: MemoryService, files: list[tuple[str, bytes]], **kwargs) -> return started, _await_job(service, started) +def test_preview_pages_the_full_manifest_like_execution(): + service = _service() + try: + started, imported = _import( + service, + [ + ("A.md", b"# A\n"), + ("B.md", b"# B\n"), + ("C.md", b"# C\n"), + ], + ) + vault_id = started["vault_id"] + assert imported["state"] == "completed" + + # Push the vault manifest past the default 10k list-page boundary with + # filler identity rows that sort before the scan set, so an unpaged + # preview loses exactly the rows a real oversized vault would lose. + for i in range(10_001): + service.store.upsert_source_import_item( + vault_id=vault_id, + source_key=hashlib.sha256(f"seed-{i}".encode()).hexdigest(), + relative_path=f"0000-seed-{i:05d}.md", + ) + + preview = service.preview_obsidian_upload( + files=[("A.md", b"# A\n"), ("D.md", b"# D\n")], + attachment_manifest=[], + workspace="alpha", + vault_label="Team notes", + vault_id=vault_id, + ) + + # An unpaged preview reads only the first list page, so manifest rows + # beyond the boundary vanish from the report entirely. The paged reader + # must surface every manifest row: A plans as unchanged and B/C — not + # part of this preview's scan — are reported as missing, not dropped. + statuses = { + row["relative_path"]: row["status"] for row in preview["files"] + } + assert statuses["A.md"] != "missing" + assert "B.md" in statuses + assert "C.md" in statuses + finally: + 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_late_manifest_truncation_marks_import_partial(monkeypatch): + service = _service() + try: + original = ObsidianImporter._all_source_items + calls = 0 + + def _late_truncation(self, *, vault_id, states=None): + nonlocal calls + calls += 1 + items, complete = original(self, vault_id=vault_id, states=states) + return items, complete if calls == 1 else False + + monkeypatch.setattr(ObsidianImporter, "_all_source_items", _late_truncation) + _, report = _import( + service, + [("A.md", b"# A\nSee [[B]].\n"), ("B.md", b"# B\n")], + ) + + assert calls >= 2 + assert report["state"] == "partial" + finally: + service.close() + + def test_preview_is_write_free_and_service_enforces_confirmation_and_upload_guards(monkeypatch): service = _service() try: