W3 reasoning-quota wall — public port (ADR-245 Decision 8)

graqle/benchmarks/benchmark_runner.py

@@ -559,7 +559,8 @@ async def _run_one(
         graph.set_default_backend(self.backend)
 
         start = time.perf_counter()
-        result: ReasoningResult = await graph.areason(question)
+        # internal=True (W3, ADR-245): benchmarking is tooling, not user reasoning.
+        result: ReasoningResult = await graph.areason(question, internal=True)
         latency = (time.perf_counter() - start) * 1000
 
         # Extract governance stats from metadata

graqle/benchmarks/run_multigov_v2.py

@@ -318,7 +318,8 @@ async def run_v2_benchmark(
             graph._orchestrator = orchestrator
 
             cg_start = time.perf_counter()
-            result: ReasoningResult = await graph.areason(q.question)
+            # internal=True (W3, ADR-245): benchmarking is tooling, not user reasoning.
+            result: ReasoningResult = await graph.areason(q.question, internal=True)
             cg_latency = (time.perf_counter() - cg_start) * 1000
 
             cg_f1 = f1_score(result.answer, q.expected_answer)

graqle/benchmarks/run_multigov_v3.py

@@ -389,7 +389,8 @@ async def run_v3_benchmark(
             graph._orchestrator = orchestrator
 
             cg_start = time.perf_counter()
-            result: ReasoningResult = await graph.areason(q.question)
+            # internal=True (W3, ADR-245): benchmarking is tooling, not user reasoning.
+            result: ReasoningResult = await graph.areason(q.question, internal=True)
             cg_latency = (time.perf_counter() - cg_start) * 1000
 
             # Get actual cost delta

graqle/cli/commands/debate.py

@@ -111,6 +111,22 @@ def debate(
     )
     orchestrator = DebateOrchestrator(debate_config, pool, cost_gate)
 
+    # W3 (ADR-245): a debate is real (multi-panelist) reasoning that never reaches
+    # graph.areason(), so the wall must be applied here explicitly — otherwise
+    # `graq debate` is a free unlimited-reasoning surface.
+    from graqle.licensing.reasoning_gate import check_reasoning_quota
+    from graqle.licensing.reasoning_quota import ReasoningQuotaExceeded
+
+    try:
+        check_reasoning_quota()
+    except ReasoningQuotaExceeded as exc:
+        console.print(
+            f"\n[bold red]Reasoning quota reached.[/bold red] {exc}\n"
+            "  Upgrade for unlimited reasoning: "
+            "[bold cyan]https://graqle.com/pricing[/bold cyan]"
+        )
+        raise typer.Exit(1) from None
+
     trace = asyncio.run(orchestrator.run(query))
 
     # Display debate results

graqle/cli/main.py

@@ -14,6 +14,13 @@
 import sys
 
 import typer
+from contextlib import contextmanager
+
+# W3 (ADR-245): the wall itself lives at the SDK primitive; the CLI only needs the
+# exception type so it can render an upgrade CTA instead of a traceback.
+from graqle.licensing.reasoning_quota import (
+    ReasoningQuotaExceeded as _ReasoningQuotaExceeded,
+)
 
 from graqle.cli.commands.activate import activate_command
 from graqle.cli.commands.audit import audit_command
@@ -147,6 +154,30 @@ def main(
 console = create_console()
 
 
+@contextmanager
+def _reasoning_quota_cta():
+    """W3 (ADR-245): render the upgrade CTA when the reasoning wall fires.
+
+    This does NOT meter — the wall lives at the SDK primitive (``graph.areason``) so
+    that CLI, MCP, chat and api are all covered by one gate (ADR-245 Decision 8; the
+    CLI-only placement was the PR #316 bypass). Metering here too would double-charge
+    every CLI reasoning call. This wrapper only turns the primitive's exception into a
+    friendly CTA + clean exit instead of a traceback.
+    """
+    try:
+        yield
+    except _ReasoningQuotaExceeded as exc:
+        console.print(
+            f"\n[bold red]Reasoning quota reached.[/bold red] {exc}\n"
+            "[dim]The free tier includes a monthly reasoning allowance; your graph and "
+            "all reads still work.[/dim]\n"
+            "  Upgrade for unlimited reasoning: "
+            "[bold cyan]https://graqle.com/pricing[/bold cyan]  "
+            "(or register free: https://graqle.com/signup)"
+        )
+        raise typer.Exit(1) from None
+
+
 # CR-004 PR-004c (sentinel 1C dedup + regex correction):
 # ANSI CSI / OSC strip helpers used by the degraded-reasoning warning
 # in run() and reason(). Extracted to module scope so the regexes are
@@ -442,10 +473,13 @@ def run(
     if coordinator:
         graph.config.coordinator.enabled = True
 
-    # Run reasoning with selected protocol
-    result = asyncio.run(
-        graph.areason(query, max_rounds=max_rounds, strategy=strategy)
-    )
+    # W3 (ADR-245): the quota wall fires inside graph.areason(); this only turns it
+    # into an upgrade CTA + clean exit instead of a traceback.
+    with _reasoning_quota_cta():
+        # Run reasoning with selected protocol
+        result = asyncio.run(
+            graph.areason(query, max_rounds=max_rounds, strategy=strategy)
+        )
 
     # CR-004 PR-004c: probe + degraded-reasoning yellow warning.
     # The probe never raises (PR-004a 3-deep defence); outer try is
@@ -966,12 +1000,14 @@ def bench(
 
     console.print(f"[cyan]Benchmarking {queries} queries, {max_rounds} max rounds...[/cyan]")
 
+    # W3 (ADR-245): benchmarking is tooling, not user reasoning — internal=True so a
+    # 50-query bench never burns (or hard-blocks on) a free contributor's quota.
     # Run queries sequentially with fail-fast: stop on first backend error
     start = time.perf_counter()
     results = []
     for i, q in enumerate(test_queries):
         try:
-            r = asyncio.run(graph.areason(q, max_rounds=max_rounds))
+            r = asyncio.run(graph.areason(q, max_rounds=max_rounds, internal=True))
             results.append(r)
         except Exception as e:
             console.print(f"\n[red]Query {i + 1}/{queries} failed: {e}[/red]")
@@ -2663,12 +2699,16 @@ def reason(
             raise typer.Exit(1)
 
         console.print(f"Batch: [green]{len(queries)} queries[/green] (max_concurrent={max_concurrent})")
-        results = asyncio.run(
-            graph.areason_batch(
-                queries, max_rounds=max_rounds, strategy=strategy,
-                max_concurrent=max_concurrent,
+        # W3 (ADR-245): areason_batch fans out to areason once per query, so an
+        # N-query batch costs N quota units — it consumes N LLM invocations, exactly
+        # like N separate calls. The CTA wrapper renders the wall if it fires mid-batch.
+        with _reasoning_quota_cta():
+            results = asyncio.run(
+                graph.areason_batch(
+                    queries, max_rounds=max_rounds, strategy=strategy,
+                    max_concurrent=max_concurrent,
+                )
             )
-        )
 
         total_cost = sum(r.cost_usd for r in results)
         total_latency = sum(r.latency_ms for r in results)
@@ -2713,9 +2753,12 @@ def reason(
     # ── Single query mode ─────────────────────────────────────────────
     console.print(f"Query: [green]{query}[/green]")
 
-    result = asyncio.run(
-        graph.areason(query, max_rounds=max_rounds, strategy=strategy)
-    )
+    # W3 (ADR-245): the quota wall fires inside graph.areason(); this only turns it
+    # into an upgrade CTA + clean exit instead of a traceback.
+    with _reasoning_quota_cta():
+        result = asyncio.run(
+            graph.areason(query, max_rounds=max_rounds, strategy=strategy)
+        )
 
     # CR-004 PR-004c: probe + degraded-reasoning yellow warning. Same
     # pattern as the run() command above. Skipped silently in JSON

graqle/core/graph.py

@@ -1853,6 +1853,11 @@ def reason(
                      (Claude Code, Cursor, Codex) for query enhancement.
             task_type: Optional task type for routing (e.g. "reason", "context").
         """
+        # NOT internal: this wrapper delegates to areason(), which is the single
+        # billing wall and charges exactly once for this call. Passing internal=True
+        # here would EXEMPT the call outright (internal means "never metered", not
+        # "already metered") — making the whole synchronous reason() API free and
+        # recreating the PR #316 bypass in mirror image. Delegation charges once.
         return asyncio.run(
             self.areason(
                 query,
@@ -1873,6 +1878,7 @@ async def areason(
         node_ids: list[str] | None = None,
         context: Any = None,
         task_type: str | None = None,
+        internal: bool = False,
     ) -> ReasoningResult:
         """Run async reasoning query — the core entry point.
 
@@ -1881,7 +1887,19 @@ async def areason(
                      (Claude Code, Cursor, Codex) for query enhancement
                      before PCST activation .
             task_type: Optional task type for task-based model routing (v0.22).
+            internal: True when this reasoning is invoked *by* another metered
+                      action (a governance scan, a benchmark runner). Exempt from
+                      the W3 reasoning quota so one user action is never charged
+                      twice. See ADR-245 Decision 8.
         """
+        # W3 (ADR-245): the reasoning-quota wall. This is the SDK primitive every
+        # surface funnels through — CLI, MCP `graq_reason`, chat agent, api.py — so
+        # gating here covers all of them. Raises ReasoningQuotaExceeded for a FREE
+        # user over the monthly cap; fail-open on any metering fault.
+        from graqle.licensing.reasoning_gate import check_reasoning_quota
+
+        check_reasoning_quota(internal=internal)
+
         from graqle.orchestration.orchestrator import Orchestrator
 
         max_rounds = max_rounds or self.config.orchestration.max_rounds
@@ -2081,13 +2099,25 @@ async def areason_stream(
         strategy: str | None = None,
         node_ids: list[str] | None = None,
         context: Any = None,
+        internal: bool = False,
     ) -> AsyncIterator:
         """Stream reasoning results as they become available.
 
         Usage:
             async for chunk in graph.areason_stream("query"):
                 print(chunk.content)
+
+        Args:
+            internal: True when invoked by another metered action — exempt from
+                      the W3 reasoning quota (ADR-245 Decision 8).
         """
+        # W3 (ADR-245): streaming builds its own StreamingOrchestrator and never
+        # reaches areason(), so it needs the wall explicitly — without this it is a
+        # free unlimited-reasoning surface (the PR #316 failure mode).
+        from graqle.licensing.reasoning_gate import check_reasoning_quota
+
+        check_reasoning_quota(internal=internal)
+
         from graqle.orchestration.streaming import StreamingOrchestrator
 
         max_rounds = max_rounds or self.config.orchestration.max_rounds

graqle/licensing/reasoning_gate.py

@@ -0,0 +1,148 @@
+"""Reasoning-quota middleware (W3, ADR-245 Decision 8) — one wall, every surface.
+
+The quota logic lives in :mod:`graqle.licensing.reasoning_quota`; this module is the
+single *enforcement point* that the SDK reasoning primitive calls on entry.
+
+Why a middleware and not a CLI hook: the first W3 attempt (PR #316, CLOSED) placed the
+wall in ``graqle/cli/main.py`` only. The MCP server's ``graq_reason`` tool, the chat
+agent, and ``api.py`` all reach :meth:`graqle.core.graph.Graqle.areason` *without* going
+through the CLI — so a primary surface had free unlimited reasoning. The rule that came
+out of that review: **a wall at one surface is not a wall**. Enforce at the primitive.
+
+``Graqle.areason()`` is the sole chokepoint — sync ``reason()`` delegates to it, and
+``areason_batch()`` fans out to it per query — so gating ``areason`` alone covers every
+caller exactly once. ``areason_batch`` deliberately does NOT gate: doing so would charge
+a 5-query batch six times.
+
+Exemption: ``internal=True`` — reasoning invoked *by* another metered action (a W2
+PR-Guardian scan, a benchmark runner). Prevents W2/W3 double-charging one user action.
+``internal=True`` is the ONLY exemption. A CI environment variable is deliberately not
+honoured — ``export CI=true`` is self-attested and would be a one-line bypass of the
+whole wall (sentinel BLOCKER-2). A 50-query benchmark stays free because the benchmark
+runners pass ``internal=True`` in code, which an end user cannot set from outside.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import sys
+from pathlib import Path
+
+logger = logging.getLogger("graqle.licensing.reasoning_gate")
+
+__all__ = [
+    "QUOTA_DIR_ENV",
+    "check_reasoning_quota",
+    "quota_exempt",
+    "resolve_quota_dir",
+]
+
+# Test/bench override for the project-local quota directory.
+QUOTA_DIR_ENV = "GRAQLE_QUOTA_DIR"
+
+_DEFAULT_QUOTA_DIR = ".graqle"
+
+def _under_pytest() -> bool:
+    """True only when pytest is genuinely loaded in THIS process.
+
+    Deliberately ``sys.modules``, not an environment variable. The first attempt at
+    this guard used ``PYTEST_CURRENT_TEST``, which is itself a self-attested env var —
+    it merely replaced one bypass with another, and was proven forgeable:
+    ``PYTEST_CURRENT_TEST=fake::call GRAQLE_QUOTA_DIR=/tmp/x`` resolved to ``/tmp/x``.
+    An attacker cannot import pytest into a process that never imported it, so
+    ``sys.modules`` closes what an env check cannot.
+
+    It is also the only correct answer for TIMING. ``PYTEST_CURRENT_TEST`` is set
+    per-test (setup/call/teardown) and is absent during module import, collection, and
+    session-scoped fixtures — measured: at import time it is ``False`` while
+    ``"pytest" in sys.modules`` is already ``True``. With the env check, a test that
+    resolved the quota dir at import or in a session fixture would have silently fallen
+    through to the developer's REAL ``./.graqle`` and mutated their actual quota count.
+    """
+    return "pytest" in sys.modules or "_pytest" in sys.modules
+
+
+def resolve_quota_dir() -> Path:
+    """Project-local ``.graqle`` directory holding the quota file.
+
+    ``GRAQLE_QUOTA_DIR`` relocates the meter, but ONLY under pytest.
+
+    ⚠️ It is deliberately INERT in normal runs. Honouring it unconditionally made it a
+    one-line unlimited-reasoning bypass of this very wall: a free user at the cap who
+    exported ``GRAQLE_QUOTA_DIR=$(mktemp -d)`` got a fresh empty counter on every
+    invocation. That is categorically worse than the local-bypass class ADR-245 already
+    accepts (deleting ``.graqle/reasoning_quota.json``), on three counts — it persists
+    via a shell profile so it is set once and never repeated, it is non-destructive so
+    nothing looks tampered with, and it needs no repeat action.
+
+    It also contradicted this module's own rule: :func:`quota_exempt` refuses to honour
+    ``CI=true`` precisely because "self-attested env exemption" is banned by ADR-245
+    Decision 8 rule 3. An env var that relocates the counter is the same bypass wearing
+    a different hat. Tests still need to redirect the meter, so the override survives —
+    scoped to a real pytest process, which a production run is not.
+    """
+    override = os.environ.get(QUOTA_DIR_ENV)
+    if override and override.strip() and _under_pytest():
+        return Path(override)
+    return Path(_DEFAULT_QUOTA_DIR)
+
+
+def quota_exempt(internal: bool) -> bool:
+    """True when this reasoning call must be neither counted nor blocked.
+
+    ``internal`` is the ONLY exemption. A CI environment variable is deliberately
+    NOT honoured: ``export CI=true`` is self-attested and would be a one-line
+    unlimited-reasoning bypass of the whole wall. Tooling that legitimately must not
+    be metered (the benchmark runners, governance sub-calls) passes ``internal=True``
+    at the callsite, which is code the user cannot set from the outside.
+    """
+    return bool(internal)
+
+
+def check_reasoning_quota(*, internal: bool = False) -> None:
+    """Enforce the monthly reasoning quota for ONE reasoning invocation.
+
+    Parameters
+    ----------
+    internal:
+        True when this reasoning is invoked by another metered action (W2 scan,
+        benchmark runner, or an internal SDK self-call). Exempt: not counted, never
+        blocked.
+
+    Raises
+    ------
+    ReasoningQuotaExceeded
+        when enforcement is on, the verified tier is FREE, and the monthly quota is
+        spent. This is the ONLY exception this function propagates.
+
+    Any other failure — a missing module, an unreadable quota file, a licensing
+    error — is swallowed. Reasoning must never break because metering hiccuped.
+    """
+    if quota_exempt(internal):
+        return
+
+    # Imported lazily: graqle.core.graph imports this module, so a module-level import
+    # of anything that reaches back into core would be circular.
+    try:
+        from graqle.licensing.reasoning_quota import ReasoningQuota, ReasoningQuotaExceeded
+    except Exception as exc:  # noqa: BLE001 — metering unavailable → fail open
+        # WARNING: an un-importable meter means the wall is off for this call.
+        logger.warning(
+            "reasoning quota module unavailable (%s) — allowing this call", exc
+        )
+        return
+
+    try:
+        ReasoningQuota(resolve_quota_dir()).check_and_record(internal=False)
+    except ReasoningQuotaExceeded:
+        # Must escape BEFORE the broad except below — this is the wall doing its job,
+        # not a metering fault. Swallowing it here would silently disable W3.
+        raise
+    except Exception as exc:  # noqa: BLE001 — never break reasoning on a meter fault
+        # WARNING, not debug: this branch grants un-metered reasoning. It must be
+        # visible in logs, otherwise a corrupt quota file or a permissions problem
+        # silently disables the wall with no operational signal.
+        logger.warning(
+            "reasoning quota could not be enforced (%s) — allowing this call", exc
+        )