chore(sync): mirror backend .py + Dockerfile to Space (hf-sync-backend)

Automated backend sync from szl-holdings/a11oy main via hf-sync-backend.
Updated (differed from the Space): Dockerfile, a11oy_code_orchestrator.py, serve.py, szl_energy_sovereign.py
Deleted (gone from the repo + Dockerfile COPY set): a11oy_react_core.py

Keeps the Space-built backend (serve.py + the Dockerfile-COPY'd .py
modules) identical to GitHub main so the Space never rebuilds from a
stale backend, new endpoints don't 404 there, and orphaned modules
removed from the repo don't linger in the Space tree.

Dockerfile

@@ -96,7 +96,7 @@ COPY _vendor_blobs.py szl_v4_fleet.py operator_shell_v4.py szl_bridge.py szl_bri
 COPY szl_formula_wiring.py a11oy_code_engine.py a11oy_code.py a11oy_seismic.py szl_warhacker_real.py szl_warhacker_demos.py NOTICE_warhacker_demos.txt szl_llm_registry.py szl_elite_console.py szl_alloy_models.py szl_scaling.py szl_allodial.py szl_entanglement.py szl_neuroplasticity.py szl_chain_of_title.py szl_sovereign_compute.py ./
 # Energy/heart/engine/revenue/harvest organ modules: present in repo but were absent
 # from every COPY line -> guarded imports threw ModuleNotFoundError -> dark 404 surfaces.
-COPY szl_energy_budget.py szl_energy_provenance.py szl_heart_blood.py szl_engine_status.py szl_backend_hardening.py revenue_endpoints.py a11oy_harvest_endpoints.py ./
+COPY szl_energy_budget.py szl_energy_sovereign.py szl_energy_provenance.py szl_heart_blood.py szl_engine_status.py szl_backend_hardening.py revenue_endpoints.py a11oy_harvest_endpoints.py ./
 # Agentic-PINN + physical-bounds mesh (pure-stdlib sibling of szl_energy_budget; serves
 # /api/a11oy/v1/pinn/*). MUST be COPY'd or serve.py's guarded import falls back to a stub
 # (merged-but-not-live) in the HF image. The optional on-metal artifacts it reads
@@ -300,14 +300,6 @@ COPY web/operator.html ./web/operator.html
 COPY web/fleet-c2.html ./web/fleet-c2.html
 COPY web/living-anatomy.html ./web/living-anatomy.html
 
-# ADDITIVE (LANE A AGENTIC CORE — Dev A, 2026-06-14): resumable ReAct agent with
-# signed receipt boundaries, SqliteSaver checkpointing, Reflexion, Generative-
-# Agents memory scoring, Letta tiering, Voyager skill library. Served at
-# /agent-loop (+ /a11oy/agent-loop) via serve.py _ptg_serve from /app/web/.
-# Per-file COPY (this Dockerfile uses no COPY . .). 0 CDN; shared engines from
-# /static/shared. Co-Authored-By: Perplexity Computer Agent <agent@perplexity.ai>
-COPY web/agent-loop.html ./web/agent-loop.html
-
 # ADDITIVE (Cross-Harness Receipt Bridge — Hermes + OpenClaw; 2026-06-01, Yachay /
 # Perplexity Computer Agent; closeout PR superseding #198 runtime files). serve.py
 # already imports szl_bridge + a11oy_v4_agent and calls .register(app) BEFORE the
@@ -606,13 +598,6 @@ ENV A11OY_ALLOY_GGUF=/app/models/qwen2.5-coder-0.5b-instruct-q4_k_m.gguf
 # ---------------------------------------------------------------------------
 COPY a11oy_live_feeds.py a11oy_signing_key.py a11oy_dev1_endpoints.py a11oy_vertical_feeds.py a11oy_deva_feeds.py a11oy_devb_endpoints.py a11oy_amaru_feeds.py szl_governance_gateway.py szl_abacus_verify.py szl_decision_uncertainty.py szl_gor_audit.py szl_sovereign_search.py szl_consensus_clusters.py szl_mission_ledger.py szl_budget_router.py szl_wave910_proofs.py szl_evidence_research.py szl_uds_fleet.py szl_readiness.py szl_quantum_bio.py szl_mosaic_governance.py ./
 COPY szl_unified_formulas.py szl_cuas_formulas.py szl_contracting.py szl_bounties.py szl_putnam.py szl_connectors_serve.py szl_connector_mcp.py szl_conjecture_factory.py ./
-
-# ADDITIVE (LANE A AGENTIC CORE — Dev A, 2026-06-14): the resumable ReAct agent
-# core module. serve.py imports it try/except-guarded and calls .register(app,
-# "a11oy", _a11oy_sign_receipt, verify_fn=_a11oy_loop_verify, pub_pem_fn=...).
-# Without this per-file COPY the import fails and the /api/a11oy/v1/agent/react/*
-# endpoints fall through to the SPA shell. Per-file COPY (no COPY . . here).
-COPY a11oy_react_core.py ./
 COPY live_snapshots/ ./live_snapshots/
 
 # ADDITIVE (Investor-WOW Layer, 2026-06-08, Dev1): a11oy_dev1_endpoints.py exposes

a11oy_code_orchestrator.py

@@ -103,6 +103,14 @@ try:
     import energy_signal as _energy_signal  # PR #356 honest power-window feed
 except Exception:  # pragma: no cover - absent until #356 merges; default grid
     _energy_signal = None
+# Sovereign-energy instrumentation (Lane C): reads REAL J/token + carbon from the
+# on-box vLLM /metrics ONLY when the live sovereign probe shows gpu_reachable, and
+# splices joules_consumed + carbon_g_co2eq into EVERY signed turn receipt — honestly
+# labeled MEASURED (real fresh exporter) or ROADMAP (no meter yet -> None, never faked).
+try:
+    import szl_energy_sovereign as _energy_sovereign  # Lane C J/token + carbon receipt fields
+except Exception:  # pragma: no cover - absent until the module merges; degrade honestly
+    _energy_sovereign = None
 
 # ---------------------------------------------------------------------------
 # Configuration
@@ -949,18 +957,39 @@ def _emit_turn_receipt(text: str, model_used: str, is_local: bool,
     try:
         prov = _turn_energy_provenance(is_local)
         out_bytes = len((text or "").encode("utf-8"))
+        # Lane C: splice REAL J/token energy + carbon into EVERY signed receipt. The
+        # helper reads the on-box vLLM /metrics ONLY when the live sovereign probe shows
+        # gpu_reachable; otherwise it returns honest ROADMAP (joules_consumed=None). It
+        # NEVER raises and NEVER fabricates a number (no meter -> no number).
+        energy_fields = {}
+        if _energy_sovereign is not None:
+            try:
+                energy_fields = _energy_sovereign.energy_fields_for_receipt() or {}
+            except Exception:  # noqa: BLE001 - energy probe must never break a turn
+                energy_fields = {"joules_consumed": None, "carbon_g_co2eq": None,
+                                 "energy_label": "ROADMAP"}
+        extra = {
+            "turn": True,
+            "model": model_used,
+            "stub": bool(stub),
+            "sovereign_local": bool(is_local),
+            "window": prov.get("window", "normal"),
+            "signal_provider": prov.get("signal_provider", "unknown"),
+            # joules_consumed + carbon_g_co2eq on every receipt (MEASURED via a live GPU
+            # exporter, else honest ROADMAP/None). joules_honesty is decided by
+            # szl_joules_truth, never by a flag.
+            "joules_consumed": energy_fields.get("joules_consumed"),
+            "carbon_g_co2eq": energy_fields.get("carbon_g_co2eq"),
+            "joules_per_token": energy_fields.get("joules_per_token"),
+            "carbon_g_co2eq_per_token": energy_fields.get("carbon_g_co2eq_per_token"),
+            "energy_label": energy_fields.get("energy_label", "ROADMAP"),
+            "joules_honesty": energy_fields.get("joules_honesty", "sample"),
+        }
         receipt = _energy_budget.track_task(
             output=text or "",
             energy_source=prov.get("energy_source", "grid"),
             joules_est=0.0,  # SAMPLE: no meter wired (joules_est_label carries it)
-            extra={
-                "turn": True,
-                "model": model_used,
-                "stub": bool(stub),
-                "sovereign_local": bool(is_local),
-                "window": prov.get("window", "normal"),
-                "signal_provider": prov.get("signal_provider", "unknown"),
-            },
+            extra=extra,
         )
         # The Bekenstein gate is the PROVEN F19/TH6 inequality (shannon<=n*8);
         # an honest receipt is always within_bound. Flag (never raise) if not.
@@ -981,6 +1010,11 @@ def _emit_turn_receipt(text: str, model_used: str, is_local: bool,
             "window": prov.get("window", "normal"),
             "joules_est": receipt.get("joules_est"),
             "joules_est_label": receipt.get("joules_est_label"),
+            # Lane C live J/token energy + carbon (MEASURED via on-box exporter / ROADMAP).
+            "joules_consumed": extra.get("joules_consumed"),
+            "carbon_g_co2eq": extra.get("carbon_g_co2eq"),
+            "energy_label": extra.get("energy_label"),
+            "joules_honesty": extra.get("joules_honesty"),
         })
         return receipt
     except Exception as exc:  # noqa: BLE001 - NEVER break a turn over a receipt

a11oy_react_core.py

@@ -1,805 +0,0 @@
-# -*- coding: utf-8 -*-
-# ============================================================================
-# a11oy_react_core.py  —  LANE A  AGENTIC CORE  (Dev A, SZL Holdings)
-# ----------------------------------------------------------------------------
-# A REAL, resumable ReAct execution graph (Thought -> Action -> Observation)
-# where EACH node transition is a SIGNED receipt boundary, plus:
-#   * SqliteSaver-style checkpointing  (crash mid-run -> /resume continues)
-#   * Reflexion inner loop             (NL reflection prepended next activation)
-#   * Generative-Agents memory scoring score(m)=a_rec*g^dt + a_imp*imp + a_rel*cos
-#   * Letta-style memory tiering       (working in-context + archival vector)
-#   * Voyager skill library            (admit a recipe ONLY after a passing receipt)
-#
-# Honest engineering (DOCTRINE v11):
-#   - The signer is the HOST app's REAL in-image ECDSA-P256 DSSE signer
-#     (_a11oy_sign_receipt), passed in via register(); we NEVER fabricate a
-#     signature. verify_fn re-verifies against /cosign.pub.
-#   - The vector store is LOCAL (sqlite + numpy, hashing-trie embeddings) — 0
-#     external CDN/service. Embeddings are a deterministic local feature hash
-#     (labelled HEURISTIC), not a remote model, so retrieval is reproducible
-#     offline. Scores are surfaced honestly with their components.
-#   - Trust / coverage framing, never bare "confidence %".
-#   - Routes are inserted at position 0 (Starlette Route) so they beat the SPA
-#     catch-all, mirroring szl_agentic_loop.
-#   - Endpoints live under the FREE /api/a11oy/v1/agent/react/* sub-namespace
-#     (run/resume/trace/checkpoints) so we do NOT collide with the existing
-#     /run, /tools, /verify-chain, /governance-standards, /_diag, /loop.
-#
-# Co-Authored-By: Perplexity Computer Agent <agent@perplexity.ai>
-# Signed-off-by: Stephen P. Lutar Jr. <stephenlutar2@gmail.com>
-# ============================================================================
-from __future__ import annotations
-
-import hashlib
-import json
-import math
-import os
-import re
-import sqlite3
-import threading
-import time
-import uuid
-from datetime import datetime, timezone
-
-# ---------------------------------------------------------------------------
-# Storage: a single local sqlite DB under /tmp (ephemeral per container, which
-# is the honest reality of a HF Space). All five subsystems persist here so a
-# crash mid-run can resume from the last committed checkpoint within the life
-# of the container. Labelled accordingly in the UI.
-# ---------------------------------------------------------------------------
-_DB_PATH = os.environ.get("A11OY_REACT_DB", "/tmp/a11oy_react_core.sqlite3")
-_LOCK = threading.RLock()
-_EMBED_DIM = 64  # local hashing-trick embedding dimension
-
-
-def _now_iso() -> str:
-    return datetime.now(timezone.utc).isoformat()
-
-
-def _now_epoch() -> float:
-    return time.time()
-
-
-def _sha(obj) -> str:
-    return hashlib.sha256(
-        json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8")
-    ).hexdigest()
-
-
-def _conn() -> sqlite3.Connection:
-    c = sqlite3.connect(_DB_PATH, timeout=30, check_same_thread=False)
-    c.row_factory = sqlite3.Row
-    c.execute("PRAGMA journal_mode=WAL;")
-    return c
-
-
-def _init_db() -> None:
-    with _LOCK, _conn() as c:
-        c.executescript(
-            """
-            CREATE TABLE IF NOT EXISTS runs (
-                run_id TEXT PRIMARY KEY,
-                goal TEXT, status TEXT, max_steps INTEGER,
-                step INTEGER, prev_hash TEXT, final_hash TEXT,
-                reflection TEXT, created_at TEXT, updated_at TEXT
-            );
-            CREATE TABLE IF NOT EXISTS receipts (
-                run_id TEXT, seq INTEGER, node TEXT, body TEXT,
-                prev_hash TEXT, hash TEXT, envelope TEXT, ts TEXT,
-                PRIMARY KEY (run_id, seq)
-            );
-            CREATE TABLE IF NOT EXISTS checkpoints (
-                run_id TEXT, step INTEGER, state TEXT, prev_hash TEXT,
-                ts TEXT, PRIMARY KEY (run_id, step)
-            );
-            CREATE TABLE IF NOT EXISTS memory (
-                mem_id TEXT PRIMARY KEY, run_id TEXT, tier TEXT, kind TEXT,
-                text TEXT, importance REAL, created_at REAL, last_access REAL,
-                embedding TEXT
-            );
-            CREATE TABLE IF NOT EXISTS skills (
-                skill_id TEXT PRIMARY KEY, name TEXT, recipe TEXT,
-                receipt_hash TEXT, receipt_verified INTEGER, embedding TEXT,
-                created_at TEXT, uses INTEGER
-            );
-            CREATE TABLE IF NOT EXISTS reflections (
-                run_id TEXT, idx INTEGER, text TEXT, ts TEXT,
-                PRIMARY KEY (run_id, idx)
-            );
-            """
-        )
-
-
-# ---------------------------------------------------------------------------
-# LOCAL embedding — deterministic hashing-trick bag-of-tokens, L2 normalised.
-# This is NOT a learned model; it is a reproducible local feature hash so that
-# cosine similarity is meaningful for lexical overlap WITHOUT any network call.
-# Labelled HEURISTIC everywhere it surfaces.
-# ---------------------------------------------------------------------------
-_TOK = re.compile(r"[a-z0-9]+")
-
-
-def _embed(text: str) -> list:
-    vec = [0.0] * _EMBED_DIM
-    toks = _TOK.findall((text or "").lower())
-    for t in toks:
-        h = int(hashlib.md5(t.encode()).hexdigest(), 16)
-        idx = h % _EMBED_DIM
-        sign = 1.0 if (h >> 8) & 1 else -1.0
-        vec[idx] += sign
-    norm = math.sqrt(sum(v * v for v in vec)) or 1.0
-    return [v / norm for v in vec]
-
-
-def _cos(a: list, b: list) -> float:
-    if not a or not b or len(a) != len(b):
-        return 0.0
-    return max(-1.0, min(1.0, sum(x * y for x, y in zip(a, b))))
-
-
-def _importance_heuristic(text: str) -> float:
-    """Local importance proxy in [0,1] (LLM-scored 1-10 in the paper; here a
-    transparent HEURISTIC: longer, decision/goal-bearing text scores higher).
-    Surfaced honestly as HEURISTIC, never claimed as an LLM judgement."""
-    t = (text or "").lower()
-    score = min(1.0, len(t) / 240.0)
-    for kw, w in (("goal", 0.2), ("decision", 0.2), ("fail", 0.25),
-                  ("error", 0.25), ("reflect", 0.2), ("verified", 0.15),
-                  ("receipt", 0.1)):
-        if kw in t:
-            score = min(1.0, score + w)
-    return round(score, 4)
-
-
-# ---------------------------------------------------------------------------
-# Generative-Agents retrieval score:
-#   score(m) = a_rec * gamma^dt_hours + a_imp * imp(m) + a_rel * cos(q, m)
-# gamma ~ 0.995 / hour (arXiv 2304.03442). Components surfaced honestly.
-# ---------------------------------------------------------------------------
-_GAMMA = 0.995          # recency decay per hour
-_A_REC = 1.0
-_A_IMP = 1.0
-_A_REL = 1.0
-
-
-def _score_memory(row, q_emb: list, now_epoch: float) -> dict:
-    dt_hours = max(0.0, (now_epoch - float(row["last_access"])) / 3600.0)
-    recency = _GAMMA ** dt_hours
-    imp = float(row["importance"])
-    try:
-        emb = json.loads(row["embedding"])
-    except Exception:
-        emb = []
-    rel = _cos(q_emb, emb)
-    rel01 = (rel + 1.0) / 2.0  # map cosine [-1,1] -> [0,1] for the weighted sum
-    total = _A_REC * recency + _A_IMP * imp + _A_REL * rel01
-    return {
-        "mem_id": row["mem_id"], "tier": row["tier"], "kind": row["kind"],
-        "text": row["text"],
-        "score": round(total, 6),
-        "components": {
-            "recency_gamma_dt": round(recency, 6),
-            "delta_t_hours": round(dt_hours, 4),
-            "importance": round(imp, 4),
-            "relevance_cos": round(rel, 6),
-            "relevance_0_1": round(rel01, 6),
-        },
-        "weights": {"alpha_recency": _A_REC, "alpha_importance": _A_IMP,
-                    "alpha_relevance": _A_REL, "gamma_per_hour": _GAMMA},
-        "label": "HEURISTIC",  # local embeddings + heuristic importance
-    }
-
-
-def _mem_add(run_id: str, tier: str, kind: str, text: str,
-             importance=None) -> str:
-    mem_id = "mem_" + uuid.uuid4().hex[:12]
-    now = _now_epoch()
-    imp = _importance_heuristic(text) if importance is None else float(importance)
-    emb = _embed(text)
-    with _LOCK, _conn() as c:
-        c.execute(
-            "INSERT INTO memory(mem_id,run_id,tier,kind,text,importance,"
-            "created_at,last_access,embedding) VALUES(?,?,?,?,?,?,?,?,?)",
-            (mem_id, run_id, tier, kind, text, imp, now, now, json.dumps(emb)),
-        )
-    return mem_id
-
-
-def _mem_retrieve(query: str, top_k: int = 5, tier=None) -> list:
-    q_emb = _embed(query)
-    now = _now_epoch()
-    with _LOCK, _conn() as c:
-        if tier:
-            rows = c.execute("SELECT * FROM memory WHERE tier=?", (tier,)).fetchall()
-        else:
-            rows = c.execute("SELECT * FROM memory").fetchall()
-    scored = [_score_memory(r, q_emb, now) for r in rows]
-    scored.sort(key=lambda s: s["score"], reverse=True)
-    top = scored[:top_k]
-    # honest "access" bump: retrieved memories refresh their recency clock
-    if top:
-        ids = [s["mem_id"] for s in top]
-        with _LOCK, _conn() as c:
-            c.executemany("UPDATE memory SET last_access=? WHERE mem_id=?",
-                          [(now, i) for i in ids])
-    return top
-
-
-# ---------------------------------------------------------------------------
-# Letta-style memory tiering. "working" = in-context (small, fast); "archival"
-# = vector store (large, searched). The agent self-manages via tool calls
-# memory_append / memory_search / memory_promote that the ReAct loop can emit.
-# ---------------------------------------------------------------------------
-_WORKING_CAP = 8  # in-context working-memory item cap (paging boundary)
-
-
-def _working_snapshot(run_id: str) -> list:
-    with _LOCK, _conn() as c:
-        rows = c.execute(
-            "SELECT * FROM memory WHERE run_id=? AND tier='working' "
-            "ORDER BY last_access DESC LIMIT ?", (run_id, _WORKING_CAP)
-        ).fetchall()
-    return [{"mem_id": r["mem_id"], "kind": r["kind"], "text": r["text"],
-             "importance": r["importance"]} for r in rows]
-
-
-def _page_out_if_full(run_id: str) -> list:
-    """Letta/MemGPT paging: when working memory exceeds the in-context cap, the
-    LEAST-recently-accessed working items are promoted (paged out) to archival
-    so the in-context window stays bounded. Returns the paged-out mem_ids."""
-    paged = []
-    with _LOCK, _conn() as c:
-        rows = c.execute(
-            "SELECT mem_id FROM memory WHERE run_id=? AND tier='working' "
-            "ORDER BY last_access ASC", (run_id,)
-        ).fetchall()
-        if len(rows) > _WORKING_CAP:
-            overflow = rows[: len(rows) - _WORKING_CAP]
-            for r in overflow:
-                c.execute("UPDATE memory SET tier='archival' WHERE mem_id=?",
-                          (r["mem_id"],))
-                paged.append(r["mem_id"])
-    return paged
-
-
-# ---------------------------------------------------------------------------
-# Voyager skill library. A tool-recipe is ADMITTED to the library ONLY after a
-# verified execution receipt (DSSE envelope verified by the host verify_fn).
-# Recipes are indexed by local embedding so the agent can retrieve a relevant
-# prior recipe. We NEVER admit a recipe whose receipt fails verification.
-# ---------------------------------------------------------------------------
-def _skill_admit(name: str, recipe: str, receipt_hash: str,
-                 receipt_verified: bool) -> dict:
-    if not receipt_verified:
-        return {"admitted": False,
-                "reason": "REJECTED — execution receipt did not verify; Voyager "
-                          "admission requires a passing signed receipt.",
-                "receipt_hash": receipt_hash}
-    skill_id = "skill_" + uuid.uuid4().hex[:12]
-    emb = _embed(name + " " + recipe)
-    with _LOCK, _conn() as c:
-        c.execute(
-            "INSERT INTO skills(skill_id,name,recipe,receipt_hash,"
-            "receipt_verified,embedding,created_at,uses) VALUES(?,?,?,?,?,?,?,0)",
-            (skill_id, name, recipe, receipt_hash, 1, json.dumps(emb), _now_iso()),
-        )
-    return {"admitted": True, "skill_id": skill_id, "name": name,
-            "receipt_hash": receipt_hash,
-            "reason": "ADMITTED — backed by a verified execution receipt."}
-
-
-def _skill_search(query: str, top_k: int = 5) -> list:
-    q_emb = _embed(query)
-    with _LOCK, _conn() as c:
-        rows = c.execute("SELECT * FROM skills").fetchall()
-    out = []
-    for r in rows:
-        try:
-            emb = json.loads(r["embedding"])
-        except Exception:
-            emb = []
-        out.append({"skill_id": r["skill_id"], "name": r["name"],
-                    "recipe": r["recipe"], "receipt_hash": r["receipt_hash"],
-                    "receipt_verified": bool(r["receipt_verified"]),
-                    "uses": r["uses"],
-                    "similarity": round(_cos(q_emb, emb), 6), "label": "LIVE"})
-    out.sort(key=lambda s: s["similarity"], reverse=True)
-    return out[:top_k]
-
-
-# ---------------------------------------------------------------------------
-# ReAct execution graph (arXiv 2210.03629). Nodes: THOUGHT -> ACTION ->
-# OBSERVATION, looping until a terminal ANSWER or max_steps. EACH node
-# transition is committed as a hash-chained receipt AND wrapped in a DSSE
-# envelope by the host signer (sign_fn). A SqliteSaver-style checkpoint is
-# written after every node so a crash resumes from the last committed step.
-#
-# The "model call" is routed through a small, deterministic in-image policy
-# (the host a11oy inference path is the production target; we keep the loop's
-# model-calls inside the app and label the planner HEURISTIC so we never fake a
-# model number — the GRAPH, RECEIPTS, CHECKPOINTING and RESUME are all REAL).
-# ---------------------------------------------------------------------------
-
-# Tool registry: small, real, deterministic tools the ReAct agent can call.
-def _tool_calc(arg: str) -> str:
-    expr = re.sub(r"[^0-9+\-*/(). ]", "", arg or "")
-    if not expr.strip():
-        return "ERR: empty expression"
-    try:
-        # safe arithmetic only (chars already filtered); no names/builtins
-        return str(eval(expr, {"__builtins__": {}}, {}))  # noqa: S307
-    except Exception as e:
-        return "ERR: %s" % type(e).__name__
-
-
-def _tool_memory_search(arg: str) -> str:
-    hits = _mem_retrieve(arg, top_k=3)
-    if not hits:
-        return "no memories"
-    return " | ".join("%s(score=%.3f)" % (h["text"][:48], h["score"]) for h in hits)
-
-
-def _tool_skill_search(arg: str) -> str:
-    hits = _skill_search(arg, top_k=3)
-    if not hits:
-        return "no skills"
-    return " | ".join("%s(sim=%.3f)" % (h["name"], h["similarity"]) for h in hits)
-
-
-def _tool_echo(arg: str) -> str:
-    return (arg or "")[:200]
-
-
-_TOOLS = {
-    "calc": _tool_calc,
-    "memory_search": _tool_memory_search,
-    "skill_search": _tool_skill_search,
-    "echo": _tool_echo,
-}
-
-
-def _plan_action(goal: str, scratch: list) -> dict:
-    """HEURISTIC planner (NOT a learned model — labelled HEURISTIC). Picks the
-    next ReAct action from the goal + scratchpad. Deterministic so the loop is
-    replayable and the demo is reproducible. The production target is the host
-    a11oy inference path; this keeps the agent loop's model-calls in-app."""
-    g = (goal or "").lower()
-    step = len(scratch)
-    # terminal: if we already produced an observation, answer.
-    last_obs = next((s for s in reversed(scratch) if s.get("node") == "OBSERVATION"), None)
-    if last_obs is not None:
-        return {"terminal": True,
-                "thought": "I have an observation; I can answer now.",
-                "answer": "Result: %s" % last_obs.get("observation", "")}
-    # arithmetic goal -> calc
-    if re.search(r"\d.*[+\-*/].*\d", g):
-        m = re.search(r"[-0-9+\-*/(). ]{3,}", goal)
-        arg = m.group(0).strip() if m else goal
-        return {"terminal": False, "tool": "calc", "tool_input": arg,
-                "thought": "This looks arithmetic; I will use the calc tool."}
-    if "memory" in g or "remember" in g or "recall" in g:
-        return {"terminal": False, "tool": "memory_search", "tool_input": goal,
-                "thought": "I should consult memory for this."}
-    if "skill" in g or "recipe" in g or "how do i" in g:
-        return {"terminal": False, "tool": "skill_search", "tool_input": goal,
-                "thought": "I should check the skill library."}
-    return {"terminal": False, "tool": "echo", "tool_input": goal,
-            "thought": "No specialised tool; I will restate and observe."}
-
-
-class _ReActEngine:
-    """Holds the host signer/verifier and runs / resumes graphs."""
-
-    def __init__(self, sign_fn, verify_fn, pub_pem_fn, ns="a11oy"):
-        self.sign_fn = sign_fn
-        self.verify_fn = verify_fn
-        self.pub_pem_fn = pub_pem_fn
-        self.ns = ns
-        _init_db()
-
-    # ---- receipt boundary: chain + DSSE sign every node transition ----
-    def _commit_receipt(self, run_id, seq, node, body, prev_hash, reflection=None):
-        rec_core = {"seq": seq, "node": node, "body": body, "prev_hash": prev_hash}
-        h = _sha(rec_core)
-        payload = {
-            "run_id": run_id, "seq": seq, "node": node, "body": body,
-            "prev_hash": prev_hash, "hash": h, "issuer": self.ns,
-            "issued_at": _now_iso(),
-            "reflection": reflection,  # Reflexion field on the receipt
-            "trust_status": "Conjecture 1 (advisory \u2014 NOT a proven oracle)",
-        }
-        try:
-            envelope = self.sign_fn(payload)
-        except Exception as e:
-            envelope = {"signed": False, "signatures": [],
-                        "honesty": "UNSIGNED \u2014 signer raised %s" % type(e).__name__,
-                        "payloadType": "application/vnd.szl.receipt+json"}
-        with _LOCK, _conn() as c:
-            c.execute(
-                "INSERT OR REPLACE INTO receipts(run_id,seq,node,body,prev_hash,"
-                "hash,envelope,ts) VALUES(?,?,?,?,?,?,?,?)",
-                (run_id, seq, node, json.dumps(body), prev_hash, h,
-                 json.dumps(envelope), _now_iso()),
-            )
-        return h, envelope
-
-    # ---- SqliteSaver-style checkpoint after every node ----
-    def _checkpoint(self, run_id, step, state, prev_hash):
-        with _LOCK, _conn() as c:
-            c.execute(
-                "INSERT OR REPLACE INTO checkpoints(run_id,step,state,prev_hash,ts)"
-                " VALUES(?,?,?,?,?)",
-                (run_id, step, json.dumps(state), prev_hash, _now_iso()),
-            )
-
-    def _load_run(self, run_id):
-        with _LOCK, _conn() as c:
-            r = c.execute("SELECT * FROM runs WHERE run_id=?", (run_id,)).fetchone()
-            cps = c.execute(
-                "SELECT * FROM checkpoints WHERE run_id=? ORDER BY step DESC LIMIT 1",
-                (run_id,)).fetchone()
-        return r, cps
-
-    def _prior_reflection(self, goal):
-        """Reflexion: prepend the most relevant prior reflection on activation."""
-        with _LOCK, _conn() as c:
-            rows = c.execute(
-                "SELECT text FROM reflections ORDER BY rowid DESC LIMIT 8").fetchall()
-        if not rows:
-            return None
-        # pick the reflection most lexically relevant to this goal
-        q = _embed(goal)
-        best, best_sim = None, -2.0
-        for r in rows:
-            sim = _cos(q, _embed(r["text"]))
-            if sim > best_sim:
-                best, best_sim = r["text"], sim
-        return best
-
-    # ---- run a (possibly partial) graph from a starting step ----
-    def _drive(self, run_id, goal, max_steps, start_step, prev_hash,
-               scratch, reflection, kill_after=None):
-        node_seq = start_step
-        status = "running"
-        steps_done = 0
-        terminal_answer = None
-        while node_seq // 3 < max_steps:
-            phase = node_seq % 3
-            cur_step = node_seq // 3
-            if phase == 0:  # THOUGHT
-                plan = _plan_action(goal, scratch)
-                scratch.append({"node": "THOUGHT", "step": cur_step,
-                                "thought": plan["thought"], "plan": plan})
-                body = {"step": cur_step, "thought": plan["thought"],
-                        "intended_tool": plan.get("tool"),
-                        "terminal": plan.get("terminal", False)}
-                prev_hash, _ = self._commit_receipt(run_id, node_seq, "THOUGHT",
-                                                    body, prev_hash, reflection)
-                if plan.get("terminal"):
-                    terminal_answer = plan.get("answer")
-                    status = "completed"
-                    self._checkpoint(run_id, node_seq + 1,
-                                     {"scratch": scratch, "answer": terminal_answer},
-                                     prev_hash)
-                    node_seq += 1
-                    break
-            elif phase == 1:  # ACTION
-                plan = scratch[-1]["plan"]
-                tool, arg = plan.get("tool", "echo"), plan.get("tool_input", "")
-                scratch.append({"node": "ACTION", "step": cur_step,
-                                "tool": tool, "tool_input": arg})
-                body = {"step": cur_step, "tool": tool, "tool_input": arg}
-                prev_hash, _ = self._commit_receipt(run_id, node_seq, "ACTION",
-                                                    body, prev_hash, reflection)
-            else:  # OBSERVATION (execute the tool for real)
-                act = next(s for s in reversed(scratch) if s.get("node") == "ACTION")
-                fn = _TOOLS.get(act["tool"], _tool_echo)
-                obs = fn(act["tool_input"])
-                scratch.append({"node": "OBSERVATION", "step": cur_step,
-                                "observation": obs})
-                # store the observation as a working memory (Letta tiering)
-                _mem_add(run_id, "working", "observation",
-                         "step %d %s->%s" % (cur_step, act["tool"], obs))
-                _page_out_if_full(run_id)
-                body = {"step": cur_step, "tool": act["tool"], "observation": obs}
-                prev_hash, _ = self._commit_receipt(run_id, node_seq, "OBSERVATION",
-                                                    body, prev_hash, reflection)
-            node_seq += 1
-            steps_done += 1
-            # CHECKPOINT after every node transition (SqliteSaver-style)
-            self._checkpoint(run_id, node_seq,
-                             {"scratch": scratch, "node_seq": node_seq}, prev_hash)
-            # honest crash injection for the resumable demo: stop mid-run
-            if kill_after is not None and steps_done >= kill_after:
-                status = "interrupted"
-                break
-        else:
-            status = "completed" if terminal_answer else "max_steps"
-
-        final_hash = prev_hash
-        with _LOCK, _conn() as c:
-            c.execute(
-                "UPDATE runs SET status=?,step=?,prev_hash=?,final_hash=?,"
-                "updated_at=? WHERE run_id=?",
-                (status, node_seq, prev_hash, final_hash, _now_iso(), run_id))
-        return {"run_id": run_id, "status": status, "node_seq": node_seq,
-                "final_hash": final_hash, "answer": terminal_answer,
-                "scratch": scratch}
-
-    def run(self, goal, max_steps=4, kill_after=None):
-        run_id = "run_" + uuid.uuid4().hex[:12]
-        reflection = self._prior_reflection(goal)
-        # seed long-term memory with the goal
-        _mem_add(run_id, "working", "goal", "GOAL: " + (goal or ""))
-        with _LOCK, _conn() as c:
-            c.execute(
-                "INSERT INTO runs(run_id,goal,status,max_steps,step,prev_hash,"
-                "final_hash,reflection,created_at,updated_at) "
-                "VALUES(?,?,?,?,?,?,?,?,?,?)",
-                (run_id, goal, "running", max_steps, 0, "GENESIS", "",
-                 reflection or "", _now_iso(), _now_iso()))
-        # genesis receipt
-        prev_hash = "GENESIS"
-        prev_hash, _ = self._commit_receipt(
-            run_id, -1, "GENESIS",
-            {"goal": goal, "max_steps": max_steps,
-             "prior_reflection_prepended": bool(reflection)},
-            prev_hash, reflection)
-        self._checkpoint(run_id, 0, {"scratch": [], "node_seq": 0}, prev_hash)
-        return self._drive(run_id, goal, max_steps, 0, prev_hash, [],
-                           reflection, kill_after=kill_after)
-
-    def resume(self, run_id):
-        r, cps = self._load_run(run_id)
-        if r is None:
-            return {"error": "unknown run_id", "run_id": run_id}
-        if r["status"] not in ("interrupted", "running", "max_steps"):
-            return {"run_id": run_id, "status": r["status"],
-                    "note": "run already %s \u2014 nothing to resume" % r["status"],
-                    "resumed": False}
-        state = json.loads(cps["state"]) if cps else {"scratch": [], "node_seq": 0}
-        node_seq = state.get("node_seq", 0)
-        scratch = state.get("scratch", [])
-        prev_hash = cps["prev_hash"] if cps else "GENESIS"
-        out = self._drive(run_id, r["goal"], r["max_steps"], node_seq, prev_hash,
-                          scratch, r["reflection"] or None)
-        out["resumed"] = True
-        out["resumed_from_checkpoint_step"] = node_seq
-        return out
-
-    def trace(self, run_id):
-        with _LOCK, _conn() as c:
-            r = c.execute("SELECT * FROM runs WHERE run_id=?", (run_id,)).fetchone()
-            recs = c.execute(
-                "SELECT * FROM receipts WHERE run_id=? ORDER BY seq", (run_id,)
-            ).fetchall()
-        if r is None:
-            return {"error": "unknown run_id", "run_id": run_id}
-        receipts, chain_ok, prev = [], True, "GENESIS"
-        for rec in recs:
-            body = json.loads(rec["body"])
-            recompute = _sha({"seq": rec["seq"], "node": rec["node"],
-                              "body": body, "prev_hash": rec["prev_hash"]})
-            link_ok = (rec["prev_hash"] == prev) and (recompute == rec["hash"])
-            env = json.loads(rec["envelope"])
-            sig_ok = None
-            if self.verify_fn is not None:
-                try:
-                    sig_ok = bool(self.verify_fn(env).get("signature_valid"))
-                except Exception:
-                    sig_ok = False
-            chain_ok = chain_ok and link_ok
-            receipts.append({"seq": rec["seq"], "node": rec["node"], "body": body,
-                             "hash": rec["hash"], "prev_hash": rec["prev_hash"],
-                             "link_ok": link_ok, "signature_valid": sig_ok,
-                             "signed": bool(env.get("signed")),
-                             "ts": rec["ts"]})
-            prev = rec["hash"]
-        return {"run_id": run_id, "goal": r["goal"], "status": r["status"],
-                "reflection": r["reflection"],
-                "chain_intact": chain_ok, "depth": len(receipts),
-                "final_hash": r["final_hash"], "receipts": receipts,
-                "trust_note": "Receipt chain + DSSE signatures are REAL; planner is "
-                              "HEURISTIC (deterministic, replayable). Trust=Conjecture 1."}
-
-    def checkpoints(self, run_id):
-        with _LOCK, _conn() as c:
-            cps = c.execute(
-                "SELECT step,prev_hash,ts FROM checkpoints WHERE run_id=? "
-                "ORDER BY step", (run_id,)).fetchall()
-            r = c.execute("SELECT status,step FROM runs WHERE run_id=?",
-                          (run_id,)).fetchone()
-        return {"run_id": run_id,
-                "status": (r["status"] if r else "unknown"),
-                "current_step": (r["step"] if r else None),
-                "checkpoints": [{"step": c["step"], "prev_hash": c["prev_hash"],
-                                 "ts": c["ts"]} for c in cps],
-                "saver": "SqliteSaver-style (local sqlite, ephemeral per container)"}
-
-    def reflect(self, run_id, reflection_text):
-        """Reflexion: store a NL reflection after a reviewed episode."""
-        with _LOCK, _conn() as c:
-            n = c.execute("SELECT COUNT(*) AS n FROM reflections WHERE run_id=?",
-                          (run_id,)).fetchone()["n"]
-            c.execute("INSERT OR REPLACE INTO reflections(run_id,idx,text,ts) "
-                      "VALUES(?,?,?,?)", (run_id, n, reflection_text, _now_iso()))
-            c.execute("UPDATE runs SET reflection=? WHERE run_id=?",
-                      (reflection_text, run_id))
-        _mem_add(run_id, "archival", "reflection", reflection_text, importance=0.85)
-        return {"run_id": run_id, "stored": True, "reflection": reflection_text,
-                "note": "Prepended to the next activation on a lexically-relevant goal."}
-
-
-# ---------------------------------------------------------------------------
-# register(app, ns, sign_fn, verify_fn, pub_pem_fn) — mirrors szl_agentic_loop.
-# Routes inserted at position 0 (Starlette Route) so they beat the SPA catch-all.
-# FREE sub-namespace /api/a11oy/v1/agent/react/* to avoid collisions with the
-# existing /run, /tools, /verify-chain, /governance-standards, /_diag, /loop.
-# ---------------------------------------------------------------------------
-def register(app, ns: str = "a11oy", sign_fn=None, verify_fn=None,
-             pub_pem_fn=None, signer_label: str = "in-image key"):
-    from starlette.routing import Route
-    from starlette.responses import JSONResponse
-
-    _init_db()
-    eng = _ReActEngine(sign_fn, verify_fn, pub_pem_fn, ns=ns)
-
-    async def _read_json(request):
-        try:
-            return await request.json()
-        except Exception:
-            return {}
-
-    async def _run(request):
-        d = await _read_json(request)
-        goal = (d.get("goal") or d.get("query") or "").strip()
-        if not goal:
-            return JSONResponse({"error": "missing 'goal'"}, status_code=400)
-        max_steps = int(d.get("max_steps", 4))
-        kill_after = d.get("kill_after")  # honest crash-injection for the demo
-        kill_after = int(kill_after) if kill_after is not None else None
-        out = eng.run(goal, max_steps=max_steps, kill_after=kill_after)
-        out["label"] = "EXPERIMENTAL"
-        return JSONResponse(out)
-
-    async def _resume(request):
-        d = await _read_json(request)
-        run_id = (d.get("run_id") or request.query_params.get("run_id") or "").strip()
-        if not run_id:
-            return JSONResponse({"error": "missing 'run_id'"}, status_code=400)
-        out = eng.resume(run_id)
-        out["label"] = "EXPERIMENTAL"
-        return JSONResponse(out)
-
-    async def _trace(request):
-        run_id = request.path_params.get("run_id") or request.query_params.get("run_id", "")
-        return JSONResponse(eng.trace(run_id))
-
-    async def _checkpoints(request):
-        run_id = request.path_params.get("run_id") or request.query_params.get("run_id", "")
-        return JSONResponse(eng.checkpoints(run_id))
-
-    async def _reflect(request):
-        d = await _read_json(request)
-        run_id = (d.get("run_id") or "").strip()
-        text = (d.get("reflection") or d.get("text") or "").strip()
-        if not run_id or not text:
-            return JSONResponse({"error": "need run_id + reflection"}, status_code=400)
-        return JSONResponse(eng.reflect(run_id, text))
-
-    async def _mem_add_ep(request):
-        d = await _read_json(request)
-        text = (d.get("text") or "").strip()
-        if not text:
-            return JSONResponse({"error": "missing 'text'"}, status_code=400)
-        mid = _mem_add(d.get("run_id", "adhoc"), d.get("tier", "archival"),
-                       d.get("kind", "note"), text, d.get("importance"))
-        return JSONResponse({"mem_id": mid, "tier": d.get("tier", "archival"),
-                             "label": "HEURISTIC"})
-
-    async def _mem_search_ep(request):
-        d = await _read_json(request)
-        q = (d.get("query") or "").strip()
-        if not q:
-            return JSONResponse({"error": "missing 'query'"}, status_code=400)
-        hits = _mem_retrieve(q, top_k=int(d.get("top_k", 5)), tier=d.get("tier"))
-        return JSONResponse({"query": q, "results": hits, "label": "HEURISTIC",
-                             "formula": "score(m)=a_rec*g^dt + a_imp*imp(m) + a_rel*cos(q,m)",
-                             "source": "Generative Agents (arXiv 2304.03442)"})
-
-    async def _mem_tiers(request):
-        run_id = request.query_params.get("run_id", "")
-        with _LOCK, _conn() as c:
-            wq = ("SELECT tier,COUNT(*) AS n FROM memory" +
-                  (" WHERE run_id=?" if run_id else "") + " GROUP BY tier")
-            rows = c.execute(wq, ((run_id,) if run_id else ())).fetchall()
-        tiers = {r["tier"]: r["n"] for r in rows}
-        return JSONResponse({"run_id": run_id or None, "tiers": tiers,
-                             "working": _working_snapshot(run_id) if run_id else [],
-                             "working_cap": _WORKING_CAP,
-                             "design": "Letta/MemGPT (arXiv 2310.08560): working "
-                                       "(in-context) + archival (vector); self-managed.",
-                             "label": "EXPERIMENTAL"})
-
-    async def _skill_admit_ep(request):
-        d = await _read_json(request)
-        name = (d.get("name") or "").strip()
-        recipe = (d.get("recipe") or "").strip()
-        run_id = (d.get("run_id") or "").strip()
-        if not name or not recipe:
-            return JSONResponse({"error": "need name + recipe"}, status_code=400)
-        # Voyager admission: require a VERIFIED execution receipt. We accept a
-        # run_id and verify its final emit receipt; OR a direct envelope.
-        verified, rhash = False, ""
-        if run_id:
-            tr = eng.trace(run_id)
-            recs = tr.get("receipts", [])
-            if recs:
-                last = recs[-1]
-                rhash = last["hash"]
-                verified = bool(last.get("signature_valid")) and tr.get("chain_intact")
-        elif d.get("envelope") and verify_fn is not None:
-            try:
-                verified = bool(verify_fn(d["envelope"]).get("signature_valid"))
-                rhash = _sha(d["envelope"])[:32]
-            except Exception:
-                verified = False
-        res = _skill_admit(name, recipe, rhash, verified)
-        res["label"] = "LIVE" if res.get("admitted") else "EXPERIMENTAL"
-        return JSONResponse(res, status_code=200 if res.get("admitted") else 422)
-
-    async def _skill_list(request):
-        q = request.query_params.get("q", "")
-        return JSONResponse({"query": q or None,
-                             "skills": _skill_search(q or "skill", top_k=50),
-                             "admission_rule": "Voyager (arXiv 2305.16291): admit a "
-                                               "recipe ONLY after a verified execution receipt.",
-                             "label": "LIVE"})
-
-    async def _diag(request):
-        with _LOCK, _conn() as c:
-            nr = c.execute("SELECT COUNT(*) AS n FROM runs").fetchone()["n"]
-            nm = c.execute("SELECT COUNT(*) AS n FROM memory").fetchone()["n"]
-            ns_ = c.execute("SELECT COUNT(*) AS n FROM skills").fetchone()["n"]
-        return JSONResponse({
-            "module": "a11oy_react_core", "status": "ok",
-            "db": _DB_PATH, "runs": nr, "memories": nm, "skills": ns_,
-            "signer": signer_label,
-            "pubkey_present": bool((pub_pem_fn() if pub_pem_fn else "")),
-            "subsystems": ["ReAct graph (2210.03629)", "SqliteSaver checkpointing",
-                           "Reflexion (2303.11366)", "Generative-Agents memory (2304.03442)",
-                           "Letta tiering (2310.08560)", "Voyager skill library (2305.16291)"],
-            "label": "EXPERIMENTAL"})
-
-    base = "/api/%s/v1/agent/react" % ns
-    routes = [
-        Route(base + "/run", _run, methods=["POST"], name="%s_react_run" % ns),
-        Route(base + "/resume", _resume, methods=["POST"], name="%s_react_resume" % ns),
-        Route(base + "/trace/{run_id}", _trace, methods=["GET"], name="%s_react_trace" % ns),
-        Route(base + "/trace", _trace, methods=["GET"], name="%s_react_trace_q" % ns),
-        Route(base + "/checkpoints/{run_id}", _checkpoints, methods=["GET"],
-              name="%s_react_cps" % ns),
-        Route(base + "/checkpoints", _checkpoints, methods=["GET"], name="%s_react_cps_q" % ns),
-        Route(base + "/reflect", _reflect, methods=["POST"], name="%s_react_reflect" % ns),
-        Route(base + "/memory/add", _mem_add_ep, methods=["POST"], name="%s_react_mem_add" % ns),
-        Route(base + "/memory/search", _mem_search_ep, methods=["POST"],
-              name="%s_react_mem_search" % ns),
-        Route(base + "/memory/tiers", _mem_tiers, methods=["GET"], name="%s_react_mem_tiers" % ns),
-        Route(base + "/skills/admit", _skill_admit_ep, methods=["POST"],
-              name="%s_react_skill_admit" % ns),
-        Route(base + "/skills", _skill_list, methods=["GET"], name="%s_react_skills" % ns),
-        Route(base + "/_diag", _diag, methods=["GET"], name="%s_react_diag" % ns),
-        # Free top-level conveniences requested by the spec (not taken elsewhere):
-        Route("/api/%s/v1/agent/resume" % ns, _resume, methods=["POST"],
-              name="%s_agent_resume_top" % ns),
-        Route("/api/%s/v1/agent/trace/{run_id}" % ns, _trace, methods=["GET"],
-              name="%s_agent_trace_top" % ns),
-        Route("/api/%s/v1/agent/checkpoints/{run_id}" % ns, _checkpoints, methods=["GET"],
-              name="%s_agent_cps_top" % ns),
-        Route("/api/%s/v1/agent/checkpoints" % ns, _checkpoints, methods=["GET"],
-              name="%s_agent_cps_top_q" % ns),
-    ]
-    for r in routes:
-        app.router.routes.insert(0, r)
-    return {"module": "a11oy_react_core", "routes": len(routes), "base": base,
-            "signer": signer_label}

serve.py

@@ -250,6 +250,19 @@ try:
 except Exception as _szl_eb_e:  # pragma: no cover
     print(f"[a11oy] Energy-budget receipt NOT registered: {_szl_eb_e!r}", file=__import__("sys").stderr)
 
+# ── Energy / Sovereign-Compute instrumentation (Lane C: sovereign-energy). Reads REAL
+# J/token + carbon + speculative-decode + KV-cache + router + carbon-schedule from the
+# on-box vLLM /metrics ONLY when the live sovereign probe shows gpu_reachable; otherwise
+# every panel is honestly labeled ROADMAP (no meter -> no number). Adds the unified
+# /energy tab + /api/a11oy/v1/energy/{sovereign,jtoken,throughput,kvcache,gateway,router,
+# carbon}. Additive, try/except-guarded, before the SPA catch-all. window.SZLLabels.
+try:
+    import szl_energy_sovereign as _szl_energy_sovereign
+    _szl_energy_sovereign.register(app, ns="a11oy")
+    print("[a11oy] Energy/Sovereign-Compute registered: /energy + /api/a11oy/v1/energy/*", file=__import__("sys").stderr)
+except Exception as _szl_es_e:  # pragma: no cover
+    print(f"[a11oy] Energy/Sovereign-Compute NOT registered: {_szl_es_e!r}", file=__import__("sys").stderr)
+
 # ── Agentic PINN + Physical-Bounds Certifier MESH (pinn-bounds) — closes the audited
 # gap where the PINN / FE-NO Physics-ML verticals lived ONLY in `platform` and were
 # NOT in a11oy's governed /api/a11oy/v1/<name> route table. Adds /api/a11oy/v1/pinn/*:
@@ -946,12 +959,6 @@ try:
     # Conjecture 1; unreachable organs render dim (no fabricated data).
     app.add_api_route("/hologram", _ptg_serve("hologram.html"), methods=["GET"], include_in_schema=False)
     app.add_api_route("/a11oy/hologram", _ptg_serve("hologram.html"), methods=["GET"], include_in_schema=False)
-    # LANE A AGENTIC CORE (2026-06-14, Dev A): resumable ReAct agent surface —
-    # "Agent Loop" + "Memory" + "Skill Library" tabs. Standalone page (0 CDN;
-    # shared label/receipt engines from /static/shared). Renders REAL traces,
-    # checkpoints and retrieval scores from /api/a11oy/v1/agent/react/*.
-    app.add_api_route("/agent-loop", _ptg_serve("agent-loop.html"), methods=["GET"], include_in_schema=False)
-    app.add_api_route("/a11oy/agent-loop", _ptg_serve("agent-loop.html"), methods=["GET"], include_in_schema=False)
 
     # /chat + /a11oy/chat -> /code consolidation (founder-directed; the only removal).
     async def _ptg_chat_to_code() -> Response:
@@ -7575,46 +7582,6 @@ except Exception as _loop_e:
 # ============================================================================
 
 
-# ============================================================================
-# BEGIN: LANE A AGENTIC CORE — a11oy (2026-06-14, Dev A, ADDITIVE, surgical)
-# Resumable ReAct execution graph (Thought->Action->Observation) where EACH
-# node transition is a SIGNED receipt boundary, with SqliteSaver-style
-# checkpointing (crash mid-run -> /resume continues), a Reflexion inner loop,
-# Generative-Agents memory retrieval scoring over a LOCAL vector store (0 CDN),
-# Letta-style working/archival tiering, and a Voyager skill library that admits
-# a tool-recipe ONLY after a verified execution receipt.
-# REUSES the host's REAL in-image signer (_a11oy_sign_receipt), verifier
-# (_a11oy_loop_verify) and public key (_a11oy_loop_pubpem). Routes inserted at
-# position 0 (Starlette Route) so they beat the SPA catch-all. FREE sub-namespace
-# /api/a11oy/v1/agent/react/* — no collision with /run, /tools, /verify-chain,
-# /governance-standards, /_diag, /loop. try/except-guarded (non-fatal).
-# Signed-off-by: Stephen P. Lutar Jr. <stephenlutar2@gmail.com>
-# Co-Authored-By: Perplexity Computer Agent <agent@perplexity.ai>
-# ============================================================================
-try:
-    import a11oy_react_core as _react_core
-    import sys as _react_sys
-    _react_status = _react_core.register(
-        app, "a11oy",
-        sign_fn=_a11oy_sign_receipt,
-        verify_fn=(_a11oy_loop_verify if "_a11oy_loop_verify" in dir() else None),
-        pub_pem_fn=(_a11oy_loop_pubpem if "_a11oy_loop_pubpem" in dir() else None),
-        signer_label=("in-image ephemeral ECDSA-P256 (signed at server boot, "
-                      "resets on rebuild, verifiable vs /cosign.pub)"),
-    )
-    print(f"[a11oy] LANE A agentic core registered: {_react_status}", file=_react_sys.stderr)
-    _REACT_DIAG = {"status": "ok", "registered": _react_status}
-except Exception as _react_e:
-    import sys as _react_sys, traceback as _react_tb
-    print(f"[a11oy] LANE A agentic core FAILED (non-fatal): {_react_e!r}", file=_react_sys.stderr)
-    _react_tb.print_exc(file=_react_sys.stderr)
-    _REACT_DIAG = {"status": "FAILED", "error": repr(_react_e),
-                   "traceback": _react_tb.format_exc()}
-# ============================================================================
-# END: LANE A AGENTIC CORE — a11oy
-# ============================================================================
-
-
 # ============================================================================
 # BEGIN: FORMULA-WIRING SURFACE — a11oy (2026-06-06, ADDITIVE, surgical)
 # Wires ALL ~80 kernel-verified theorems to REAL, executed mechanisms (shared,

szl_energy_sovereign.py

@@ -0,0 +1,883 @@
+# SPDX-License-Identifier: Apache-2.0
+# © 2026 Lutar, Stephen P. — SZL Holdings · ORCID 0009-0001-0110-4173
+# Doctrine v11 LOCKED: locked-proven=8 · Λ=Conjecture 1 · SLSA L1 honest / L2 attested / L3 roadmap
+# Co-Authored-By: Perplexity Computer Agent
+"""
+szl_energy_sovereign.py — ADDITIVE Energy / Sovereign-Compute instrumentation for
+a11oy. Makes the energy + sovereign-compute story MEASURED and HONEST.
+
+The whole point of this module: read REAL energy / throughput / speculative-decode /
+KV-cache / router metrics from the on-box vLLM `/metrics` (Prometheus) endpoint WHEN
+the live sovereign-inference probe shows the GPU is reachable — and label everything
+ROADMAP / pending when it is not. It NEVER fabricates a sovereign or energy number.
+
+Honesty spine (doctrine v11):
+  * sovereign:true ONLY from the orchestrator's live `_sovereign_inference_state()`
+    (which itself gates on a real `_local_endpoint_reachable()` /models probe).
+  * the joules honesty label is decided ONLY by `szl_joules_truth.joules_label()` off a
+    REAL, FRESH NVML/exporter sample — never off a bare flag or a forwarded string.
+  * J/token is computed from the research formula  E_token = P_GPU · T_forward / N_tokens
+    (Watt-Counts arXiv:2604.09048; Energy-per-Token arXiv:2603.20224; Where-Do-Joules-Go
+    arXiv:2601.22076) — and ONLY when the box emits power.draw + token counters. Else it
+    is left None and labeled ROADMAP. No meter -> no number.
+
+Capability tiers (rendered with honest labels mapped to window.SZLLabels):
+  MEASURED  -> a real on-box exporter sample is present & fresh (mapped to LIVE tone)
+  ROADMAP   -> wiring is in place; the box is not emitting the metric yet (EXPERIMENTAL tone)
+
+Routes (NEW; never collide):
+  GET /api/{ns}/v1/energy/sovereign            — full JSON posture (machine-readable)
+  GET /api/{ns}/v1/energy/jtoken               — J/token + carbon panel (MEASURED/ROADMAP)
+  GET /api/{ns}/v1/energy/throughput           — speculative-decode tokens/s panel
+  GET /api/{ns}/v1/energy/kvcache              — LMCache TTFT before/after panel
+  GET /api/{ns}/v1/energy/gateway              — LiteLLM gateway status (honest)
+  GET /api/{ns}/v1/energy/router               — RouteLLM Thompson Beta posteriors
+  GET /api/{ns}/v1/energy/carbon               — Carbon-Aware batch schedule
+  GET /energy                                  — unified mobile-first Energy/Sovereign tab
+
+Pure stdlib + (optional) the shared szl_joules_truth. Defensive: a probe/parse failure
+NEVER raises out of a handler — it degrades to an honest ROADMAP/pending posture.
+"""
+from __future__ import annotations
+
+import json as _json
+import math as _math
+import os as _os
+import random as _random
+import time as _time
+import urllib.request as _ureq
+from datetime import datetime, timezone
+
+# --- joules honesty: the SINGLE source of truth (never decide a label locally) ----
+try:
+    from szl_joules_truth import (
+        joules_label as _joules_label,
+        joules_evidence as _joules_evidence,
+        is_real_fresh_sample as _is_real_fresh_sample,
+    )
+except Exception:  # pragma: no cover — defensive: doctrine default is always sample
+    def _joules_label(_s, now=None):  # type: ignore
+        return "sample"
+    def _joules_evidence(_s, now=None):  # type: ignore
+        return {}
+    def _is_real_fresh_sample(_s, now=None):  # type: ignore
+        return False
+
+DOCTRINE = {
+    "version": "v11",
+    "locked_proven": ["F1", "F4", "F7", "F11", "F12", "F18", "F19", "F22"],
+    "locked_count": 8,
+    "corpus": "749/14/163",
+    "kernel_commit": "c7c0ba17",
+    "lambda": "Conjecture 1 (advisory floor; uniqueness machine-checked FALSE unconditionally; NOT a theorem)",
+    "slsa": "L1 honest / L2 attested (.att emitted, not independently verified) / L3 roadmap",
+}
+
+# Carbon intensity for the GPU's grid region. SAMPLE/ESTIMATE default until a real
+# Carbon-Aware SDK / WattTime / ElectricityMaps feed is wired (see FORGE_BOX_ENERGY.md).
+# Hetzner Falkenstein (DE) ~ 380 gCO2eq/kWh annual average — labeled, not claimed live.
+_DEFAULT_CARBON_G_PER_KWH = 380.0
+_J_PER_KWH = 3.6e6  # 1 kWh = 3.6 MJ
+
+# Research-cited speculative-decoding speedup model (vLLM speculative decoding):
+#   S(k, alpha) = (k + 1) / (k * (1 - alpha) + 1)
+# k = number of draft tokens proposed per step; alpha = empirical acceptance rate.
+_SPEC_DRAFT_MODEL = "Qwen/Qwen2.5-Coder-1.5B-Instruct"
+_SPEC_K_DEFAULT = 4
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _carbon_g_per_kwh() -> float:
+    """Grid carbon intensity (gCO2eq/kWh). Env-overridable when a real feed sets it;
+    otherwise the labeled SAMPLE/ESTIMATE regional average. Never raises."""
+    try:
+        v = _os.environ.get("A11OY_GRID_CARBON_G_PER_KWH")
+        if v:
+            f = float(v)
+            if f > 0 and f == f:
+                return f
+    except Exception:
+        pass
+    return _DEFAULT_CARBON_G_PER_KWH
+
+
+def _carbon_feed_is_live() -> bool:
+    """True ONLY when a real carbon-intensity feed is wired (env set by the box)."""
+    return bool((_os.environ.get("A11OY_GRID_CARBON_G_PER_KWH") or "").strip())
+
+
+# ---------------------------------------------------------------------------
+# LIVE sovereign-inference probe (delegated to the orchestrator — the authority).
+# ---------------------------------------------------------------------------
+def _sovereign_state() -> dict:
+    """Authoritative, LIVE sovereign-inference posture.
+
+    Delegates to a11oy_code_orchestrator._sovereign_inference_state(), which gates
+    sovereign:true on a real _local_endpoint_reachable() /models probe. We NEVER
+    decide sovereignty here. Honest default on any failure: not sovereign.
+    """
+    try:
+        import a11oy_code_orchestrator as _orch  # type: ignore
+        st = _orch._sovereign_inference_state()
+        if isinstance(st, dict):
+            return st
+    except Exception:
+        pass
+    return {"inference": "unknown", "mode": "unknown", "backend": "unknown",
+            "sovereign": False, "base_url": None,
+            "honest_note": "orchestrator sovereign-state unavailable in-process; honest default not-sovereign."}
+
+
+def _gpu_reachable(state: dict | None = None) -> bool:
+    """gpu_reachable == the orchestrator reports a live self-hosted GPU (sovereign)."""
+    st = state if state is not None else _sovereign_state()
+    return bool(st.get("sovereign") is True and st.get("inference") == "self-hosted-gpu")
+
+
+def _vllm_metrics_base() -> str | None:
+    """Base URL of the on-box vLLM server for its Prometheus /metrics endpoint.
+
+    Prefer an explicit metrics URL; else derive from the serving base. Returns None
+    when no local endpoint is configured (so we stay ROADMAP rather than guess)."""
+    explicit = (_os.environ.get("A11OY_VLLM_METRICS_URL") or "").strip()
+    if explicit:
+        return explicit.rstrip("/")
+    base = (_os.environ.get("A11OY_MODEL_BASE_URL") or "").strip().rstrip("/")
+    if base and "router.huggingface.co" not in base:
+        # vLLM exposes /metrics at the server root (strip a trailing /v1).
+        root = base[:-3] if base.endswith("/v1") else base
+        return root.rstrip("/")
+    return None
+
+
+def _fetch_metrics_text(timeout: float = 2.5) -> str | None:
+    """Fetch the on-box vLLM Prometheus /metrics text. None on any failure (honest)."""
+    base = _vllm_metrics_base()
+    if not base:
+        return None
+    for path in ("/metrics", ""):
+        try:
+            req = _ureq.Request(base + path, headers={"User-Agent": "szl-energy-sovereign"})
+            with _ureq.urlopen(req, timeout=timeout) as r:  # noqa: S310
+                if 200 <= getattr(r, "status", 200) < 300:
+                    body = r.read().decode("utf-8", "replace")
+                    if "# " in body or "_total" in body:
+                        return body
+        except Exception:  # noqa: BLE001 — any failure => not emitting; stay honest
+            continue
+    return None
+
+
+def _parse_prom(text: str) -> dict:
+    """Minimal Prometheus text parser: {metric_name: float} for samples we care about.
+
+    Sums across label sets for counters (e.g. vllm:* counters carry model_name labels).
+    Best-effort and exception-tolerant; unknown lines are ignored.
+    """
+    out: dict[str, float] = {}
+    if not text:
+        return out
+    for line in text.splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        try:
+            # name{labels} value   OR   name value
+            if "{" in line:
+                name = line[: line.index("{")]
+                value = line.rsplit(" ", 1)[1]
+            else:
+                name, value = line.rsplit(" ", 1)
+            v = float(value)
+            out[name] = out.get(name, 0.0) + v
+        except Exception:
+            continue
+    return out
+
+
+def _exporter_sample_from_metrics(prom: dict) -> dict | None:
+    """Build a szl_joules_truth exporter_sample from on-box metrics, or None.
+
+    We treat a fresh power.draw export (via a node/GPU exporter scraped INTO vLLM
+    /metrics or a sidecar) as the real NVML reading. Doctrine: only a numeric
+    joules_measured_total + a fresh exporter_last_seen_ts can yield a MEASURED label.
+    """
+    if not prom:
+        return None
+    # Common exporter names for cumulative GPU energy (joules). Box wires one of these
+    # (nvidia-smi power.draw integrated, or DCGM energy counter). See FORGE_BOX_ENERGY.md.
+    joules_total = None
+    for key in ("a11oy_gpu_energy_joules_total", "gpu_energy_joules_total",
+                "DCGM_FI_DEV_TOTAL_ENERGY_CONSUMPTION", "nvidia_gpu_energy_joules_total"):
+        if key in prom:
+            joules_total = prom[key]
+            break
+    if joules_total is None:
+        return None
+    power_w = None
+    for key in ("a11oy_gpu_power_watts", "gpu_power_watts",
+                "DCGM_FI_DEV_POWER_USAGE", "nvidia_gpu_power_watts"):
+        if key in prom:
+            power_w = prom[key]
+            break
+    return {
+        "joules_measured_total": joules_total,
+        "exporter_node": _os.environ.get("A11OY_GPU_LABEL") or "sovereign-gpu",
+        "exporter_last_seen_ts": _time.time(),  # scraped just now => fresh by construction
+        "power_w_sample": power_w,
+    }
+
+
+# ---------------------------------------------------------------------------
+# (#1) J/token + carbon — the Tier-1 receipt fields.
+# ---------------------------------------------------------------------------
+def _jtoken_from_metrics(prom: dict, sample: dict | None) -> dict:
+    """Compute MEASURED energy-per-token from on-box counters, else honest ROADMAP.
+
+    Formula (research-cited): E_token = P_GPU · T_forward / N_tokens. We realize it as
+    cumulative_energy_joules / cumulative_generated_tokens when both counters exist —
+    equivalent to the time-integral form averaged over the window. Returned None +
+    ROADMAP when the box is not emitting the counters.
+    """
+    measured = _is_real_fresh_sample(sample)
+    gen_tokens = None
+    for key in ("vllm:generation_tokens_total", "vllm_generation_tokens_total",
+                "vllm:generation_tokens"):
+        if key in prom:
+            gen_tokens = prom[key]
+            break
+    joules_total = (sample or {}).get("joules_measured_total")
+    j_per_token = None
+    if measured and joules_total and gen_tokens and gen_tokens > 0:
+        j_per_token = float(joules_total) / float(gen_tokens)
+    label = "MEASURED" if (measured and j_per_token is not None) else "ROADMAP"
+    carbon_g_per_token = None
+    if j_per_token is not None:
+        carbon_g_per_token = (j_per_token / _J_PER_KWH) * _carbon_g_per_kwh()
+    return {
+        "metric": "energy_per_token",
+        "label": label,
+        "joules_per_token": (round(j_per_token, 6) if j_per_token is not None else None),
+        "carbon_g_co2eq_per_token": (round(carbon_g_per_token, 9)
+                                     if carbon_g_per_token is not None else None),
+        "generated_tokens_total": gen_tokens,
+        "gpu_energy_joules_total": joules_total,
+        "power_w_sample": (sample or {}).get("power_w_sample"),
+        "carbon_g_per_kwh": _carbon_g_per_kwh(),
+        "carbon_feed_live": _carbon_feed_is_live(),
+        "formula": "E_token = P_GPU · T_forward / N_tokens (= ΣJ / Σgenerated_tokens over window)",
+        "citations": ["Watt-Counts arXiv:2604.09048", "Energy-per-Token arXiv:2603.20224",
+                      "Where-Do-Joules-Go arXiv:2601.22076"],
+        "joules_honesty": _joules_label(sample),
+        "joules_evidence": _joules_evidence(sample),
+        "note": ("Real per-token energy from the on-box GPU exporter + vLLM token counters."
+                 if label == "MEASURED" else
+                 "Wiring is ready: when the box emits power.draw→/metrics + token counters, "
+                 "J/token and carbon become MEASURED. Until then, honestly pending (no meter, no number)."),
+    }
+
+
+def energy_fields_for_receipt() -> dict:
+    """SPLICE-INTO-RECEIPT helper for the orchestrator's _emit_turn_receipt().
+
+    Returns joules_consumed + carbon_g_co2eq for EVERY signed turn receipt, honestly
+    labeled MEASURED (real fresh on-box exporter) or ROADMAP (no meter yet → None).
+    NEVER raises and NEVER fabricates a number. The orchestrator splats this into the
+    receipt body so the joules claim is self-verifying from the receipt itself.
+    """
+    try:
+        state = _sovereign_state()
+        if not _gpu_reachable(state):
+            return {
+                "joules_consumed": None,
+                "carbon_g_co2eq": None,
+                "energy_label": "ROADMAP",
+                "joules_honesty": _joules_label(None),
+                "energy_source_note": ("No sovereign GPU reachable (sovereign:%s). Per-receipt "
+                                       "joules/carbon are pending the on-box NVML exporter — "
+                                       "honest ROADMAP, never fabricated." % state.get("sovereign")),
+            }
+        prom = _parse_prom(_fetch_metrics_text() or "")
+        sample = _exporter_sample_from_metrics(prom)
+        jt = _jtoken_from_metrics(prom, sample)
+        gen_tokens = jt.get("generated_tokens_total")
+        jpt = jt.get("joules_per_token")
+        # Per-receipt instantaneous reading: total measured joules is the honest figure;
+        # per-token rate is the MEASURED derivation. We report the fresh cumulative joules
+        # as joules_consumed only when MEASURED; else None (ROADMAP).
+        measured = jt.get("label") == "MEASURED"
+        joules_consumed = (sample or {}).get("joules_measured_total") if measured else None
+        carbon = None
+        if measured and joules_consumed is not None:
+            carbon = (float(joules_consumed) / _J_PER_KWH) * _carbon_g_per_kwh()
+        return {
+            "joules_consumed": (round(float(joules_consumed), 6)
+                                if joules_consumed is not None else None),
+            "carbon_g_co2eq": (round(carbon, 9) if carbon is not None else None),
+            "joules_per_token": jpt,
+            "carbon_g_co2eq_per_token": jt.get("carbon_g_co2eq_per_token"),
+            "energy_label": "MEASURED" if measured else "ROADMAP",
+            "joules_honesty": _joules_label(sample),
+            "joules_evidence": _joules_evidence(sample),
+            "carbon_g_per_kwh": _carbon_g_per_kwh(),
+            "carbon_feed_live": _carbon_feed_is_live(),
+            "energy_source_note": ("Per-token energy + carbon MEASURED from on-box GPU exporter."
+                                   if measured else
+                                   "Sovereign GPU reachable but power/token exporter not emitting "
+                                   "yet → joules/carbon honestly ROADMAP (never fabricated)."),
+        }
+    except Exception as exc:  # noqa: BLE001 — a receipt helper must NEVER raise
+        return {
+            "joules_consumed": None,
+            "carbon_g_co2eq": None,
+            "energy_label": "ROADMAP",
+            "joules_honesty": "sample",
+            "energy_source_note": "energy_fields_for_receipt fail-open: %s" % (str(exc)[:120],),
+        }
+
+
+# ---------------------------------------------------------------------------
+# (#2) Speculative decoding throughput panel (tokens/s with-vs-without).
+# ---------------------------------------------------------------------------
+def _spec_speedup(k: int, alpha: float) -> float:
+    """vLLM speculative-decoding speedup model S = (k+1)/(k(1-alpha)+1)."""
+    denom = (k * (1.0 - alpha)) + 1.0
+    return (k + 1.0) / denom if denom > 0 else 1.0
+
+
+def _throughput_panel(prom: dict, gpu_reachable: bool) -> dict:
+    """tokens/s with-vs-without speculative decoding.
+
+    MEASURED tokens/s when vLLM emits generation tokens + an accept-rate counter;
+    else ROADMAP with the theoretical speedup model (k=4, draft Qwen2.5-Coder-1.5B)
+    shown ILLUSTRATIVELY so the panel is never blank but never overclaims.
+    """
+    tps = None
+    for key in ("vllm:avg_generation_throughput_toks_per_s",
+                "vllm_avg_generation_throughput_toks_per_s",
+                "vllm:generation_tokens_per_second"):
+        if key in prom:
+            tps = prom[key]
+            break
+    # Empirical acceptance rate alpha from vLLM spec-decode counters when present.
+    accepted = prom.get("vllm:spec_decode_num_accepted_tokens_total")
+    drafted = prom.get("vllm:spec_decode_num_draft_tokens_total")
+    alpha = None
+    if accepted is not None and drafted and drafted > 0:
+        alpha = max(0.0, min(1.0, accepted / drafted))
+    measured = bool(gpu_reachable and tps is not None and alpha is not None)
+    k = _SPEC_K_DEFAULT
+    # Theoretical curve for the panel (ILLUSTRATIVE unless alpha is measured).
+    alpha_for_model = alpha if alpha is not None else 0.8
+    speedup = _spec_speedup(k, alpha_for_model)
+    tps_without = (tps / speedup) if (measured and speedup > 0) else None
+    return {
+        "metric": "speculative_decode_throughput",
+        "label": "MEASURED" if measured else "ROADMAP",
+        "draft_model": _SPEC_DRAFT_MODEL,
+        "num_speculative_tokens_k": k,
+        "acceptance_rate_alpha": (round(alpha, 4) if alpha is not None else None),
+        "acceptance_rate_alpha_label": ("MEASURED" if alpha is not None else "ILLUSTRATIVE (α=0.8 assumed)"),
+        "tokens_per_s_with_spec": (round(tps, 3) if tps is not None else None),
+        "tokens_per_s_without_spec": (round(tps_without, 3) if tps_without is not None else None),
+        "modeled_speedup_x": round(speedup, 3),
+        "speedup_formula": "S = (k+1) / (k·(1−α)+1)",
+        "note": ("Real tokens/s + empirical α from the on-box vLLM speculative-decode counters."
+                 if measured else
+                 "Wiring ready: with --speculative-model %s --num-speculative-tokens %d the box emits "
+                 "tokens/s + α. Until then the speedup curve is ILLUSTRATIVE (α=0.8 → ~%.2f×)."
+                 % (_SPEC_DRAFT_MODEL, k, speedup)),
+        "citations": ["vLLM speculative decoding", "Watt-Counts arXiv:2604.09048"],
+    }
+
+
+# ---------------------------------------------------------------------------
+# (#3) LMCache KV-cache TTFT before/after panel.
+# ---------------------------------------------------------------------------
+def _kvcache_panel(prom: dict, gpu_reachable: bool) -> dict:
+    """TTFT before/after KV-cache reuse on repeated prompt prefixes (LMCache)."""
+    hit = None
+    for key in ("vllm:prefix_cache_hits_total", "lmcache_cache_hits_total",
+                "vllm:gpu_prefix_cache_hits_total"):
+        if key in prom:
+            hit = prom[key]
+            break
+    queries = None
+    for key in ("vllm:prefix_cache_queries_total", "lmcache_cache_queries_total",
+                "vllm:gpu_prefix_cache_queries_total"):
+        if key in prom:
+            queries = prom[key]
+            break
+    ttft = None
+    for key in ("vllm:time_to_first_token_seconds_sum", "vllm_time_to_first_token_seconds_sum"):
+        if key in prom:
+            ttft = prom[key]
+            break
+    hit_rate = (hit / queries) if (hit is not None and queries and queries > 0) else None
+    measured = bool(gpu_reachable and hit_rate is not None and ttft is not None)
+    return {
+        "metric": "kv_cache_ttft",
+        "label": "MEASURED" if measured else "ROADMAP",
+        "backend": "LMCache (KV-cache offload: GPU→CPU/disk)",
+        "prefix_cache_hit_rate": (round(hit_rate, 4) if hit_rate is not None else None),
+        "ttft_seconds_sum": ttft,
+        "ttft_before_after_note": ("Cold prompt = full prefill TTFT; warm repeat prefix = "
+                                   "KV reuse → lower TTFT. Real delta from on-box counters."
+                                   if measured else
+                                   "Wiring ready: LMCache + vLLM prefix-cache counters give the real "
+                                   "TTFT before/after delta on repeated prefixes once the box emits them."),
+        "citations": ["LMCache github.com/LMCache/LMCache"],
+    }
+
+
+# ---------------------------------------------------------------------------
+# (#4) LiteLLM gateway status (unified endpoint, budget, cloud fallback).
+# ---------------------------------------------------------------------------
+def _gateway_panel(state: dict) -> dict:
+    """Honest LiteLLM unified-gateway status.
+
+    LIVE only when A11OY_LITELLM_BASE_URL is set AND answers; else ROADMAP. The
+    gateway fronts Ollama/vLLM with budget enforcement + cloud fallback on GPU OOM.
+    """
+    base = (_os.environ.get("A11OY_LITELLM_BASE_URL") or "").strip().rstrip("/")
+    reachable = False
+    if base:
+        for path in ("/health", "/v1/models", ""):
+            try:
+                req = _ureq.Request(base + path, headers={"User-Agent": "szl-energy-sovereign"})
+                with _ureq.urlopen(req, timeout=2.0) as r:  # noqa: S310
+                    if 200 <= getattr(r, "status", 200) < 500:
+                        reachable = True
+                        break
+            except Exception:
+                continue
+    measured = bool(base and reachable)
+    budget = (_os.environ.get("A11OY_LITELLM_BUDGET_USD") or "").strip()
+    return {
+        "metric": "litellm_gateway",
+        "label": "MEASURED" if measured else "ROADMAP",
+        "gateway": "LiteLLM (OpenAI-compatible proxy over Ollama + vLLM)",
+        "base_url_configured": bool(base),
+        "reachable": reachable,
+        "budget_usd_configured": budget or None,
+        "cloud_fallback": "on GPU OOM / 5xx → HF Router (honest fallback, logged)",
+        "note": ("Unified gateway live: budget enforced, cloud fallback armed."
+                 if measured else
+                 "Wiring ready: set A11OY_LITELLM_BASE_URL to the on-box LiteLLM proxy. Until then "
+                 "the orchestrator serves direct (sovereign-gated) and the gateway is honestly ROADMAP."),
+        "citations": ["LiteLLM github.com/BerriAI/litellm"],
+    }
+
+
+# ---------------------------------------------------------------------------
+# (#5) RouteLLM Thompson-sampling router — Beta posteriors per model.
+# ---------------------------------------------------------------------------
+# Per-model Beta(α, β) posteriors. α = #successful (accepted) routes, β = #failures.
+# Doctrine: these are HONEST in-memory observations. With no real routing traffic yet
+# the posteriors are the priors Beta(1,1) (uniform) — labeled ROADMAP, never faked.
+_ROUTER_MODELS = {
+    "local-7b":  {"alpha": 1.0, "beta": 1.0, "tier": "easy→local",   "model": "qwen2.5-coder:7b"},
+    "cloud-32b": {"alpha": 1.0, "beta": 1.0, "tier": "hard→cloud/32B", "model": "Qwen/Qwen2.5-Coder-32B-Instruct"},
+}
+_ROUTER_OBS = 0  # number of real routing observations recorded (0 => priors only)
+
+
+def record_route_outcome(model_key: str, success: bool) -> None:
+    """Record a real routing outcome into the Beta posterior (α=success, β=failure).
+
+    Called by the router when a route resolves. Pure in-memory, monotone, never raises.
+    Until real traffic calls this, posteriors stay at the Beta(1,1) prior (ROADMAP).
+    """
+    global _ROUTER_OBS
+    try:
+        m = _ROUTER_MODELS.get(model_key)
+        if m is None:
+            return
+        if success:
+            m["alpha"] += 1.0
+        else:
+            m["beta"] += 1.0
+        _ROUTER_OBS += 1
+    except Exception:
+        pass
+
+
+def _thompson_sample(alpha: float, beta: float) -> float:
+    """Sample θ ~ Beta(α, β) (stdlib random.betavariate). Used for argmax selection."""
+    try:
+        return _random.betavariate(max(alpha, 1e-6), max(beta, 1e-6))
+    except Exception:
+        return alpha / (alpha + beta) if (alpha + beta) > 0 else 0.5
+
+
+def _router_panel() -> dict:
+    """RouteLLM Thompson-sampling router posteriors per model (Beta), honestly labeled."""
+    measured = _ROUTER_OBS > 0
+    models = []
+    samples = {}
+    for key, m in _ROUTER_MODELS.items():
+        a, b = m["alpha"], m["beta"]
+        mean = a / (a + b)
+        theta = _thompson_sample(a, b)
+        samples[key] = theta
+        models.append({
+            "model_key": key, "model": m["model"], "route_tier": m["tier"],
+            "beta_alpha_successes": a, "beta_beta_failures": b,
+            "posterior_mean": round(mean, 4),
+            "thompson_sample_theta": round(theta, 4),
+            "n_observations": int(a + b - 2),  # minus the Beta(1,1) prior
+        })
+    chosen = max(samples, key=samples.get) if samples else None
+    return {
+        "metric": "routellm_thompson",
+        "label": "MEASURED" if measured else "ROADMAP",
+        "policy": "Thompson sampling over per-model Beta posteriors; argmax θ wins the route",
+        "models": models,
+        "thompson_choice_this_draw": chosen,
+        "total_observations": _ROUTER_OBS,
+        "note": ("Posteriors updated from real routing outcomes (α=accepted, β=failed)."
+                 if measured else
+                 "Wiring ready: record_route_outcome() updates Beta(α,β) per model on every route. "
+                 "With no live traffic yet the posteriors are the Beta(1,1) uniform prior — honest ROADMAP."),
+        "citations": ["RouteLLM github.com/lm-sys/routellm",
+                      "Thompson sampling: sample θ_k~Beta(α_k,β_k), pick argmax"],
+    }
+
+
+# ---------------------------------------------------------------------------
+# (#6) Carbon-Aware SDK batch scheduling — low-carbon windows.
+# ---------------------------------------------------------------------------
+def _carbon_panel() -> dict:
+    """Carbon-aware batch schedule: shift non-urgent inference to low-carbon windows.
+
+    LIVE forecast only when a real carbon feed is wired; otherwise a labeled SAMPLE
+    diurnal curve (low overnight / midday solar; high evening peak) so the schedule
+    is illustrative-but-honest. carbon_g_co2eq per job is computed from the same J/token.
+    """
+    live = _carbon_feed_is_live()
+    g_per_kwh = _carbon_g_per_kwh()
+    # 24h relative intensity curve (SAMPLE shape unless a real feed overrides).
+    base = [0.55, 0.50, 0.48, 0.47, 0.50, 0.58, 0.72, 0.85, 0.80, 0.70, 0.60, 0.55,
+            0.52, 0.55, 0.62, 0.72, 0.85, 0.95, 1.00, 0.95, 0.85, 0.75, 0.68, 0.60]
+    windows = [{"hour_utc": h, "rel_intensity": base[h],
+                "g_co2eq_per_kwh_est": round(g_per_kwh * base[h], 2)} for h in range(24)]
+    low = sorted(range(24), key=lambda h: base[h])[:3]
+    return {
+        "metric": "carbon_aware_schedule",
+        "label": "MEASURED" if live else "ROADMAP",
+        "scheduler": "Carbon-Aware SDK — shift non-urgent batch inference to low-carbon windows",
+        "carbon_feed_live": live,
+        "carbon_g_per_kwh_now": g_per_kwh,
+        "carbon_intensity_label": ("LIVE" if live else "SAMPLE (regional diurnal curve)"),
+        "low_carbon_windows_utc": sorted(low),
+        "schedule_24h": windows,
+        "note": ("Live grid carbon feed wired: batch jobs deferred to the greenest window."
+                 if live else
+                 "Wiring ready: with the Carbon-Aware SDK feed the schedule becomes a LIVE forecast. "
+                 "Until then the diurnal curve is SAMPLE and per-job carbon_g_co2eq is ESTIMATE."),
+        "citations": ["Carbon-Aware SDK github.com/Green-Software-Foundation/carbon-aware-sdk"],
+    }
+
+
+# ---------------------------------------------------------------------------
+# Full posture (machine-readable) — composes every panel honestly.
+# ---------------------------------------------------------------------------
+def _posture() -> dict:
+    state = _sovereign_state()
+    reachable = _gpu_reachable(state)
+    prom = _parse_prom(_fetch_metrics_text() or "") if reachable else {}
+    sample = _exporter_sample_from_metrics(prom) if reachable else None
+    jtoken = _jtoken_from_metrics(prom, sample)
+    panels = {
+        "jtoken": jtoken,
+        "throughput": _throughput_panel(prom, reachable),
+        "kvcache": _kvcache_panel(prom, reachable),
+        "gateway": _gateway_panel(state),
+        "router": _router_panel(),
+        "carbon": _carbon_panel(),
+    }
+    measured_count = sum(1 for p in panels.values() if p.get("label") == "MEASURED")
+    summary = ("SOVEREIGN ENERGY LIVE (%d/%d MEASURED)" % (measured_count, len(panels))
+               if reachable and measured_count else
+               "WIRED — pending sovereign GPU metrics (honest ROADMAP)")
+    return {
+        "service": "energy-sovereign",
+        "doctrine": DOCTRINE["version"],
+        "summary": summary,
+        "sovereign": bool(state.get("sovereign")),
+        "gpu_reachable": reachable,
+        "inference_state": state,
+        "measured_panels": measured_count,
+        "total_panels": len(panels),
+        "panels": panels,
+        "doctrine_lock": DOCTRINE,
+        "honesty": ("Every panel reads live from the on-box vLLM /metrics ONLY when the live "
+                    "sovereign probe shows gpu_reachable; otherwise it is honestly labeled ROADMAP. "
+                    "The joules label is decided solely by szl_joules_truth off a real fresh exporter "
+                    "sample — never off a flag. No meter → no number."),
+        "computed_at": _now_iso(),
+    }
+
+
+# ---------------------------------------------------------------------------
+# Unified Energy / Sovereign Compute HTML tab (0 CDN; window.SZLLabels).
+# ---------------------------------------------------------------------------
+def _html(p: dict) -> str:
+    panels = p["panels"]
+
+    def row(left: str, right: str) -> str:
+        return ('<div class="kv"><span class="k">%s</span><span class="v">%s</span></div>'
+                % (left, right))
+
+    def fmt(v):
+        return "—" if v is None else (str(v))
+
+    jt = panels["jtoken"]
+    tp = panels["throughput"]
+    kv = panels["kvcache"]
+    gw = panels["gateway"]
+    rt = panels["router"]
+    ca = panels["carbon"]
+
+    # Each card carries a data-szl-label so the inline script renders the honest pill
+    # via window.SZLLabels.badgeHTML (MEASURED→LIVE tone; ROADMAP→EXPERIMENTAL tone).
+    def card(title: str, label: str, body: str, note: str) -> str:
+        return ('<article class="card" data-szl="%s"><div class="row"><h3>%s</h3>'
+                '<span class="pill-slot" data-label="%s"></span></div>%s'
+                '<p class="note">%s</p></article>'
+                % (title, title, label, body, note))
+
+    jt_body = (row("J / token", fmt(jt["joules_per_token"]))
+               + row("gCO₂eq / token", fmt(jt["carbon_g_co2eq_per_token"]))
+               + row("GPU energy (J, cum.)", fmt(jt["gpu_energy_joules_total"]))
+               + row("generated tokens", fmt(jt["generated_tokens_total"]))
+               + row("carbon (gCO₂eq/kWh)", fmt(jt["carbon_g_per_kwh"]))
+               + row("joules honesty", fmt(jt["joules_honesty"])))
+    tp_body = (row("tokens/s WITH spec", fmt(tp["tokens_per_s_with_spec"]))
+               + row("tokens/s WITHOUT", fmt(tp["tokens_per_s_without_spec"]))
+               + row("acceptance rate α", "%s <small>(%s)</small>" % (fmt(tp["acceptance_rate_alpha"]), tp["acceptance_rate_alpha_label"]))
+               + row("modeled speedup", "%s×" % fmt(tp["modeled_speedup_x"]))
+               + row("draft model", "<code>%s</code>" % tp["draft_model"]))
+    kv_body = (row("prefix cache hit-rate", fmt(kv["prefix_cache_hit_rate"]))
+               + row("TTFT Σ (s)", fmt(kv["ttft_seconds_sum"]))
+               + row("backend", kv["backend"]))
+    gw_body = (row("gateway", "LiteLLM proxy")
+               + row("base configured", fmt(gw["base_url_configured"]))
+               + row("reachable", fmt(gw["reachable"]))
+               + row("budget (USD)", fmt(gw["budget_usd_configured"]))
+               + row("cloud fallback", gw["cloud_fallback"]))
+    rt_rows = "".join(
+        row("%s (%s)" % (m["model_key"], m["route_tier"]),
+            "Beta(α=%s, β=%s) μ=%s θ=%s" % (m["beta_alpha_successes"], m["beta_beta_failures"],
+                                            m["posterior_mean"], m["thompson_sample_theta"]))
+        for m in rt["models"])
+    rt_body = rt_rows + row("Thompson choice", fmt(rt["thompson_choice_this_draw"])) + row("observations", fmt(rt["total_observations"]))
+    ca_low = ", ".join("%02d:00" % h for h in ca["low_carbon_windows_utc"])
+    ca_body = (row("carbon now (gCO₂eq/kWh)", fmt(ca["carbon_g_per_kwh_now"]))
+               + row("intensity feed", ca["carbon_intensity_label"])
+               + row("low-carbon windows (UTC)", ca_low)
+               + row("scheduler", "Carbon-Aware SDK"))
+
+    cards = "".join([
+        card("J/token + Carbon", jt["label"], jt_body, jt["note"]),
+        card("Speculative Decoding", tp["label"], tp_body, tp["note"]),
+        card("KV-Cache TTFT (LMCache)", kv["label"], kv_body, kv["ttft_before_after_note"]),
+        card("LiteLLM Gateway", gw["label"], gw_body, gw["note"]),
+        card("RouteLLM Router (Thompson)", rt["label"], rt_body, rt["note"]),
+        card("Carbon-Aware Schedule", ca["label"], ca_body, ca["note"]),
+    ])
+    d = p["doctrine_lock"]
+    st = p["inference_state"]
+    # NOTE: window.SZLLabels has no native MEASURED/ROADMAP keys, so we map
+    # MEASURED→LIVE (ok tone) and ROADMAP→EXPERIMENTAL (warn tone) with a label override.
+    return """<!doctype html>
+<html lang="en"><head><meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1.0, viewport-fit=cover">
+<meta name="theme-color" content="#0a0e14"><title>Energy / Sovereign Compute — a11oy</title>
+<style>
+ :root{color-scheme:dark} *{box-sizing:border-box}
+ body{margin:0;background:#0a0e14;color:#e6edf3;font:16px/1.5 -apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,sans-serif;
+   padding:max(16px,env(safe-area-inset-top)) 16px calc(24px + env(safe-area-inset-bottom))}
+ header{max-width:980px;margin:0 auto 18px}
+ h1{font-size:clamp(1.4rem,5vw,2rem);margin:0 0 6px}
+ .summary{font-weight:700;font-size:1.05rem;color:#42d392;margin:0 0 4px}
+ .sub{color:#9aa7b4;font-size:.85rem;margin:.2rem 0}
+ .state{font-family:ui-monospace,monospace;font-size:.76rem;color:#6b7785;word-break:break-word}
+ .grid{max-width:980px;margin:0 auto;display:grid;gap:14px;grid-template-columns:1fr}
+ @media(min-width:720px){.grid{grid-template-columns:1fr 1fr}}
+ .card{background:#111722;border:1px solid #1e2a3a;border-radius:14px;padding:15px}
+ .row{display:flex;align-items:center;justify-content:space-between;gap:10px;margin-bottom:8px}
+ h3{font-size:1rem;margin:0}
+ .kv{display:flex;justify-content:space-between;gap:12px;font-size:.84rem;padding:3px 0;border-bottom:1px solid #18222e}
+ .kv .k{color:#9aa7b4} .kv .v{color:#e6edf3;font-family:ui-monospace,monospace;text-align:right;word-break:break-word}
+ .kv .v code{font-size:.78rem;color:#9fd0ff} .kv .v small{color:#6b7785}
+ .note{color:#c4cdd6;font-size:.8rem;margin:10px 0 0}
+ footer{max-width:980px;margin:22px auto 0;color:#6b7785;font-size:.76rem}
+ .lock{font-family:ui-monospace,monospace;color:#9aa7b4}
+</style></head><body>
+<header>
+  <h1>Energy / Sovereign Compute</h1>
+  <p class="summary">__SUMMARY__</p>
+  <p class="sub">Each tile is read LIVE from the on-box vLLM <code>/metrics</code> when the sovereign GPU probe is reachable, else honestly labeled <b>ROADMAP</b>. No meter → no number.</p>
+  <p class="state">inference=__INF__ · sovereign=__SOV__ · gpu_reachable=__REACH__ · measured=__MC__/__TC__</p>
+</header>
+<main class="grid">__CARDS__</main>
+<footer>
+  <p class="lock">Doctrine __DV__ LOCKED · locked-proven=__LC__ {__LP__} · __CORPUS__ @ __KC__ · Λ = Conjecture 1 (NOT a theorem) · __SLSA__</p>
+  <p>MEASURED = real on-box exporter sample (live) · ROADMAP = wiring ready, box not emitting yet (never faked). Sources: Watt-Counts arXiv:2604.09048 · Energy-per-Token arXiv:2603.20224 · vLLM spec-decode · LMCache · LiteLLM · RouteLLM · Carbon-Aware SDK.</p>
+</footer>
+<script src="/static/shared/szl_label_engine.js"></script>
+<script>
+(function(){
+  function pill(label){
+    var key = (label === "MEASURED") ? "LIVE" : "EXPERIMENTAL";
+    if (window.SZLLabels && window.SZLLabels.badgeHTML){
+      return window.SZLLabels.badgeHTML(key, {label: label,
+        title: (label === "MEASURED")
+          ? "Real on-box exporter sample — live and honest."
+          : "Wiring ready; the box is not emitting this metric yet. ROADMAP, never faked."});
+    }
+    return '<span>' + label + '</span>';
+  }
+  if (window.SZLLabels && window.SZLLabels.ensureStyle){ window.SZLLabels.ensureStyle(document); }
+  var slots = document.querySelectorAll('.pill-slot');
+  for (var i=0;i<slots.length;i++){ slots[i].innerHTML = pill(slots[i].getAttribute('data-label')); }
+})();
+</script>
+</body></html>""".replace("__SUMMARY__", p["summary"]) \
+    .replace("__INF__", str(st.get("inference"))) \
+    .replace("__SOV__", str(p["sovereign"])) \
+    .replace("__REACH__", str(p["gpu_reachable"])) \
+    .replace("__MC__", str(p["measured_panels"])) \
+    .replace("__TC__", str(p["total_panels"])) \
+    .replace("__CARDS__", cards) \
+    .replace("__DV__", d["version"]) \
+    .replace("__LC__", str(d["locked_count"])) \
+    .replace("__LP__", ", ".join(d["locked_proven"])) \
+    .replace("__CORPUS__", d["corpus"]) \
+    .replace("__KC__", d["kernel_commit"]) \
+    .replace("__SLSA__", d["slsa"])
+
+
+# ---------------------------------------------------------------------------
+# Registration (additive; mirrors szl_sovereign_compute.register).
+# ---------------------------------------------------------------------------
+def register(app, ns: str = "a11oy") -> dict:
+    from fastapi.responses import HTMLResponse, JSONResponse
+
+    base = "/api/%s/v1/energy" % ns
+
+    @app.get("%s/sovereign" % base)
+    async def _es_json():  # noqa: ANN202
+        return JSONResponse(_posture())
+
+    @app.get("%s/jtoken" % base)
+    async def _es_jtoken():  # noqa: ANN202
+        state = _sovereign_state()
+        reachable = _gpu_reachable(state)
+        prom = _parse_prom(_fetch_metrics_text() or "") if reachable else {}
+        sample = _exporter_sample_from_metrics(prom) if reachable else None
+        return JSONResponse(_jtoken_from_metrics(prom, sample))
+
+    @app.get("%s/throughput" % base)
+    async def _es_throughput():  # noqa: ANN202
+        state = _sovereign_state()
+        reachable = _gpu_reachable(state)
+        prom = _parse_prom(_fetch_metrics_text() or "") if reachable else {}
+        return JSONResponse(_throughput_panel(prom, reachable))
+
+    @app.get("%s/kvcache" % base)
+    async def _es_kvcache():  # noqa: ANN202
+        state = _sovereign_state()
+        reachable = _gpu_reachable(state)
+        prom = _parse_prom(_fetch_metrics_text() or "") if reachable else {}
+        return JSONResponse(_kvcache_panel(prom, reachable))
+
+    @app.get("%s/gateway" % base)
+    async def _es_gateway():  # noqa: ANN202
+        return JSONResponse(_gateway_panel(_sovereign_state()))
+
+    @app.get("%s/router" % base)
+    async def _es_router():  # noqa: ANN202
+        return JSONResponse(_router_panel())
+
+    @app.get("%s/carbon" % base)
+    async def _es_carbon():  # noqa: ANN202
+        return JSONResponse(_carbon_panel())
+
+    @app.get("/energy", response_class=HTMLResponse)
+    async def _es_panel():  # noqa: ANN202
+        return HTMLResponse(_html(_posture()))
+
+    return {"ok": True, "ns": ns,
+            "routes": ["%s/sovereign" % base, "%s/jtoken" % base, "%s/throughput" % base,
+                       "%s/kvcache" % base, "%s/gateway" % base, "%s/router" % base,
+                       "%s/carbon" % base, "/energy"]}
+
+
+# ---------------------------------------------------------------------------
+# No-server self-test (proves the honesty gates without a live GPU).
+# ---------------------------------------------------------------------------
+def _selftest() -> dict:
+    out: dict = {}
+    # (a) energy_fields_for_receipt is always present + honest when no GPU.
+    f = energy_fields_for_receipt()
+    assert f["joules_consumed"] is None and f["carbon_g_co2eq"] is None, f
+    assert f["energy_label"] == "ROADMAP", f
+    assert f["joules_honesty"] == "sample", f
+    out["receipt_fields_honest_roadmap"] = True
+
+    # (b) J/token with no real sample => ROADMAP, no fabricated number.
+    jt = _jtoken_from_metrics({}, None)
+    assert jt["label"] == "ROADMAP" and jt["joules_per_token"] is None, jt
+    out["jtoken_roadmap_no_number"] = True
+
+    # (c) J/token WITH a fresh fabricated-but-real-shaped sample + token counter => MEASURED.
+    prom = {"vllm:generation_tokens_total": 1000.0}
+    sample = {"joules_measured_total": 50.0, "exporter_node": "rig-0",
+              "exporter_last_seen_ts": _time.time(), "power_w_sample": 210.0}
+    jt2 = _jtoken_from_metrics(prom, sample)
+    assert jt2["label"] == "MEASURED", jt2
+    assert abs(jt2["joules_per_token"] - 0.05) < 1e-9, jt2
+    assert jt2["carbon_g_co2eq_per_token"] is not None, jt2
+    out["jtoken_measured_with_real_sample"] = True
+
+    # (d) speculative speedup model: S=(k+1)/(k(1-α)+1); k=4, α=0.8 -> 5/1.8 ≈ 2.78x.
+    s = _spec_speedup(4, 0.8)
+    assert 2.7 < s < 2.85, s
+    out["spec_speedup_model"] = round(s, 3)
+
+    # (e) Thompson router starts at Beta(1,1) priors => ROADMAP; recording flips MEASURED.
+    rp = _router_panel()
+    assert rp["label"] == "ROADMAP", rp
+    record_route_outcome("local-7b", True)
+    rp2 = _router_panel()
+    assert rp2["label"] == "MEASURED", rp2
+    # reset prior so the module ships clean
+    _ROUTER_MODELS["local-7b"]["alpha"] = 1.0
+    globals()["_ROUTER_OBS"] = 0
+    out["router_thompson_gate"] = True
+
+    # (f) carbon panel: no live feed => ROADMAP + SAMPLE intensity label.
+    cp = _carbon_panel()
+    assert cp["label"] == "ROADMAP" and "SAMPLE" in cp["carbon_intensity_label"], cp
+    assert len(cp["schedule_24h"]) == 24, cp
+    out["carbon_roadmap_sample"] = True
+
+    # (g) full posture renders + html is non-trivial + no forbidden raw claims pattern.
+    p = _posture()
+    h = _html(p)
+    assert "Energy / Sovereign Compute" in h and len(h) > 2000, len(h)
+    assert "100%" not in h and "tamper-proof" not in h.lower(), "forbidden raw claim"
+    out["html_bytes"] = len(h)
+
+    out["ok"] = True
+    return out
+
+
+if __name__ == "__main__":
+    import sys
+    print(_json.dumps(_selftest(), indent=2), file=sys.stderr)