CR-LIC-03a: HARD node-cap enforcement — the wall now bites

graqle/cli/commands/scan.py

@@ -1912,11 +1912,52 @@ def _show_cloud_onboarding_nudge(graph_data: dict, graqle_dir: Path) -> None:
         pass
 
 
+def _enforce_node_cap_before_write(graph_data: dict, graqle_dir: Path) -> None:
+    """HARD-enforce the node cap BEFORE the graph is persisted (CR-LIC-03a, ADR-245).
+
+    Called immediately BEFORE the graph file is written, so a genuine block PREVENTS
+    the graph from growing past the cap (not a cosmetic error after the write — that
+    was the sentinel BLOCKER-1). Enforcement is ON by default with a grace window.
+
+    Only a real NodeCapExceeded stops the write; any metering FAILURE (bad file, etc.)
+    is swallowed so a metering hiccup never breaks a legitimate scan. Reads are never
+    on this path. Unlimited tiers / below-cap / within-grace never raise.
+    """
+    import typer
+
+    from graqle.licensing.meter import NodeCapExceeded
+
+    try:
+        from graqle.licensing.limits import resolve_limits
+        from graqle.licensing.manager import _get_manager
+        from graqle.licensing.meter import UsageMeter
+
+        limits = resolve_limits(_get_manager().license)
+        meter = UsageMeter(graqle_dir)
+        # record() advances the HWM + grace stamp (never raises); enforce() decides.
+        reading = meter.record(len(graph_data.get("nodes", [])), limits)
+        meter.enforce(reading, limits)  # raises NodeCapExceeded once grace elapses
+    except NodeCapExceeded as exc:
+        console.print(
+            f"\n[bold red]Node cap reached.[/bold red] {exc}\n"
+            "[dim]Your existing graph still works — this only blocks growing it "
+            "further on the current plan.[/dim]\n"
+            "  Upgrade: [bold cyan]https://graqle.com/pricing[/bold cyan]  "
+            "(or register free for 1,000 nodes: https://graqle.com/signup)"
+        )
+        raise typer.Exit(1) from None
+    except Exception as exc:
+        import logging
+
+        logging.getLogger("graqle.cli.scan").debug("cap enforcement skipped: %s", exc)
+
+
 def _record_usage_meter(graph_data: dict, graqle_dir: Path) -> None:
-    """Record scan size against tier-derived limits — warn-only (CR-LIC-01).
+    """Print the WARN/AT_CAP usage line after a scan (telemetry/messaging only).
 
-    Prints one line at WARN/AT_CAP (suppressed in non-TTY). Never raises:
-    metering failure must never break a scan. Enforcement is CR-LIC-03.
+    Never raises — the HARD block is done separately by
+    :func:`_enforce_node_cap_before_write` BEFORE the graph is persisted. This runs
+    AFTER the write purely to show the status line + upgrade nudge.
     """
     try:
         from graqle.licensing.limits import resolve_limits
@@ -1943,8 +1984,7 @@ def _record_usage_meter(graph_data: dict, graqle_dir: Path) -> None:
             console.print(
                 f"\n[yellow]Graph reached the current plan's node cap "
                 f"({reading.node_count:,}/{reading.max_nodes:,}).[/yellow] "
-                "[dim]Nothing is blocked or deleted. Plans: "
-                "[cyan]https://graqle.com/pricing[/cyan][/dim]"
+                "[dim]Plans: [cyan]https://graqle.com/pricing[/cyan][/dim]"
             )
     except Exception as exc:
         import logging
@@ -2159,6 +2199,10 @@ def _scan_repo_impl(
         except Exception:
             pass  # Non-blocking — never fail a scan for backup
 
+    # CR-LIC-03a: hard-enforce the node cap BEFORE persisting the graph, so a real
+    # block PREVENTS growth past the cap (not a cosmetic post-write error).
+    _enforce_node_cap_before_write(data, Path(".graqle"))
+
     _write_with_lock(str(out_path), json.dumps(data, indent=2, default=str))
 
     # Print summary
@@ -2716,6 +2760,8 @@ def _save_graph_data(graph_path: Path, nodes: dict, edges: dict) -> None:
 
     content = json.dumps(data, indent=2, default=str)
     graph_path.parent.mkdir(parents=True, exist_ok=True)
+    # CR-LIC-03a: enforce the node cap BEFORE writing (prevents growth past cap).
+    _enforce_node_cap_before_write(data, Path(".graqle"))
     _write_with_lock(str(graph_path), content)
     console.print(f"[dim]Saved graph:[/dim] {graph_path} ({len(nodes)} nodes, {len(edges)} edges)")
 

graqle/licensing/meter.py

@@ -33,15 +33,73 @@
 
 logger = logging.getLogger("graqle.licensing.meter")
 
-__all__ = ["ENFORCE_ENV", "METER_FILENAME", "MeterStatus", "MeterReading", "UsageMeter"]
+__all__ = [
+    "ENFORCE_ENV",
+    "GRACE_ENV",
+    "METER_FILENAME",
+    "MeterStatus",
+    "MeterReading",
+    "NodeCapExceeded",
+    "UsageMeter",
+    "enforcement_enabled",
+]
 
-# Reserved for CR-LIC-03 enforcement. This module never reads it.
+# CR-LIC-03a hard enforcement (ADR-245). Enforcement is ON BY DEFAULT — this env
+# var is an OPT-OUT for our own CI/tests only (set to a falsy value). "User must
+# set a flag to be enforced" would be a trivial bypass and is deliberately rejected.
 ENFORCE_ENV = "GRAQLE_ENFORCE_CAPS"
+# Grace window (days) after a project FIRST hits the cap: the meter warns during
+# grace, then blocks. Gives an existing user time to upgrade instead of a scan
+# breaking on the day enforcement lands. Overridable for tests.
+GRACE_ENV = "GRAQLE_ENFORCE_GRACE_DAYS"
+_DEFAULT_GRACE_DAYS = 7
+
+_FALSY = frozenset({"0", "false", "no", "off", ""})
 
 METER_FILENAME = "meter.json"
 _SCHEMA_VERSION = 1
 
 
+class NodeCapExceeded(Exception):
+    """Raised when a WRITE would push the graph past the plan's node cap.
+
+    Carries the numbers so the CLI can render a clean upgrade message. Reads are
+    NEVER blocked — only operations that would GROW the graph past the cap.
+    """
+
+    def __init__(self, node_count: int, max_nodes: int, plan_source: str) -> None:
+        super().__init__(
+            f"graph would reach {node_count:,} nodes, over the "
+            f"{max_nodes:,}-node cap on the current plan ({plan_source})"
+        )
+        self.node_count = node_count
+        self.max_nodes = max_nodes
+        self.plan_source = plan_source
+
+
+def enforcement_enabled() -> bool:
+    """True unless explicitly opted out via GRACE_ENV=falsy. Default: ON.
+
+    ADR-245: enforcement is on by default; the env var only DISABLES it (for CI).
+    """
+    raw = os.environ.get(ENFORCE_ENV)
+    if raw is None:
+        return True  # default ON
+    return raw.strip().lower() not in _FALSY
+
+
+def _grace_days() -> int:
+    raw = os.environ.get(GRACE_ENV)
+    if raw:
+        try:
+            v = int(raw)
+            if v >= 0:
+                return v
+        except (TypeError, ValueError):
+            pass
+    return _DEFAULT_GRACE_DAYS
+
+
 class MeterStatus(str, Enum):
     OK = "OK"
     WARN = "WARN"          # >= warn threshold (80% of cap by default)
@@ -140,15 +198,20 @@ def record(self, node_count: int, limits: EffectiveLimits) -> MeterReading:
         # a scan that changes nothing must not dirty a committed working tree
         # (shared-repo diff churn — CR-LIC-01 pre-merge debate, point 3).
         if hwm > prev_hwm or not self._path.exists():
-            self._store(
-                {
-                    "schema_version": _SCHEMA_VERSION,
-                    "high_water_mark": hwm,
-                    "last_node_count": count,
-                    "limit_source": limits.source,
-                    "updated_at": datetime.now(timezone.utc).isoformat(),
-                }
-            )
+            payload = {
+                "schema_version": _SCHEMA_VERSION,
+                "high_water_mark": hwm,
+                "last_node_count": count,
+                "limit_source": limits.source,
+                "updated_at": datetime.now(timezone.utc).isoformat(),
+            }
+            # CR-LIC-03a: preserve the enforcement grace stamp across HWM writes —
+            # a scan that advances the HWM must NOT reset the grace clock, or a user
+            # could dodge the block forever by growing the graph one node at a time.
+            existing = self._load()
+            if existing.get("cap_first_hit_at"):
+                payload["cap_first_hit_at"] = existing["cap_first_hit_at"]
+            self._store(payload)
             self._ensure_gitignore()
 
         if limits.unlimited:
@@ -168,3 +231,68 @@ def record(self, node_count: int, limits: EffectiveLimits) -> MeterReading:
             high_water_mark=hwm,
             max_nodes=limits.max_nodes,
         )
+
+    def enforce(self, reading: MeterReading, limits: EffectiveLimits) -> None:
+        """Hard-enforce the node cap on a WRITE (CR-LIC-03a, ADR-245). May raise.
+
+        Contract:
+          - Enforcement is ON BY DEFAULT (``enforcement_enabled()``); ``record`` itself
+            still never raises — callers opt into hard-block by calling this.
+          - Never blocks reads or uncapped tiers (unlimited → no-op).
+          - Grace window: the FIRST time a project is AT_CAP we stamp
+            ``cap_first_hit_at`` and WARN (do not raise); once the grace window has
+            elapsed, subsequent AT_CAP writes raise :class:`NodeCapExceeded`.
+          - High-water-mark based (via ``reading``): deleting nodes cannot dodge it.
+
+        Raises
+        ------
+        NodeCapExceeded
+            when enforcement is on, the tier is capped, the graph is AT_CAP, and the
+            grace window has elapsed.
+        """
+        if not enforcement_enabled():
+            return
+        if limits.unlimited or limits.max_nodes is None:
+            return
+        if reading.status is not MeterStatus.AT_CAP:
+            return
+
+        now = datetime.now(timezone.utc)
+        data = self._load()
+        first_hit_raw = data.get("cap_first_hit_at")
+
+        if not first_hit_raw:
+            # First time at cap → start the grace window, warn (no raise this scan).
+            data["cap_first_hit_at"] = now.isoformat()
+            self._store(data)
+            self._ensure_gitignore()
+            return
+
+        try:
+            first_hit = datetime.fromisoformat(str(first_hit_raw))
+            if first_hit.tzinfo is None:
+                first_hit = first_hit.replace(tzinfo=timezone.utc)
+        except (TypeError, ValueError):
+            # Corrupt stamp → treat NOW as the start of grace (fail-open on grace,
+            # not on the block: we never harden retroactively off a bad timestamp).
+            # WARN so a surgical "reset only the grace clock" tamper leaves an audit
+            # trail (MAJOR-1) — the accepted local-bypass class; server-side later.
+            logger.warning(
+                "meter: cap_first_hit_at was unparseable (%r) — restarting the grace "
+                "window. If this recurs, the local meter may be being tampered with.",
+                first_hit_raw,
+            )
+            data["cap_first_hit_at"] = now.isoformat()
+            self._store(data)
+            return
+
+        grace_days = _grace_days()
+        elapsed_days = (now - first_hit).total_seconds() / 86400.0
+        if elapsed_days < grace_days:
+            return  # still within grace — warn-only handled by the caller
+
+        raise NodeCapExceeded(
+            node_count=reading.node_count,
+            max_nodes=limits.max_nodes,
+            plan_source=limits.source,
+        )