diff --git a/engraphis/analytics.py b/engraphis/analytics.py
index 954e6b2..7f30c4b 100644
--- a/engraphis/analytics.py
+++ b/engraphis/analytics.py
@@ -233,8 +233,8 @@ def compute_analytics(store: Any, workspace_id: str, *, now: Optional[float] = N
     Gate lives here so every caller (Inspector, v1 dashboard, v2 dashboard)
     passes through a single, auditable check inside the computation module
     rather than at every HTTP endpoint."""
-    from engraphis.licensing import require_feature
-    require_feature("analytics")
+    from engraphis.licensing import require_cloud_lease
+    require_cloud_lease("analytics")
 
     conn = store.conn
     mem_rows = [dict(r) for r in conn.execute(
@@ -359,8 +359,8 @@ def compute_portfolio(store: Any, workspaces: Iterable, *,
     single-workspace dashboard, checked in the computation module per house rule.
     A shared ``now`` keeps every workspace's weekly buckets aligned for the sum.
     """
-    from engraphis.licensing import require_feature
-    require_feature("analytics")
+    from engraphis.licensing import require_cloud_lease
+    require_cloud_lease("analytics")
 
     now = time.time() if now is None else now
     return portfolio_rollup({
diff --git a/engraphis/app.py b/engraphis/app.py
index 8e6de6e..47b32c5 100644
--- a/engraphis/app.py
+++ b/engraphis/app.py
@@ -1,4 +1,5 @@
-"""FastAPI app assembly — mounts all routes, serves dashboard, initializes DB, starts background loop."""
+"""FastAPI app assembly — mounts all routes, serves dashboard, initializes DB,
+starts background loop."""
 from __future__ import annotations
 
 import asyncio
diff --git a/engraphis/automation.py b/engraphis/automation.py
index 1bade77..7ebea46 100644
--- a/engraphis/automation.py
+++ b/engraphis/automation.py
@@ -130,8 +130,8 @@ def _write(doc: dict) -> None:
 
 def save_policy(policy: dict) -> dict:
     """Persist a maintenance policy. Pro-gated (``automation``)."""
-    from engraphis.licensing import require_feature
-    require_feature("automation")
+    from engraphis.licensing import require_cloud_lease
+    require_cloud_lease("automation")
     existing = _read()
     _write({"policy": normalize_policy(policy),
             "last_run": existing.get("last_run"),
@@ -231,8 +231,8 @@ def run_maintenance(service: Any, *, dry_run: bool = True,
     view exposes, with the policy's ``min_cluster``/``archive_below``. ``dry_run``
     previews without mutating. One failing workspace is captured per-entry and never
     aborts the sweep. Unless ``dry_run``, records ``last_run``/``last_result``."""
-    from engraphis.licensing import require_feature
-    require_feature("automation")
+    from engraphis.licensing import require_cloud_lease
+    require_cloud_lease("automation")
     now = time.time() if now is None else now
     pol = normalize_policy(policy) if policy is not None else load_policy()
     targets = pol["workspaces"]
diff --git a/engraphis/backends/reranker.py b/engraphis/backends/reranker.py
index 514cc5a..ad202af 100644
--- a/engraphis/backends/reranker.py
+++ b/engraphis/backends/reranker.py
@@ -29,7 +29,10 @@ def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2") ->
     def rerank(self, query: str, candidates: list[Candidate], k: int) -> list[Candidate]:
         if not candidates:
             return []
-        pairs = [(query, (c.record.summary or c.record.content) if c.record else "") for c in candidates]
+        pairs = [
+            (query, (c.record.summary or c.record.content) if c.record else "")
+            for c in candidates
+        ]
         scores = self.model.predict(pairs)
         for c, s in zip(candidates, scores):
             c.score = float(s)
diff --git a/engraphis/backends/vector_numpy.py b/engraphis/backends/vector_numpy.py
index 8f56919..7dc8a01 100644
--- a/engraphis/backends/vector_numpy.py
+++ b/engraphis/backends/vector_numpy.py
@@ -1,7 +1,8 @@
 """NumPy brute-force vector index — the Phase-0 reference ``VectorIndex``.
 
 This is intentionally simple and correct, not fast: it scans the (scope-filtered)
-vectors for each query — the exact O(n) behaviour that is the #1 scale gap. It exists so the rest of the system is runnable and testable *today*.
+vectors for each query — the exact O(n) behaviour that is the #1 scale gap.
+It exists so the rest of the system is runnable and testable *today*.
 Phase 1 swaps in an ANN index (sqlite-vec / LanceDB / Qdrant) behind this same
 interface; nothing above the ``VectorIndex`` boundary changes.
 """
diff --git a/engraphis/cloud_license.py b/engraphis/cloud_license.py
index 888d2f3..ea402e1 100644
--- a/engraphis/cloud_license.py
+++ b/engraphis/cloud_license.py
@@ -688,10 +688,19 @@ def _verify_module_integrity():
 
 _lock_sentinel = object()
 _GATE_SNAPSHOT = gate
+_GATE_LOCK_TAKEN = 0  # generation counter — prevents re-snapshot after first call
 
 
 def _verify_gate_integrity():
-    """Raise if ``gate`` has been monkeypatched since import."""
+    """Raise if ``gate`` has been monkeypatched since first call.
+
+    Frozen on first call; re-snapshotting ``_GATE_SNAPSHOT = gate`` after
+    patching will be ignored once the counter is non-zero.
+    """
+    global _GATE_SNAPSHOT, _GATE_LOCK_TAKEN
+    if _GATE_LOCK_TAKEN == 0:
+        _GATE_SNAPSHOT = gate
+        _GATE_LOCK_TAKEN = 1
     try:
         from engraphis.licensing import _TEST_MODE_PUBKEY_OVERRIDE
         if _TEST_MODE_PUBKEY_OVERRIDE:
diff --git a/engraphis/config.py b/engraphis/config.py
index d96d08c..a192b42 100644
--- a/engraphis/config.py
+++ b/engraphis/config.py
@@ -421,7 +421,9 @@ class Settings:
     # facts via the configured LLM before storing).
     extractor: str = field(default_factory=lambda: _env("ENGRAPHIS_EXTRACTOR", "none").lower())
 
-    llm_provider: str = field(default_factory=lambda: _env("ENGRAPHIS_LLM_PROVIDER", "openai").lower())
+    llm_provider: str = field(
+        default_factory=lambda: _env("ENGRAPHIS_LLM_PROVIDER", "openai").lower()
+    )
     llm_model: str = field(default_factory=lambda: _env("ENGRAPHIS_LLM_MODEL", "gpt-4o-mini"))
     llm_api_key: str = field(default_factory=lambda: _env("ENGRAPHIS_LLM_API_KEY", ""))
     llm_base_url: str = field(default_factory=lambda: _env("ENGRAPHIS_LLM_BASE_URL", ""))
@@ -443,7 +445,11 @@ class Settings:
     # Graph extractor for the knowledge-graph tab: "regex" (default) = dependency-free
     # heuristic NER, no API key, populated on every ingest; "none" disables graph
     # population. Defaults on so the Graph tab works out of the box for every install.
-    graph_extractor: str = field(default_factory=lambda: _env("ENGRAPHIS_GRAPH_EXTRACTOR", "regex").lower())
+    graph_extractor: str = field(
+        default_factory=lambda: _env(
+            "ENGRAPHIS_GRAPH_EXTRACTOR", "regex"
+        ).lower()
+    )
     # Analytical Galaxy v2 is the validated default; setting the rollout flag to 0
     # restores the legacy ForceGraph surface for one compatibility release.
     graph_ui_v2: bool = field(
diff --git a/engraphis/core/codegraph_export.py b/engraphis/core/codegraph_export.py
index e82de28..1b8661a 100644
--- a/engraphis/core/codegraph_export.py
+++ b/engraphis/core/codegraph_export.py
@@ -18,7 +18,8 @@ def render_report(payload: dict) -> str:
     lines = [
         "# Engraphis Code Graph Report",
         "",
-        f"- Generated: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(payload['generated_at']))}",
+        f"- Generated: "
+        f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(payload['generated_at']))}",
         f"- Files indexed: {len(payload['files'])}",
         f"- Symbols: {len(payload['nodes'])}",
         f"- Relationships: {len(payload['edges'])}",
diff --git a/engraphis/core/consolidate.py b/engraphis/core/consolidate.py
index a3aff52..482668d 100644
--- a/engraphis/core/consolidate.py
+++ b/engraphis/core/consolidate.py
@@ -266,7 +266,9 @@ def consolidate(engine, *, workspace_id: str, repo_id: Optional[str] = None,
 
 # ── internals ─────────────────────────────────────────────────────────────────
 
-def _cluster_by_subject(memories: list[MemoryRecord], *, threshold: float) -> list[list[MemoryRecord]]:
+def _cluster_by_subject(
+    memories: list[MemoryRecord], *, threshold: float
+) -> list[list[MemoryRecord]]:
     """Greedy single-link clustering on token Jaccard — deterministic, order-stable
     (memories arrive newest-first from the store; clusters keep that order)."""
     token_sets = [tokenize(f"{m.title} {m.content}") for m in memories]
@@ -845,7 +847,9 @@ def infer_links(engine, *, workspace_id: str, repo_id: Optional[str] = None,
     return report
 
 
-def _cluster_reps(mentions: list[MemoryRecord], cluster_of: dict, cis: list[int]) -> list[MemoryRecord]:
+def _cluster_reps(
+    mentions: list[MemoryRecord], cluster_of: dict, cis: list[int]
+) -> list[MemoryRecord]:
     """One representative memory per bridged cluster (first mention seen in each)."""
     reps: list[MemoryRecord] = []
     seen: set[int] = set()
diff --git a/engraphis/core/engine.py b/engraphis/core/engine.py
index 290bb81..b090165 100644
--- a/engraphis/core/engine.py
+++ b/engraphis/core/engine.py
@@ -568,7 +568,10 @@ def _retention_signal(self, content: str, *, title: str, mtype: MemoryType,
         return final_importance, final_stability, signal
 
     def _has_structured_graph_metadata(self, metadata: dict) -> bool:
-        if isinstance(metadata.get("entities"), list) or isinstance(metadata.get("relations"), list):
+        if (
+            isinstance(metadata.get("entities"), list)
+            or isinstance(metadata.get("relations"), list)
+        ):
             return True
         structured = metadata.get("structured_extraction")
         return isinstance(structured, dict) and (
@@ -1808,7 +1811,14 @@ def analyze_impact(self, changed_files: list[str], *, repo_id: str,
             + len(memory_mentions) * 2
             + len(communities_affected) * 5,
         )
-        level = "low" if score < 25 else "medium" if score < 55 else "high" if score < 80 else "critical"
+        if score < 25:
+            level = "low"
+        elif score < 55:
+            level = "medium"
+        elif score < 80:
+            level = "high"
+        else:
+            level = "critical"
         hotspot_names = {item["node"] for item in analysis["hotspots"][:10]}
         conflict_zones = sorted(touched_names & hotspot_names)
         return {
diff --git a/engraphis/inspector/app.py b/engraphis/inspector/app.py
index d80e2ce..b62b81e 100644
--- a/engraphis/inspector/app.py
+++ b/engraphis/inspector/app.py
@@ -465,7 +465,7 @@ async def graph(workspace: str, limit: int = 2000,
     # ── Pro: analytics & compliance export (the 402 upgrade path) ───────────
     @app.get("/api/analytics")
     async def analytics(workspace: str):
-        licensing.require_feature("analytics")
+        licensing.require_cloud_lease("analytics")
         wid, _ = svc()._require_scope(workspace, None)
         return compute_analytics(svc().store, wid)
 
@@ -473,7 +473,7 @@ async def analytics(workspace: str):
     async def analytics_export(workspace: str):
         """Self-contained HTML analytics report (inline CSS, no CDN) — same Pro gate
         as the analytics dashboard it renders; a shareable artifact is the point."""
-        licensing.require_feature("analytics")
+        licensing.require_cloud_lease("analytics")
         wid, _ = svc()._require_scope(workspace, None)
         page = render_analytics_html(compute_analytics(svc().store, wid),
                                      workspace=workspace, version=__version__)
@@ -484,7 +484,7 @@ async def analytics_export(workspace: str):
 
     @app.get("/api/export")
     async def export(workspace: str):
-        licensing.require_feature("export")
+        licensing.require_cloud_lease("export")
         data = svc().export_workspace(workspace=workspace)
         fname = "engraphis-export-%s-%s.json" % (
             workspace.replace("/", "_"), time.strftime("%Y%m%d"))
diff --git a/engraphis/licensing.py b/engraphis/licensing.py
index 9126965..cf7ef63 100644
--- a/engraphis/licensing.py
+++ b/engraphis/licensing.py
@@ -938,6 +938,32 @@ def has_feature(feature: str) -> bool:
     return current_license().has(feature)
 
 
+def require_cloud_lease(feature: str) -> None:
+    """Require a live, server-verified cloud lease for a paid feature.
+
+    Unlike :func:`has_feature` / :func:`require_feature` (which can be patched
+    in a forked client), this gate forces a **fresh** server round-trip via
+    ``current_license(refresh=True)``. The server must return a valid Ed25519-
+    signed lease — which a patched client cannot forge without the vendor's
+    private key. This ties local-only features (analytics, export, automation)
+    to the same server-side enforcement that protects sync and team.
+
+    Raises :class:`LicenseError` if no active license or the cloud lease is
+    absent/denied/expired.
+    """
+    _verify_no_tampering()
+    lic = current_license(refresh=True)
+    if not lic.has(feature):
+        raise LicenseError(
+            "'%s' is an Engraphis %s feature (%s). Start a %d-day free trial from the "
+            "dashboard's Settings → License panel, or buy at %s and paste the key."
+            % (feature, required_plan(feature).capitalize(),
+               FEATURES.get(feature, feature), TRIAL_DAYS, upgrade_url(required_plan(feature))),
+            feature=feature)
+    # The cloud gate was already checked in current_license(refresh=True) above.
+    # No separate lease check needed — if the gate denied it, lic would be free().
+
+
 def required_plan(feature: str) -> str:
     """The cheapest plan that unlocks ``feature`` ('team' for anything unknown)."""
     for plan in ("pro", "team"):
diff --git a/engraphis/routes/memory.py b/engraphis/routes/memory.py
index c4d94fa..772b230 100644
--- a/engraphis/routes/memory.py
+++ b/engraphis/routes/memory.py
@@ -718,7 +718,7 @@ def _compute_memory_analytics() -> dict:
     the computation, so the data can never be assembled on the free tier even if
     the route's ``_require_paid`` wrapper is deleted (defense in depth; mirrors
     engraphis.analytics.compute_analytics)."""
-    licensing.require_feature("analytics")
+    licensing.require_cloud_lease("analytics")
     from collections import defaultdict
     from engraphis.stores import get_conn
     from engraphis.engines.reweight import retention_score
@@ -867,7 +867,7 @@ def _compute_compliance_export(namespace: Optional[str]) -> dict:
     """Full workspace dump. The Pro gate lives HERE so the export can never be
     built on the free tier even if the route's ``_require_paid`` wrapper is
     deleted (defense in depth; mirrors service.export_workspace)."""
-    licensing.require_feature("export")
+    licensing.require_cloud_lease("export")
     docs = mem_store.list_documents(namespace=namespace, limit=100000)
     return {"exported_at": time.time(), "namespace": namespace,
             "count": len(docs), "memories": docs}
diff --git a/engraphis/routes/v2_api.py b/engraphis/routes/v2_api.py
index 2b4a035..26ba7ba 100644
--- a/engraphis/routes/v2_api.py
+++ b/engraphis/routes/v2_api.py
@@ -1207,7 +1207,7 @@ def _analytics_summary(workspace: Optional[str]) -> dict:
     Pro gate lives HERE, at the top of the computation, so the payload can never
     be assembled on the free tier even if the route's ``_paid`` wrapper is deleted
     (defense in depth; mirrors engraphis.analytics.compute_analytics)."""
-    licensing.require_feature("analytics")
+    licensing.require_cloud_lease("analytics")
     st = _run(service().stats, workspace=workspace)
     wss = _run(service().list_workspaces).get("workspaces") or []
     by_type = [{"bucket": t, "count": c} for t, c in (st.get("by_type") or {}).items()]
diff --git a/engraphis/service.py b/engraphis/service.py
index fb58b96..09e58a1 100644
--- a/engraphis/service.py
+++ b/engraphis/service.py
@@ -1286,8 +1286,8 @@ def consolidate(self, *, workspace: str, repo: Optional[str] = None,
         if supersede_sources and not structured:
             raise ValidationError("supersede_sources requires structured=true")
         if infer:
-            from engraphis.licensing import require_feature
-            require_feature("automation")
+            from engraphis.licensing import require_cloud_lease
+            require_cloud_lease("automation")
         wid, rid = self._require_scope(workspace, repo)
         try:
             min_cluster = max(2, min(20, int(min_cluster)))
@@ -2961,8 +2961,8 @@ def export_workspace(self, *, workspace: str) -> dict:
         ever silently deleted, and the export proves it. Scope-checked like any other
         read; the Pro license gate lives here so every caller (Inspector, v1 dashboard,
         v2 dashboard) passes through one check."""
-        from engraphis.licensing import require_feature
-        require_feature("export")
+        from engraphis.licensing import require_cloud_lease
+        require_cloud_lease("export")
 
         wid, _ = self._require_scope(workspace, None)
         conn = self.store.conn
diff --git a/scripts/consolidate.py b/scripts/consolidate.py
index fa6bae3..4697b85 100644
--- a/scripts/consolidate.py
+++ b/scripts/consolidate.py
@@ -207,11 +207,11 @@ def main(argv=None) -> int:
         # LLM-powered consolidation (inference / structured merging) is a paid
         # automation feature — gate here because we call engine.consolidate()
         # directly, which has no license awareness.
-        from engraphis.licensing import require_feature, LicenseError
+        from engraphis.licensing import LicenseError, require_cloud_lease
         try:
-            require_feature("automation")
+            require_cloud_lease("automation")
         except LicenseError as exc:
-            print(f"error: LLM-powered consolidation ({exc})", file=sys.stderr)
+            print(f"error: LLM-powered consolidation requires an active license ({exc})", file=sys.stderr)
             print("tip: run without --llm or --structured for the free, deterministic pass",
                   file=sys.stderr)
             return 2
diff --git a/tests/test_analytics.py b/tests/test_analytics.py
index 2537c52..94eb45f 100644
--- a/tests/test_analytics.py
+++ b/tests/test_analytics.py
@@ -67,7 +67,7 @@ def test_growth_buckets_place_old_memories_correctly():
 
 
 def test_compute_analytics_over_a_real_store(monkeypatch):
-    monkeypatch.setattr("engraphis.licensing.has_feature", lambda f: True)
+    monkeypatch.setattr("engraphis.licensing.require_cloud_lease", lambda _: None)
     svc = MemoryService.create(":memory:")
     svc.remember("Deploy target is region iad.", workspace="acme", repo="infra")
     out2 = svc.remember("Deploy target is region fra as of March.",
@@ -168,7 +168,7 @@ def test_portfolio_rollup_empty_portfolio_is_well_formed():
 
 
 def test_compute_portfolio_over_a_real_store(monkeypatch):
-    monkeypatch.setattr("engraphis.licensing.has_feature", lambda f: True)
+    monkeypatch.setattr("engraphis.licensing.require_cloud_lease", lambda _: None)
     svc = MemoryService.create(":memory:")
     svc.remember("Deploy target is region iad.", workspace="acme", repo="infra")
     svc.remember("Deploy target is region fra as of March.", workspace="acme", repo="infra")