From 200c858e88d2b3ed6149c56603270526f9fd3de0 Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Fri, 21 Aug 2026 00:00:13 -0400 Subject: [PATCH 01/12] fix: deterministic source-import missing detection and denial-guard supersession Two failing offline-gate tests and two latent defects: - mark_source_import_items_missing() gains an explicit source_keys path: a complete scan now marks exactly the planned-missing rows instead of relying on the last_seen_at < run_started heuristic, which a renamed file defeated (rename stamps last_seen_at=now, so the next run's delete never matched). The documents importer shares ObsidianImporter.import_scan and inherits this. - _clear_superseded_denial() now supersedes a billing denial on persisted-state content change, not wall-clock comparison. Equal-timestamp entitlements made the strict '>' stick forever after a valid reconnect, and a naive '>=' would let a pre-denial record saved in the same coarse clock tick resurrect grants. cloud_session.saved_session_digest() fingerprints the session bytes so write order is observed without exposing credentials. - tests: deflake ULID same-millisecond ordering (repair cursor sweep count) and the archived_at == valid_from boundary (half-open temporal exclusion). Verified: full offline suite green, ruff/pyright clean, eval gates unchanged. --- engraphis/cloud_session.py | 23 +++++++++++++++ engraphis/core/store.py | 9 ++++++ engraphis/obsidian_import.py | 1 + engraphis/routes/v2_api.py | 55 ++++++++++++++++++++++++++++++++---- tests/test_consolidate.py | 10 ++++++- 5 files changed, 92 insertions(+), 6 deletions(-) diff --git a/engraphis/cloud_session.py b/engraphis/cloud_session.py index 5c2dd3e8..c6e1299b 100644 --- a/engraphis/cloud_session.py +++ b/engraphis/cloud_session.py @@ -531,6 +531,29 @@ def saved_entitlement() -> dict: return {} +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 6ba3cf18..f57d7b47 100644 --- a/engraphis/core/store.py +++ b/engraphis/core/store.py @@ -3830,9 +3830,11 @@ 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, + source_keys: Iterable[str] = (), ) -> int: if self._source_vault_row(vault_id) is None: return 0 + source_keys = {str(k) for k in source_keys if k} 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,6 +3842,13 @@ def mark_source_import_items_missing( "AND relative_path=? AND state NOT IN ('missing','conflict')", (float(seen_before), vault_id, relative_path), ) + if source_keys: + placeholders = ",".join("?" for _ in source_keys) + return int(self.conn.execute( + f"UPDATE source_imports SET state='missing', missing_at=? WHERE vault_id=? " + f"AND source_key IN ({placeholders})", + (now_ts(), vault_id, *source_keys), + ).rowcount) 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 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"} @@ -3049,23 +3050,67 @@ 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 with _ENTITLEMENT_REFRESH_LOCK: _authoritative_denial_at = time.time() + _denied_state_digests = { + source: _persisted_state_digest(source) for source in ("session", "cloud") + } _AUTHORITATIVE_DENIAL_PENDING.set() -def _clear_superseded_denial(checked_at: float) -> bool: - """Clear the process guard only for a newer active authoritative answer.""" +def _clear_superseded_denial(known_source: str) -> bool: + """Clear the process guard only for a state rewritten after the denial. + + A genuine reconnect rewrites the session (the control plane rotates the refresh + credential) or the entitlement cache with a fresh answer. If the bytes backing + ``known_source`` are exactly the ones the denial observed, the active-looking record + predates the denial — however equal their wall-clock stamps are — and must not + resurrect grants the control plane just refused. + """ global _authoritative_denial_at with _ENTITLEMENT_REFRESH_LOCK: + current = _persisted_state_digest(known_source) if ( _AUTHORITATIVE_DENIAL_PENDING.is_set() - and checked_at > _authoritative_denial_at + and current is not None + and current != _denied_state_digests.get(known_source) ): _AUTHORITATIVE_DENIAL_PENDING.clear() _authoritative_denial_at = 0.0 @@ -3357,7 +3402,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) ): return _resolved_entitlement(known, source=known_source) with _ENTITLEMENT_REFRESH_LOCK: diff --git a/tests/test_consolidate.py b/tests/test_consolidate.py index efd95ed9..ce9545d9 100644 --- a/tests/test_consolidate.py +++ b/tests/test_consolidate.py @@ -1710,6 +1710,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, @@ -1956,7 +1960,11 @@ def test_archive_preserves_vector_for_historical_recall(): ) eng.store.conn.commit() - archived_at = time.time() + # 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 into the future so the + # midpoint sits strictly inside [valid_from, valid_to). + archived_at = time.time() + 0.5 report = consolidate(eng, workspace_id=wid, now=archived_at) assert [row["id"] for row in report["archived"]] == [stale] From e374bc1b79ae1be6966cf71cc11ed6a33164762a Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Fri, 21 Aug 2026 02:35:04 -0400 Subject: [PATCH 02/12] =?UTF-8?q?fix:=20review=20findings=20=E2=80=94=20un?= =?UTF-8?q?known-baseline=20guard=20and=20full-manifest=20paging?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1 (v2_api): _clear_superseded_denial now requires both the captured baseline digest and the current digest to be known before treating a difference as a superseding rewrite. An unreadable state file at denial time left None as the baseline; a later recovered read then differed from it and cleared the guard even though the file still held the pre-denial active entitlement. Unknown baselines now stick (fail-closed) until the next denial cycle or restart. P2 (source imports): list_source_import_items gains offset paging and ObsidianImporter._all_source_items() pages the full manifest (bounded at 200k rows) for both import planning and link reconciliation. A manifest grown past the 10k single-page cap left historical rows beyond the page invisible, so the explicit source_keys missing-marking path skipped them while the run reported itself complete. Verified: unknown-baseline fail-closed probe, reconnect-supersedes probe, 10,050-row paging probe (complete, duplicate-free), full affected test files (obsidian importer/schema, document importer, hosted plan) exit 0, ruff+pyright clean. --- engraphis/core/store.py | 5 +++-- engraphis/obsidian_import.py | 25 ++++++++++++++++++++++--- engraphis/routes/v2_api.py | 10 ++++++++-- 3 files changed, 33 insertions(+), 7 deletions(-) diff --git a/engraphis/core/store.py b/engraphis/core/store.py index f57d7b47..1497c3c2 100644 --- a/engraphis/core/store.py +++ b/engraphis/core/store.py @@ -3733,7 +3733,7 @@ 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, offset: int = 0) -> list[dict]: if self._source_vault_row(vault_id) is None: return [] params: list[Any] = [vault_id] @@ -3743,8 +3743,9 @@ def list_source_import_items(self, *, vault_id: str, states: Optional[list[str]] 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 LIMIT ? OFFSET ?" params.append(max(1, min(100_000, int(limit)))) + params.append(max(0, int(offset))) return [dict(row) for row in self.conn.execute(sql, params).fetchall()] def upsert_source_import_item(self, *, vault_id: str, source_key: str, relative_path: str, diff --git a/engraphis/obsidian_import.py b/engraphis/obsidian_import.py index ae4017f6..96c08370 100644 --- a/engraphis/obsidian_import.py +++ b/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 = 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( @@ -948,12 +948,31 @@ def _metadata( }, } + def _all_source_items(self, *, vault_id: str, + states: Optional[list[str]] = None) -> list[dict]: + """Page through the full manifest so truncation cannot hide historical rows. + + ``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, or rows beyond the first page silently stay live while + the run reports itself complete. + """ + items: list[dict] = [] + page_size = 10_000 + 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, offset=len(items), + ) + items.extend(page) + if len(page) < page_size: + break + return items + 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 = self._all_source_items( vault_id=vault_id, states=["imported", "unchanged", "renamed", "skipped", "missing"], ) diff --git a/engraphis/routes/v2_api.py b/engraphis/routes/v2_api.py index 065045c5..e9204637 100644 --- a/engraphis/routes/v2_api.py +++ b/engraphis/routes/v2_api.py @@ -3101,16 +3101,22 @@ def _clear_superseded_denial(known_source: str) -> bool: credential) or the entitlement cache with a fresh answer. If the bytes backing ``known_source`` are exactly the ones the denial observed, the active-looking record predates the denial — however equal their wall-clock stamps are — and must not - resurrect grants the control plane just refused. + resurrect grants the control plane just refused. Both digests must be known: an + unreadable state file at denial time leaves ``None`` as the baseline, and a later + recovered read must never look like the superseding rewrite it cannot prove — the + guard then stays set until the next denial cycle or process restart, which fails + closed. """ global _authoritative_denial_at with _ENTITLEMENT_REFRESH_LOCK: current = _persisted_state_digest(known_source) + baseline = _denied_state_digests.get(known_source) if ( _AUTHORITATIVE_DENIAL_PENDING.is_set() + and baseline is not None and current is not None - and current != _denied_state_digests.get(known_source) + and current != baseline ): _AUTHORITATIVE_DENIAL_PENDING.clear() _authoritative_denial_at = 0.0 From fa15f95ced829366bb373fd681a0c9fb9578a719 Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Fri, 21 Aug 2026 03:10:43 -0400 Subject: [PATCH 03/12] =?UTF-8?q?fix:=20round-2=20review=20=E2=80=94=20gen?= =?UTF-8?q?eration-guarded=20missing=20marks=20and=20bounded=20paging?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three follow-up findings on the source-import missing path: - Concurrent imports: the exact-key update is now per-key and conditioned on the row still matching its planned generation (last_seen_at, last_seen_job_id, live state). A newer run that re-upserts a row after an older run planned it missing keeps its fresh state instead of being clobbered back to missing. - SQLite host-parameter limits: the oversized IN clause is gone entirely — per-key updates in one transaction scale to any manifest (200k keys marked in ~6s locally). - Paging cap: _all_source_items() now reports whether the full manifest was read. Beyond the 200k-row memory bound the run is marked partial, missing finalization is skipped, and link reconciliation refuses to retire edges on the incomplete view. Probes: stale-plan-vs-refreshed-row preserved; 200,050-row manifest flags truncation and marks all 200k planned keys; affected suites (obsidian importer/schema, document importer, hosted plan, consolidate) exit 0; ruff+pyright clean. --- engraphis/core/store.py | 37 +++++++++++++++++++++++++++--------- engraphis/obsidian_import.py | 33 ++++++++++++++++++++++---------- 2 files changed, 51 insertions(+), 19 deletions(-) diff --git a/engraphis/core/store.py b/engraphis/core/store.py index 1497c3c2..2ab27359 100644 --- a/engraphis/core/store.py +++ b/engraphis/core/store.py @@ -3831,11 +3831,24 @@ 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, - source_keys: Iterable[str] = (), + 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 - source_keys = {str(k) for k in source_keys if k} + 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( @@ -3843,13 +3856,19 @@ def mark_source_import_items_missing( "AND relative_path=? AND state NOT IN ('missing','conflict')", (float(seen_before), vault_id, relative_path), ) - if source_keys: - placeholders = ",".join("?" for _ in source_keys) - return int(self.conn.execute( - f"UPDATE source_imports SET state='missing', missing_at=? WHERE vault_id=? " - f"AND source_key IN ({placeholders})", - (now_ts(), vault_id, *source_keys), - ).rowcount) + 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 list[dict]: + states: Optional[list[str]] = None) -> tuple[list[dict], bool]: """Page through the full manifest so truncation cannot hide historical rows. ``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, or rows beyond the first page silently stay live while - the run reports itself complete. + the run reports itself complete. Returns the rows and whether the whole manifest + was read: the paging bound (200k rows) is a memory cap, not an assumption, and a + manifest beyond it must push the run to partial rather than pass silently. """ items: list[dict] = [] page_size = 10_000 @@ -965,17 +972,23 @@ def _all_source_items(self, *, vault_id: str, ) items.extend(page) if len(page) < page_size: - break - return items + return items, True + return items, False def _reconcile_links( self, scan: _ImportScan, *, vault_id: str, job_id: Optional[str] = None, cancel_check: Optional[Callable[[], bool]] = None, ) -> list[dict]: - items = self._all_source_items( + """Resolve derived links in bounded, cancellable, replay-safe batches.""" + 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") From afdcd3cab488702dde77f7b783155dbbd52a3f4d Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Fri, 21 Aug 2026 03:31:07 -0400 Subject: [PATCH 04/12] =?UTF-8?q?fix:=20round-3=20review=20=E2=80=94=20par?= =?UTF-8?q?se-bound=20denial=20digests=20and=20keyset=20manifest=20paging?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Denial guard race: the supersession digest now travels with the parse. _session_entitlement_snapshot() and _read_entitlement_cache_snapshot() return the entitlement plus a sha256 of the exact bytes it was parsed from, and _clear_superseded_denial() compares that observed digest against the denial baseline instead of re-reading a file that may have changed since. The denial persistence write landing between a license read's parse and its check can no longer pose as a superseding reconnect. cloud_session gains saved_entitlement_snapshot(); _session_entitlement()/_read_entitlement_cache() remain as thin wrappers for their other callers. - Manifest paging: OFFSET is replaced by a (relative_path, id) keyset cursor. OFFSET on a live ORDER BY lets a concurrent rename shift an unread row across the page boundary so it is silently skipped while the pager reports a complete read; the keyset cursor returns every row at or after the cursor exactly once, and a row renamed below the read range degrades into content-hash rename detection. - Cap boundary: _all_source_items() probes one row past the 200k-row memory cap, so a manifest of exactly 200,000 rows reads as complete instead of forcing every such import to partial. Probes: mid-read denial write leaves the guard set while a post-write reconnect parse clears it; exactly-at-cap manifest reports complete; beyond-cap reports partial with a duplicate-free plan set; 200k-row keyset read in ~2s. Affected suites (obsidian importer/schema, document importer, hosted plan, consolidate) exit 0; ruff+pyright clean. --- engraphis/cloud_session.py | 37 +++++++++++++ engraphis/core/store.py | 16 +++++- engraphis/obsidian_import.py | 26 ++++++--- engraphis/routes/v2_api.py | 104 +++++++++++++++++++++-------------- 4 files changed, 133 insertions(+), 50 deletions(-) diff --git a/engraphis/cloud_session.py b/engraphis/cloud_session.py index c6e1299b..4bdb7f80 100644 --- a/engraphis/cloud_session.py +++ b/engraphis/cloud_session.py @@ -531,6 +531,43 @@ 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. diff --git a/engraphis/core/store.py b/engraphis/core/store.py index 2ab27359..054348bc 100644 --- a/engraphis/core/store.py +++ b/engraphis/core/store.py @@ -3733,19 +3733,29 @@ 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, offset: int = 0) -> 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 ? OFFSET ?" + sql += " ORDER BY relative_path, id LIMIT ?" params.append(max(1, min(100_000, int(limit)))) - params.append(max(0, int(offset))) return [dict(row) for row in self.conn.execute(sql, params).fetchall()] def upsert_source_import_item(self, *, vault_id: str, source_key: str, relative_path: str, diff --git a/engraphis/obsidian_import.py b/engraphis/obsidian_import.py index f37d7980..2a4c110d 100644 --- a/engraphis/obsidian_import.py +++ b/engraphis/obsidian_import.py @@ -955,25 +955,37 @@ def _metadata( def _all_source_items(self, *, vault_id: str, states: Optional[list[str]] = None) -> tuple[list[dict], bool]: - """Page through the full manifest so truncation cannot hide historical rows. + """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, or rows beyond the first page silently stay live while - the run reports itself complete. Returns the rows and whether the whole manifest - was read: the paging bound (200k rows) is a memory cap, not an assumption, and a - manifest beyond it must push the run to partial rather than pass silently. + 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, offset=len(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 - return items, False + 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, diff --git a/engraphis/routes/v2_api.py b/engraphis/routes/v2_api.py index e9204637..cbbdcf9f 100644 --- a/engraphis/routes/v2_api.py +++ b/engraphis/routes/v2_api.py @@ -2860,27 +2860,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 = { @@ -2896,36 +2896,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. @@ -2933,7 +2947,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): @@ -2955,7 +2969,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 @@ -2972,7 +2986,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: @@ -3094,29 +3118,29 @@ def _mark_authoritative_denial() -> None: _AUTHORITATIVE_DENIAL_PENDING.set() -def _clear_superseded_denial(known_source: str) -> bool: - """Clear the process guard only for a state rewritten after the denial. +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. - A genuine reconnect rewrites the session (the control plane rotates the refresh - credential) or the entitlement cache with a fresh answer. If the bytes backing - ``known_source`` are exactly the ones the denial observed, the active-looking record - predates the denial — however equal their wall-clock stamps are — and must not - resurrect grants the control plane just refused. Both digests must be known: an - unreadable state file at denial time leaves ``None`` as the baseline, and a later - recovered read must never look like the superseding rewrite it cannot prove — the - guard then stays set until the next denial cycle or process restart, which fails - closed. + ``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: - current = _persisted_state_digest(known_source) baseline = _denied_state_digests.get(known_source) if ( _AUTHORITATIVE_DENIAL_PENDING.is_set() and baseline is not None - and current is not None - and current != baseline + and observed_digest is not None + and observed_digest != baseline ): _AUTHORITATIVE_DENIAL_PENDING.clear() _authoritative_denial_at = 0.0 @@ -3396,10 +3420,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) @@ -3408,7 +3432,7 @@ def _plan_entitlement() -> dict: if ( known and bool(known.get("cloud_access_active")) - and _clear_superseded_denial(known_source) + and _clear_superseded_denial(known_source, observed_digest) ): return _resolved_entitlement(known, source=known_source) with _ENTITLEMENT_REFRESH_LOCK: From ff4aba6ef943131ee6e251be314d5d19327e0895 Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Fri, 21 Aug 2026 03:41:19 -0400 Subject: [PATCH 05/12] =?UTF-8?q?fix:=20round-4=20review=20=E2=80=94=20fin?= =?UTF-8?q?alize=20only=20rows=20the=20guarded=20update=20actually=20marke?= =?UTF-8?q?d?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mark_source_import_items_missing() returns the source_keys it actually marked instead of a count: the guarded per-key path collects keys whose generation predicates matched, and the heuristic path selects its rows before updating (chunked IN under host-parameter limits). import_scan records job history and the completed report from that reality — rows the generation guard left live because a concurrent import refreshed them are recorded as skipped, never as missing, so the job receipt can no longer claim a live, newer source was removed. Probe: stale plan vs refreshed row -> only the stale key returned and marked; refreshed row stays imported and out of the missing report; heuristic path returns keys. Affected suites exit 0; ruff+pyright clean. --- engraphis/core/store.py | 40 +++++++++++++++++++--------- engraphis/obsidian_import.py | 23 +++++++++++++--- tests/test_obsidian_import_schema.py | 2 +- 3 files changed, 48 insertions(+), 17 deletions(-) diff --git a/engraphis/core/store.py b/engraphis/core/store.py index 054348bc..ac315243 100644 --- a/engraphis/core/store.py +++ b/engraphis/core/store.py @@ -3842,7 +3842,7 @@ 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: + ) -> 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 @@ -3851,9 +3851,11 @@ def mark_source_import_items_missing( 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")) @@ -3867,24 +3869,38 @@ def mark_source_import_items_missing( (float(seen_before), vault_id, relative_path), ) if planned: - updated = 0 + marked: list[str] = [] stamp = now_ts() for key, seen_at, seen_job in planned: - updated += int(self.conn.execute( + 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) - 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 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 2a4c110d..700e1e72 100644 --- a/engraphis/obsidian_import.py +++ b/engraphis/obsidian_import.py @@ -367,18 +367,33 @@ def import_scan( progress(dict(outcome)) self._check_cancel(job_id, cancel_check) if can_finalize_missing: - self.store.mark_source_import_items_missing( + marked_keys = set(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: + )) + 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 item in finalized: + 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 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( From 4813f6a772c802af495cdda811d95af582d010d4 Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Fri, 21 Aug 2026 03:47:38 -0400 Subject: [PATCH 06/12] =?UTF-8?q?fix:=20round-5=20review=20=E2=80=94=20con?= =?UTF-8?q?stant-time=20finalized=20check=20in=20missing=20finalization?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The guard-skipped loop tested membership against the finalized list, scanning up to 200k dicts per missing item — quadratic job finalization that can appear hung on a large complete import. Test the item's source_key against the already-built marked_keys set instead. --- engraphis/obsidian_import.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/engraphis/obsidian_import.py b/engraphis/obsidian_import.py index 700e1e72..7a920ade 100644 --- a/engraphis/obsidian_import.py +++ b/engraphis/obsidian_import.py @@ -383,7 +383,7 @@ def import_scan( planned_action="missing", result_state="missing", ) for item in missing: - if item in finalized: + 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 From d7a178ae50c909730e0396be3cb81d23771db663 Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Sat, 22 Aug 2026 10:05:44 -0400 Subject: [PATCH 07/12] test(hosted): pin byte-identical denial-supersession invariant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add the invariant test the review round identified as missing: replaying the exact pre-denial session bytes after _mark_authoritative_denial must NOT clear the process guard — supersession is content-digest-based, never timestamp-based, so coarse-clock ties cannot resurrect grants. - Clarify the finalized-keys check in obsidian_import: hash-set membership (O(1) average) replaced a quadratic list scan; it is a complexity fix, not a timing-sensitive comparison (commit 4813f6a's "constant-time" wording overclaimed). --- engraphis/obsidian_import.py | 3 +++ tests/test_hosted_plan_resolution.py | 36 ++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/engraphis/obsidian_import.py b/engraphis/obsidian_import.py index 7a920ade..076b8899 100644 --- a/engraphis/obsidian_import.py +++ b/engraphis/obsidian_import.py @@ -367,6 +367,9 @@ def import_scan( progress(dict(outcome)) self._check_cancel(job_id, cancel_check) if can_finalize_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), diff --git a/tests/test_hosted_plan_resolution.py b/tests/test_hosted_plan_resolution.py index 1b3369ee..cfa6861c 100644 --- a/tests/test_hosted_plan_resolution.py +++ b/tests/test_hosted_plan_resolution.py @@ -1061,6 +1061,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.""" From 8fba5d103e7b221e1c01cb22bc8d5a239216d18d Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Sun, 23 Aug 2026 08:46:55 -0400 Subject: [PATCH 08/12] fix(import): page import previews like execution Preview plans read the vault manifest through the same keyset-paged reader as execution, so manifests larger than one list page no longer make previews silently drop beyond-boundary rows; they are reported as missing like any other unseen source. Regression test seeds a real 10k-row manifest and fails against the unpaged reader (stash-verified). --- CHANGELOG.md | 3 +++ engraphis/obsidian_import.py | 4 ++- tests/test_obsidian_service.py | 47 ++++++++++++++++++++++++++++++++++ 3 files changed, 53 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f118ca59..9a77078e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -108,6 +108,9 @@ 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. - 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/obsidian_import.py b/engraphis/obsidian_import.py index 076b8899..0d6894a1 100644 --- a/engraphis/obsidian_import.py +++ b/engraphis/obsidian_import.py @@ -511,7 +511,9 @@ 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, _ = 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: diff --git a/tests/test_obsidian_service.py b/tests/test_obsidian_service.py index bfcdc810..5a47632d 100644 --- a/tests/test_obsidian_service.py +++ b/tests/test_obsidian_service.py @@ -1,6 +1,7 @@ """Real-service coverage for the owner-only Obsidian import facade.""" from __future__ import annotations +import hashlib import time import pytest @@ -43,6 +44,52 @@ 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_is_write_free_and_service_enforces_confirmation_and_upload_guards(monkeypatch): service = _service() try: From 66724b8e167dee7f6e70f5729bbc7088593757c4 Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Sun, 23 Aug 2026 20:19:19 -0400 Subject: [PATCH 09/12] fix(import): stabilize manifest pagination under concurrent renames --- CHANGELOG.md | 8 ++-- engraphis/obsidian_import.py | 75 +++++++++++++++++++++++---------- tests/test_obsidian_importer.py | 41 ++++++++++++++++++ 3 files changed, 99 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a77078e..b2ada47c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -108,9 +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. +- 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/obsidian_import.py b/engraphis/obsidian_import.py index 0d6894a1..ccef01ef 100644 --- a/engraphis/obsidian_import.py +++ b/engraphis/obsidian_import.py @@ -979,33 +979,64 @@ def _all_source_items(self, *, vault_id: str, ``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. + 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 = "" - 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, + 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() ) - return items, not extra + 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, 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() From 82884086114c58a04aa7a8f41c9da245ea096ed6 Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Mon, 24 Aug 2026 05:42:56 -0400 Subject: [PATCH 10/12] fix(import): fail closed on truncated previews and denials --- engraphis/obsidian_import.py | 28 ++++++++++++++++++---- engraphis/routes/v2_api.py | 8 ++++++- tests/test_hosted_plan_resolution.py | 26 ++++++++++++++++++++ tests/test_obsidian_service.py | 36 +++++++++++++++++++++++++++- 4 files changed, 92 insertions(+), 6 deletions(-) diff --git a/engraphis/obsidian_import.py b/engraphis/obsidian_import.py index ccef01ef..222da940 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( @@ -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], diff --git a/engraphis/routes/v2_api.py b/engraphis/routes/v2_api.py index eaf64e7b..fabe9b80 100644 --- a/engraphis/routes/v2_api.py +++ b/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: diff --git a/tests/test_hosted_plan_resolution.py b/tests/test_hosted_plan_resolution.py index 8c0eda36..67dfb05c 100644 --- a/tests/test_hosted_plan_resolution.py +++ b/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.""" diff --git a/tests/test_obsidian_service.py b/tests/test_obsidian_service.py index 5a47632d..9a6578b3 100644 --- a/tests/test_obsidian_service.py +++ b/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: From 4b0a3ea95366357f183f137bc59760b47f5d204d Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Mon, 24 Aug 2026 05:50:25 -0400 Subject: [PATCH 11/12] fix(import): propagate late manifest truncation --- engraphis/obsidian_import.py | 10 ++++++---- tests/test_obsidian_service.py | 24 ++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/engraphis/obsidian_import.py b/engraphis/obsidian_import.py index 222da940..30d6e859 100644 --- a/engraphis/obsidian_import.py +++ b/engraphis/obsidian_import.py @@ -409,11 +409,13 @@ def import_scan( # 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 @@ -1050,7 +1052,7 @@ def _all_source_items(self, *, vault_id: str, 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, manifest_complete = self._all_source_items( vault_id=vault_id, @@ -1060,7 +1062,7 @@ def _reconcile_links( # 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 [] + return [], False memory_by_path = { str(item["relative_path"]): str(item["memory_id"]) for item in items if item.get("memory_id") @@ -1228,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.""" diff --git a/tests/test_obsidian_service.py b/tests/test_obsidian_service.py index 9a6578b3..d11ec80d 100644 --- a/tests/test_obsidian_service.py +++ b/tests/test_obsidian_service.py @@ -124,6 +124,30 @@ def _truncated(_self, *, vault_id, states=None): 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: From 73cc897cf9d5a38c789e14f9991bc72ff56d9ec2 Mon Sep 17 00:00:00 2001 From: Coding-Dev-Tools Date: Mon, 24 Aug 2026 05:54:44 -0400 Subject: [PATCH 12/12] fix(entitlement): release lock during denial probes --- engraphis/routes/v2_api.py | 23 ++++++++++-------- tests/test_hosted_plan_resolution.py | 35 +++++++++++++++++++++++++++- 2 files changed, 47 insertions(+), 11 deletions(-) diff --git a/engraphis/routes/v2_api.py b/engraphis/routes/v2_api.py index fabe9b80..a562a90b 100644 --- a/engraphis/routes/v2_api.py +++ b/engraphis/routes/v2_api.py @@ -3124,18 +3124,21 @@ def _mark_authoritative_denial() -> None: """Make an authoritative cloud denial visible before persistence starts.""" global _authoritative_denial_at, _denied_state_digests + denial_at = time.time() 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_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() - _denied_state_digests = {} - _denied_state_digests = { - source: _persisted_state_digest(source) for source in ("session", "cloud") - } + 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: diff --git a/tests/test_hosted_plan_resolution.py b/tests/test_hosted_plan_resolution.py index 67dfb05c..b139a153 100644 --- a/tests/test_hosted_plan_resolution.py +++ b/tests/test_hosted_plan_resolution.py @@ -1144,7 +1144,7 @@ def test_denial_guard_publishes_before_blocked_digest_probe(monkeypatch) -> None def _blocked_digest(_source): entered.set() assert v2_api._AUTHORITATIVE_DENIAL_PENDING.is_set() - assert v2_api._denied_state_digests == {} + assert v2_api._denied_state_digests == {"session": None, "cloud": None} assert release.wait(timeout=5.0) return "before-denial" @@ -1160,6 +1160,39 @@ def _blocked_digest(_source): 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."""