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, src/a11oy/harvest/__init__.py, src/a11oy/harvest/harvest_budget.py, src/a11oy/harvest/wasted_energy_harvest.py, szl_anatomy_loop.py
Deleted (gone from the repo + Dockerfile COPY set): (none)

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

@@ -311,6 +311,9 @@ COPY src/a11oy/formulas/bloom_filter.py ./src/a11oy/formulas/bloom_filter.py
 COPY src/a11oy/formulas/kalman.py ./src/a11oy/formulas/kalman.py
 COPY src/a11oy/formulas/hnsw_retrieval.py ./src/a11oy/formulas/hnsw_retrieval.py
 COPY src/a11oy/formulas/reidemeister.py ./src/a11oy/formulas/reidemeister.py
+COPY src/a11oy/harvest/__init__.py ./src/a11oy/harvest/__init__.py
+COPY src/a11oy/harvest/wasted_energy_harvest.py ./src/a11oy/harvest/wasted_energy_harvest.py
+COPY src/a11oy/harvest/harvest_budget.py ./src/a11oy/harvest/harvest_budget.py
 # ADDITIVE (Formulas SECTION page — closeout): serve.py imports a11oy_formulas_page
 # and calls .register(app) BEFORE the SPA catch-all, mounting GET /formulas/wired
 # (premium Inca-palette list of every live formula + thesis citation + Lean permalink
@@ -670,7 +673,7 @@ COPY szl_connectors/ ./szl_connectors/
 COPY szl_hf_bucket.py szl_metrics_prom.py ./
 # Forge fix: these modules are on main + imported by serve.py (try/except) but were NEVER COPY'd
 # into the image -> ModuleNotFoundError at startup -> /api/a11oy/v1/research/* + dark surfaces 404.
-COPY szl_research_infra.py szl_dark_surfaces_register.py ./
+COPY szl_research_infra.py szl_dark_surfaces_register.py szl_anatomy_loop.py ./
 
 
 CMD ["python", "serve.py"]

src/a11oy/harvest/__init__.py

@@ -0,0 +1,31 @@
+# SPDX-License-Identifier: Apache-2.0
+# © 2026 Lutar, Stephen P. — SZL Holdings
+"""a11oy.harvest — vendored wasted-energy harvest module.
+
+Doctrine (binding):
+  - NO free-energy / over-unity. Harvests ALREADY-WASTED grid energy only.
+  - joules_label ALWAYS "sample" off-box; only on-box NVML flips it to "measured".
+  - All feeds are FREE and PUBLIC (no token).
+  - Reactive turns are NEVER gated by harvest posture.
+  - Locked-8 theorems untouched.
+"""
+from .wasted_energy_harvest import (
+    current_harvest_posture,
+    harvest_provenance,
+    scan_world_renshare,
+    HarvestPosture,
+    FeedReading,
+    POSTURE_RANK,
+)
+from .harvest_budget import plan_soak, SoakPlan
+
+__all__ = [
+    "current_harvest_posture",
+    "harvest_provenance",
+    "scan_world_renshare",
+    "HarvestPosture",
+    "FeedReading",
+    "POSTURE_RANK",
+    "plan_soak",
+    "SoakPlan",
+]

src/a11oy/harvest/harvest_budget.py

@@ -0,0 +1,735 @@
+"""harvest_budget.py — Proven sponge budget for the wasted-energy harvest.
+
+Grounds the soak-window batch admission in the founder's own kernel-proven
+formulas so the harvest is PROVABLY BOUNDED, not just heuristic.
+
+Doctrine (binding, v11/v12):
+  - NO free-energy / over-unity.  This computes an INFORMATION cap (Bekenstein,
+    proven by Lean) and an ENERGY FLOOR (Landauer, proven by Lean). Joule
+    figures are labelled SAMPLE/ESTIMATE until a real on-box NVML meter feeds them.
+  - The Bekenstein cap is on INFORMATION (bits) — this IS provable and the Lean
+    theorem is kernel-checked.  The energy floor (Landauer) is also proven.
+    Joule figures stay SAMPLE; information figures are bounded by proven math.
+  - The soak loop is bounded by an Ouroboros-style hard-max + budget counter so
+    the sponge can NEVER run away. Reactive turns always preempt.
+  - Locked-proven set stays EXACTLY 8 (F1,F4,F7,F11,F12,F18,F19,F22 @ c7c0ba17).
+    This module cites those theorems but does NOT modify or add to the locked-8.
+
+Formula citations (all kernel-proven / machine-checked):
+  ┌─────────────────────────────────────────────────────────────────────────────┐
+  │ BEKENSTEIN CAP (information bound)                                          │
+  │   Lean theorem: bekenstein_bound_additive (n m : Nat) :                    │
+  │     bekensteinBits (n + m) = bekensteinBits n + bekensteinBits m           │
+  │     ∧ bekensteinBits n ≤ bekensteinBits (n + m)                            │
+  │   Lean theorem: info_within_bound (n m e : Nat) (h : e ≤ bekensteinBits n) │
+  │     : e ≤ bekensteinBits (n + m)                                           │
+  │   Source: Showcase/Frontier/EnergyBudgetWitness.lean                       │
+  │           lutar-lean PR #239 (KEYSTONE, 0-sorry)                           │
+  │   Composition: mirrors F19 (f19_budget_monotone, s1 ≤ s1+s2)              │
+  ├─────────────────────────────────────────────────────────────────────────────┤
+  │ LANDAUER FLOOR (energy lower bound)                                         │
+  │   Lean theorem: landauer_floor_pos (n q : Nat) (hn : 0<n) (hq : 0<q) :   │
+  │     0 < energyFloor n q                                                    │
+  │   Lean def:     energyFloor n q = n * q  (integer shadow of n·kT·ln2)     │
+  │   Source: Showcase/Frontier/LandauerFloorWitness.lean                      │
+  │           lutar-lean PR #240                                                │
+  ├─────────────────────────────────────────────────────────────────────────────┤
+  │ MONOTONE LEDGER                                                             │
+  │   Lean theorem: energy_ledger_monotone (start : Nat) (draws : List Nat) :  │
+  │     start ≤ ledgerSum start draws                                          │
+  │   Source: Showcase/Frontier/EnergyBudgetWitness.lean (PR #239)            │
+  │   Composition: mirrors F19 (f19_budget_monotone)                           │
+  ├─────────────────────────────────────────────────────────────────────────────┤
+  │ OUROBOROS BOUNDED RECURSION                                                 │
+  │   Source: szl-holdings/ouroboros packages/ouroboros/src/loop-kernel.ts    │
+  │   Primitive: runLoop({ maxSteps }) — the loop NEVER exceeds maxSteps       │
+  │   iterations; exits 'budgetExhausted' when the cap is hit.  We mirror      │
+  │   this exact pattern: OUROBOROS_MAX_SOAK_STEPS hard cap + a work-budget    │
+  │   counter so the proactive sponge loop halts unconditionally.              │
+  └─────────────────────────────────────────────────────────────────────────────┘
+"""
+from __future__ import annotations
+
+import math
+import datetime
+from dataclasses import dataclass, field
+from typing import Any, Optional
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+# Boltzmann constant * room temperature (300 K) * ln 2 — the Landauer minimum
+# energy per irreversible bit erasure.  SAMPLE until an on-box calorimeter
+# feeds real values.  Used only to ensure SAMPLE estimates never CLAIM to beat
+# the floor; the floor itself is the proven lower bound.
+_K_B = 1.380649e-23   # J/K  (exact by SI 2019 redefinition)
+_LN2 = 0.6931471805599453
+
+# Default room temperature for Landauer floor calculations.
+LANDAUER_DEFAULT_TEMP_K: float = 300.0
+
+# Ouroboros bounded-recursion cap.
+# Mirrors loop-kernel.ts DEFAULT_MAX_STEPS=8 / runLoop({ maxSteps }) hard cap.
+# The proactive soak loop NEVER exceeds this iteration count.
+OUROBOROS_MAX_SOAK_STEPS: int = 32   # generous but hard; reactive always preempts
+
+# ---------------------------------------------------------------------------
+# Part 1 — Bekenstein information cap
+# ---------------------------------------------------------------------------
+# Python mirror of:
+#   def bekensteinBits (n : Nat) : Nat := n * 8
+#   theorem bekenstein_bound_additive ...  [PR #239, EnergyBudgetWitness.lean]
+#   theorem info_within_bound ...          [PR #239, EnergyBudgetWitness.lean]
+
+
+def bekenstein_info_cap(n_bytes: int) -> int:
+    """Return the Bekenstein bit-cap for an n-byte output register: n * 8.
+
+    Python mirror of `bekensteinBits (n : Nat) : Nat := n * 8` from
+    Showcase/Frontier/EnergyBudgetWitness.lean (lutar-lean PR #239, keystone,
+    0-sorry). This is the maximum information that can be carried by an n-byte
+    register — the ceiling any admitted batch job must stay within.
+
+    Citation:
+      bekenstein_bound_additive (n m : Nat) :
+        bekensteinBits (n+m) = bekensteinBits n + bekensteinBits m
+        ∧ bekensteinBits n ≤ bekensteinBits (n+m)
+      [EnergyBudgetWitness.lean, PR #239, F19-composition]
+    """
+    if n_bytes < 0:
+        raise ValueError(f"n_bytes must be non-negative, got {n_bytes}")
+    return n_bytes * 8
+
+
+class BekensteinAccumulator:
+    """Additive Bekenstein-cap accumulator for a soak window.
+
+    Implements the info_within_bound / bekenstein_bound_additive pattern:
+    each admitted job contributes its info load to a running total; a new job
+    is REFUSED if admitting it would push the total past the window cap.
+
+    All information figures are in BITS (integers, matching the Lean Nat model).
+
+    Citation:
+      info_within_bound (n m e : Nat) (h : e ≤ bekensteinBits n) :
+        e ≤ bekensteinBits (n + m)
+      bekenstein_bound_additive ...
+      [Showcase/Frontier/EnergyBudgetWitness.lean, lutar-lean PR #239, 0-sorry]
+    """
+
+    def __init__(self, window_cap_bytes: int) -> None:
+        """
+        Args:
+            window_cap_bytes: Total byte budget for the soak window. The
+                Bekenstein cap is window_cap_bytes * 8 bits.
+        """
+        if window_cap_bytes <= 0:
+            raise ValueError(f"window_cap_bytes must be positive, got {window_cap_bytes}")
+        self._cap_bits: int = bekenstein_info_cap(window_cap_bytes)
+        self._used_bits: int = 0
+        self._admitted: list[dict] = []
+
+    @property
+    def cap_bits(self) -> int:
+        return self._cap_bits
+
+    @property
+    def used_bits(self) -> int:
+        return self._used_bits
+
+    @property
+    def remaining_bits(self) -> int:
+        return max(0, self._cap_bits - self._used_bits)
+
+    def try_admit(self, job_id: Any, job_info_bits: int) -> bool:
+        """Attempt to admit a job that will process job_info_bits bits.
+
+        Returns True (admitted) iff used_bits + job_info_bits ≤ cap_bits,
+        mirroring info_within_bound: if e ≤ bekensteinBits n then e ≤ cap.
+        Returns False and does NOT update state when the job would exceed the cap.
+
+        Citation: bekenstein_bound_additive / info_within_bound
+          [EnergyBudgetWitness.lean, lutar-lean PR #239, keystone, 0-sorry]
+        """
+        if job_info_bits < 0:
+            raise ValueError(f"job_info_bits must be non-negative, got {job_info_bits}")
+        if self._used_bits + job_info_bits > self._cap_bits:
+            return False
+        self._used_bits += job_info_bits
+        self._admitted.append({"job_id": job_id, "info_bits": job_info_bits})
+        return True
+
+    def admitted_jobs(self) -> list[dict]:
+        return list(self._admitted)
+
+
+# ---------------------------------------------------------------------------
+# Part 2 — Landauer floor
+# ---------------------------------------------------------------------------
+# Python mirror of:
+#   def energyFloor (n q : Nat) : Nat := n * q
+#   theorem landauer_floor_pos ...         [PR #240, LandauerFloorWitness.lean]
+
+
+def landauer_floor_joules(bits: int, temp_K: float = LANDAUER_DEFAULT_TEMP_K) -> float:
+    """Minimum energy (joules, SAMPLE) to irreversibly erase `bits` bits at temp_K.
+
+    Computes bits * k_B * T * ln2, the Landauer minimum.  This is a LOWER BOUND;
+    a real system dissipates at least this much.  A SAMPLE joule estimate for
+    any batch job must never CLAIM to beat this floor (anti-over-unity guard).
+
+    The value returned is labelled SAMPLE/ESTIMATE because it uses nominal
+    k_B and T; a real on-box calorimeter is needed for a measured value.
+
+    Citation:
+      energyFloor (n q : Nat) : Nat := n * q  (integer shadow of n·kT·ln2)
+      landauer_floor_pos (n q : Nat) (hn : 0<n) (hq : 0<q) : 0 < energyFloor n q
+      landauer_floor_additive (n m q : Nat) : energyFloor (n+m) q = ...
+      [Showcase/Frontier/LandauerFloorWitness.lean, lutar-lean PR #240]
+    """
+    if bits < 0:
+        raise ValueError(f"bits must be non-negative, got {bits}")
+    if temp_K <= 0:
+        raise ValueError(f"temp_K must be positive, got {temp_K}")
+    return bits * _K_B * temp_K * _LN2  # SAMPLE/ESTIMATE (doctrine)
+
+
+def assert_sample_beats_landauer_floor(
+    bits: int,
+    sample_joules: float,
+    temp_K: float = LANDAUER_DEFAULT_TEMP_K,
+) -> None:
+    """Assert that a SAMPLE joule estimate does NOT claim to beat the Landauer floor.
+
+    If sample_joules < landauer_floor_joules(bits, temp_K), this raises
+    AssertionError — the estimate would be physically impossible (over-unity /
+    free-energy claim), violating doctrine.
+
+    Note: sample_joules=0 (unknown / not estimated) always passes — zero means
+    "not yet measured", not "costs nothing". Pass the actual SAMPLE estimate
+    only when you have one.
+
+    Citation: LandauerFloorWitness.lean, lutar-lean PR #240
+    """
+    if sample_joules <= 0:
+        return  # 0 / negative = not estimated; floor check not applicable
+    floor = landauer_floor_joules(bits, temp_K)
+    assert sample_joules >= floor, (
+        f"DOCTRINE VIOLATION: SAMPLE joule estimate {sample_joules:.3e} J "
+        f"claims to beat the Landauer floor {floor:.3e} J for {bits} bits at "
+        f"{temp_K} K. This would be a free-energy / over-unity claim. "
+        f"[LandauerFloorWitness.lean, PR #240]"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Part 3 — Monotone SoakLedger
+# ---------------------------------------------------------------------------
+# Python mirror of:
+#   theorem energy_ledger_monotone (start : Nat) (draws : List Nat) :
+#     start ≤ ledgerSum start draws
+#   [EnergyBudgetWitness.lean, PR #239, F19-composition: f19_budget_monotone]
+
+
+@dataclass
+class LedgerEntry:
+    """One append-only entry in the SoakLedger."""
+    job_id: Any
+    info_bits: int                        # proven-bounded
+    joules_sample: float                  # SAMPLE/ESTIMATE until NVML
+    joules_label: str = "sample"          # NEVER changes
+    timestamp_utc: str = field(default_factory=lambda: datetime.datetime.now(datetime.timezone.utc).isoformat())
+    landauer_floor_joules: float = 0.0    # floor for audit
+    beats_floor: bool = True              # False = over-unity claim detected
+
+
+class SoakLedger:
+    """Append-only monotone ledger for the wasted-energy soak.
+
+    Models `energy_ledger_monotone` from EnergyBudgetWitness.lean (PR #239):
+    the running cumulative totals (info bits, sample joules) are monotone-
+    nondecreasing — appending a new draw never lowers the total.
+
+    A monotonicity assertion is checked on every append.
+
+    Citation:
+      energy_ledger_monotone (start : Nat) (draws : List Nat) :
+        start ≤ ledgerSum start draws
+      ledger_step_monotone (s d : Nat) : s ≤ s + d  (= f19_budget_monotone)
+      [Showcase/Frontier/EnergyBudgetWitness.lean, lutar-lean PR #239, 0-sorry]
+    """
+
+    def __init__(self) -> None:
+        self._entries: list[LedgerEntry] = []
+        self._total_info_bits: int = 0
+        self._total_joules_sample: float = 0.0
+
+    @property
+    def total_info_bits(self) -> int:
+        return self._total_info_bits
+
+    @property
+    def total_joules_sample(self) -> float:
+        return self._total_joules_sample
+
+    @property
+    def entries(self) -> list[LedgerEntry]:
+        return list(self._entries)
+
+    def append(
+        self,
+        job_id: Any,
+        info_bits: int,
+        joules_sample: float = 0.0,
+        temp_K: float = LANDAUER_DEFAULT_TEMP_K,
+    ) -> LedgerEntry:
+        """Append one entry; assert monotonicity after each append.
+
+        Mirrors `ledger_step_monotone`: prev_total ≤ prev_total + draw.
+        Checks that the SAMPLE joule estimate does not beat the Landauer floor.
+
+        Joules are labelled SAMPLE/ESTIMATE and never treated as measured.
+        """
+        if info_bits < 0:
+            raise ValueError(f"info_bits must be non-negative, got {info_bits}")
+        if joules_sample < 0:
+            raise ValueError(f"joules_sample must be non-negative, got {joules_sample}")
+
+        prev_bits = self._total_info_bits
+        prev_joules = self._total_joules_sample
+
+        # Landauer floor guard (anti-over-unity)
+        floor = landauer_floor_joules(info_bits, temp_K)
+        beats_floor = True
+        if joules_sample > 0 and joules_sample < floor:
+            beats_floor = False  # flag; do not raise (SAMPLE may be zero)
+
+        entry = LedgerEntry(
+            job_id=job_id,
+            info_bits=info_bits,
+            joules_sample=joules_sample,
+            joules_label="sample",
+            landauer_floor_joules=floor,
+            beats_floor=beats_floor,
+        )
+
+        self._total_info_bits += info_bits
+        self._total_joules_sample += joules_sample
+        self._entries.append(entry)
+
+        # Monotonicity assertion — mirrors energy_ledger_monotone
+        assert self._total_info_bits >= prev_bits, (
+            f"SoakLedger monotonicity violated on info_bits: "
+            f"{prev_bits} → {self._total_info_bits} [EnergyBudgetWitness.lean PR #239]"
+        )
+        assert self._total_joules_sample >= prev_joules, (
+            f"SoakLedger monotonicity violated on joules: "
+            f"{prev_joules} → {self._total_joules_sample} [EnergyBudgetWitness.lean PR #239]"
+        )
+
+        return entry
+
+    def provenance(self) -> dict:
+        """Return a receipt-shaped provenance block for the full soak window."""
+        return {
+            "soak_jobs": len(self._entries),
+            "total_info_bits": self._total_info_bits,
+            "total_joules_sample": self._total_joules_sample,
+            "joules_label": "sample",
+            "ledger_monotone": True,
+            "citation": (
+                "energy_ledger_monotone / ledger_step_monotone "
+                "[EnergyBudgetWitness.lean, lutar-lean PR #239, 0-sorry]"
+            ),
+        }
+
+
+# ---------------------------------------------------------------------------
+# Part 4 — Ouroboros bounded-recursion soak loop guard
+# ---------------------------------------------------------------------------
+# Models loop-kernel.ts runLoop({ maxSteps }):
+#   "runs steps until … we hit maxSteps ('budgetExhausted')"
+# The proactive soak loop is capped at OUROBOROS_MAX_SOAK_STEPS; reactive
+# turns are NEVER subject to this gate and always preempt.
+
+
+@dataclass
+class OuroborosBudget:
+    """A single-use Ouroboros-style work budget for one soak session.
+
+    Mirrors the loop-kernel.ts runLoop({ maxSteps }) primitive:
+    the kernel NEVER exceeds maxSteps — it exits 'budgetExhausted' when the
+    cap is hit.  This budget wraps the Python soak loop equivalently.
+
+    Citation:
+      szl-holdings/ouroboros packages/ouroboros/src/loop-kernel.ts
+      runLoop({ maxSteps }) — exits 'budgetExhausted' at hard cap
+    """
+    max_steps: int = OUROBOROS_MAX_SOAK_STEPS
+    _steps_taken: int = field(default=0, init=False, repr=False)
+    _exhausted: bool = field(default=False, init=False, repr=False)
+
+    def step(self) -> bool:
+        """Consume one step. Returns True if the loop may continue, False if exhausted.
+
+        Once exhausted, every subsequent call returns False — the soak loop halts
+        unconditionally, mirroring 'budgetExhausted' in loop-kernel.ts.
+        """
+        if self._exhausted or self._steps_taken >= self.max_steps:
+            self._exhausted = True
+            return False
+        self._steps_taken += 1
+        return True
+
+    @property
+    def steps_taken(self) -> int:
+        return self._steps_taken
+
+    @property
+    def is_exhausted(self) -> bool:
+        return self._exhausted or self._steps_taken >= self.max_steps
+
+    @property
+    def exit_reason(self) -> str:
+        if self._exhausted or self._steps_taken >= self.max_steps:
+            return "budgetExhausted"
+        return "running"
+
+
+# ---------------------------------------------------------------------------
+# Part 5 — plan_soak: wire everything together
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class SoakPlan:
+    """Result of plan_soak(): the admitted batch jobs and the proven bounds respected."""
+    admitted: list[dict]           # subset of input jobs admitted within bounds
+    refused: list[dict]            # jobs refused (would exceed Bekenstein cap)
+    ledger: SoakLedger
+    bekenstein_cap_bits: int
+    bekenstein_used_bits: int
+    ouroboros_steps_taken: int
+    ouroboros_max_steps: int
+    ouroboros_exit_reason: str
+    posture: str
+    wasted_energy_available: bool
+    proven_bounds_respected: list[str]
+    joules_label: str = "sample"
+    honest_note: str = (
+        "Information cap (Bekenstein) and ledger monotonicity are PROVEN by "
+        "kernel-checked Lean theorems (0-sorry). The Ouroboros loop bound is "
+        "test-checked (mirrors loop-kernel.ts maxSteps). Joule figures stay "
+        "SAMPLE/ESTIMATE until a real on-box NVML meter feeds them."
+    )
+
+
+def plan_soak(
+    window: dict,
+    jobs: list[dict],
+    window_cap_bytes: Optional[int] = None,
+    max_soak_steps: int = OUROBOROS_MAX_SOAK_STEPS,
+    temp_K: float = LANDAUER_DEFAULT_TEMP_K,
+) -> SoakPlan:
+    """Admit batch jobs into the soak window, bounded by proven formulas.
+
+    Given:
+      window  — the harvest posture dict (from harvest_posture_bridge /
+                should_soak_wasted_energy). Must have
+                wasted_energy_available=True for any jobs to be admitted.
+      jobs    — list of dicts, each with:
+                  'id'         : any hashable job identifier
+                  'info_bits'  : int, information load the job will process
+                  'joules_est' : float (optional), SAMPLE energy estimate
+      window_cap_bytes — byte budget for the soak window; Bekenstein cap =
+                window_cap_bytes * 8 bits. If None, defaults to
+                sum(job['info_bits'] for job in jobs) // 8 + 1  (admits
+                exactly the jobs that fit within the cap individually but
+                refuses those that together overflow it). Callers should
+                supply a real window byte budget.
+      max_soak_steps — Ouroboros hard cap (default OUROBOROS_MAX_SOAK_STEPS).
+      temp_K         — temperature for Landauer floor (default 300 K).
+
+    Returns:
+      SoakPlan with admitted subset, refused subset, proven bounds, ledger.
+
+    Proven bounds respected (in order):
+      1. wasted_energy_available gate (posture check)
+      2. Bekenstein additive cap (information bound, Lean proven PR #239)
+      3. Landauer floor (energy lower bound, Lean proven PR #240)
+      4. SoakLedger monotonicity (append-only, Lean proven PR #239)
+      5. Ouroboros bounded-recursion cap (loop-kernel.ts maxSteps pattern)
+    """
+    posture = window.get("posture", "normal")
+    wasted = bool(window.get("wasted_energy_available", False))
+
+    # Default cap: sum of all job info bits translated back to bytes (generous)
+    if window_cap_bytes is None:
+        total_bits = sum(j.get("info_bits", 0) for j in jobs)
+        # Use sum+1 as the cap; the additive check still refuses over-cap combos
+        window_cap_bytes = max(1, (total_bits // 8) + 1)
+
+    accumulator = BekensteinAccumulator(window_cap_bytes)
+    ledger = SoakLedger()
+    budget = OuroborosBudget(max_steps=max_soak_steps)
+
+    admitted: list[dict] = []
+    refused: list[dict] = []
+    proven_bounds: list[str] = []
+
+    # Gate 1: posture check — only soak when wasted energy is available
+    if not wasted:
+        proven_bounds.append(
+            "posture_gate: wasted_energy_available=False — no jobs admitted "
+            "(grid not in negative-price/curtailed window)"
+        )
+        return SoakPlan(
+            admitted=admitted,
+            refused=list(jobs),
+            ledger=ledger,
+            bekenstein_cap_bits=accumulator.cap_bits,
+            bekenstein_used_bits=0,
+            ouroboros_steps_taken=0,
+            ouroboros_max_steps=max_soak_steps,
+            ouroboros_exit_reason="posture_gate",
+            posture=posture,
+            wasted_energy_available=wasted,
+            proven_bounds_respected=proven_bounds,
+        )
+
+    proven_bounds.append(
+        "posture_gate: wasted_energy_available=True — soak window open"
+    )
+
+    # Main bounded soak loop — capped by Ouroboros budget
+    for job in jobs:
+        # Gate 2: Ouroboros bounded-recursion cap
+        if not budget.step():
+            # budgetExhausted — remaining jobs refused (not processed)
+            refused.append({**job, "refused_reason": "ouroboros_budget_exhausted"})
+            continue
+
+        job_id = job.get("id", f"job_{len(admitted)+len(refused)}")
+        info_bits = int(job.get("info_bits", 0))
+        joules_est = float(job.get("joules_est", 0.0))
+
+        # Gate 3: Bekenstein additive cap (proven: bekenstein_bound_additive,
+        # info_within_bound — EnergyBudgetWitness.lean PR #239, 0-sorry)
+        if not accumulator.try_admit(job_id, info_bits):
+            refused.append({
+                **job,
+                "refused_reason": "bekenstein_cap_exceeded",
+                "cap_bits": accumulator.cap_bits,
+                "used_bits": accumulator.used_bits,
+                "job_info_bits": info_bits,
+            })
+            continue
+
+        # Gate 4: Landauer floor check — SAMPLE estimate must not beat the floor
+        floor = landauer_floor_joules(info_bits, temp_K)
+        if joules_est > 0 and joules_est < floor:
+            # Over-unity claim — refuse
+            refused.append({
+                **job,
+                "refused_reason": "sample_estimate_beats_landauer_floor",
+                "sample_joules": joules_est,
+                "landauer_floor_joules": floor,
+            })
+            # Undo Bekenstein accumulation for the refused job
+            accumulator._used_bits -= info_bits
+            accumulator._admitted.pop()
+            continue
+
+        # Gate 5: Append to monotone SoakLedger
+        # (proven: energy_ledger_monotone — EnergyBudgetWitness.lean PR #239)
+        ledger.append(job_id, info_bits, joules_est, temp_K)
+        admitted.append({**job, "joules_label": "sample"})
+
+    # Capture proven bounds respected
+    proven_bounds.extend([
+        (
+            f"bekenstein_additive_cap: {accumulator.used_bits}/{accumulator.cap_bits} bits used "
+            f"[bekenstein_bound_additive / info_within_bound, "
+            f"EnergyBudgetWitness.lean, lutar-lean PR #239, 0-sorry]"
+        ),
+        (
+            f"landauer_floor: all admitted SAMPLE estimates ≥ floor "
+            f"[landauer_floor_pos, LandauerFloorWitness.lean, lutar-lean PR #240]"
+        ),
+        (
+            f"ledger_monotone: SoakLedger append-only, "
+            f"total_info_bits={ledger.total_info_bits}, "
+            f"total_joules_sample={ledger.total_joules_sample:.3e} SAMPLE "
+            f"[energy_ledger_monotone / ledger_step_monotone, "
+            f"EnergyBudgetWitness.lean, PR #239]"
+        ),
+        (
+            f"ouroboros_bound: loop halted after {budget.steps_taken}/{max_soak_steps} steps "
+            f"(exit: {budget.exit_reason}) "
+            f"[szl-holdings/ouroboros loop-kernel.ts runLoop maxSteps]"
+        ),
+    ])
+
+    return SoakPlan(
+        admitted=admitted,
+        refused=refused,
+        ledger=ledger,
+        bekenstein_cap_bits=accumulator.cap_bits,
+        bekenstein_used_bits=accumulator.used_bits,
+        ouroboros_steps_taken=budget.steps_taken,
+        ouroboros_max_steps=max_soak_steps,
+        ouroboros_exit_reason=budget.exit_reason,
+        posture=posture,
+        wasted_energy_available=wasted,
+        proven_bounds_respected=proven_bounds,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Self-test — no network needed; posture stubbed as negative-price
+# ---------------------------------------------------------------------------
+
+def _selftest() -> dict:
+    """Self-test: prove all five proven properties hold.
+
+    (a) Bekenstein additive cap REFUSES the over-budget job
+    (b) Landauer floor is NEVER undercut by a SAMPLE estimate
+    (c) SoakLedger is MONOTONE
+    (d) Ouroboros bound HALTS a runaway loop
+    (e) Reactive preemption still wins (soak gate=False when posture=normal)
+
+    Stubbed posture: negative-price (wasted_energy_available=True, soak_hard=True).
+    """
+    checks = 0
+
+    # --- (a) Bekenstein additive cap refuses the over-budget job ------------
+    acc = BekensteinAccumulator(window_cap_bytes=10)   # cap = 80 bits
+    assert acc.cap_bits == 80, "cap_bits should be 80"
+
+    ok1 = acc.try_admit("job_A", 40)   # 40 bits <= 80: admit
+    assert ok1, "(a) job_A should be admitted (40 bits within 80-bit cap)"
+    checks += 1
+
+    ok2 = acc.try_admit("job_B", 30)   # 40+30=70 <= 80: admit
+    assert ok2, "(a) job_B should be admitted (70 bits within 80-bit cap)"
+    checks += 1
+
+    ok3 = acc.try_admit("job_C", 20)   # 70+20=90 > 80: REFUSED
+    assert not ok3, "(a) job_C should be REFUSED (would exceed 80-bit Bekenstein cap)"
+    checks += 1
+
+    assert acc.used_bits == 70, f"(a) used_bits should be 70, got {acc.used_bits}"
+    checks += 1
+
+    # --- (b) Landauer floor is never undercut by a SAMPLE estimate ----------
+    floor_300K_1000bits = landauer_floor_joules(1000, temp_K=300.0)
+    assert floor_300K_1000bits > 0, "(b) Landauer floor must be positive"
+    checks += 1
+
+    # A legitimate estimate: 10x the floor (fine)
+    try:
+        assert_sample_beats_landauer_floor(1000, floor_300K_1000bits * 10, temp_K=300.0)
+        checks += 1
+    except AssertionError:
+        raise AssertionError("(b) Legitimate sample should NOT raise floor violation")
+
+    # An over-unity estimate: half the floor (forbidden)
+    violation_raised = False
+    try:
+        assert_sample_beats_landauer_floor(1000, floor_300K_1000bits * 0.5, temp_K=300.0)
+    except AssertionError:
+        violation_raised = True
+    assert violation_raised, "(b) Sub-floor SAMPLE estimate should raise floor violation"
+    checks += 1
+
+    # Joules=0 (unknown) always passes
+    assert_sample_beats_landauer_floor(1000, 0.0, temp_K=300.0)
+    checks += 1
+
+    # --- (c) SoakLedger is MONOTONE ----------------------------------------
+    ledger = SoakLedger()
+    assert ledger.total_info_bits == 0
+    checks += 1
+
+    ledger.append("j1", 100, joules_sample=1e-18)
+    assert ledger.total_info_bits == 100, "(c) ledger total should be 100 after j1"
+    checks += 1
+
+    ledger.append("j2", 200, joules_sample=2e-18)
+    assert ledger.total_info_bits == 300, "(c) ledger total should be 300 after j2"
+    checks += 1
+
+    ledger.append("j3", 0, joules_sample=0.0)   # zero draw: still non-decreasing
+    assert ledger.total_info_bits >= 300, "(c) zero-draw must not decrease ledger"
+    checks += 1
+
+    # Verify all entries have joules_label='sample' (doctrine)
+    for e in ledger.entries:
+        assert e.joules_label == "sample", f"(c) joules_label must be 'sample', got {e.joules_label}"
+    checks += 1
+
+    # --- (d) Ouroboros bound HALTS a runaway loop ---------------------------
+    budget = OuroborosBudget(max_steps=4)
+    iterations = 0
+    while budget.step():
+        iterations += 1
+    assert iterations == 4, f"(d) Ouroboros should halt after 4 steps, got {iterations}"
+    checks += 1
+    assert budget.is_exhausted, "(d) budget should be exhausted"
+    checks += 1
+    assert budget.exit_reason == "budgetExhausted", f"(d) exit_reason should be 'budgetExhausted', got {budget.exit_reason}"
+    checks += 1
+    # Further step() calls return False (loop cannot resume)
+    assert not budget.step(), "(d) exhausted budget.step() must return False"
+    checks += 1
+
+    # --- (e) Reactive preemption: soak gate=False when posture=normal -------
+    normal_window = {
+        "posture": "normal",
+        "wasted_energy_available": False,
+        "soak_hard": False,
+    }
+    jobs_e = [{"id": "reactive_job", "info_bits": 100, "joules_est": 0.0}]
+    plan_e = plan_soak(normal_window, jobs_e, window_cap_bytes=200)
+    assert len(plan_e.admitted) == 0, "(e) no jobs admitted when posture=normal"
+    checks += 1
+    assert len(plan_e.refused) == 1, "(e) job refused when posture=normal (reactive preemption)"
+    checks += 1
+    assert plan_e.ouroboros_exit_reason == "posture_gate", f"(e) exit should be posture_gate"
+    checks += 1
+
+    # Full plan_soak with negative-price stub
+    neg_price_window = {
+        "posture": "negative-price",
+        "wasted_energy_available": True,
+        "soak_hard": True,
+        "joules_label": "sample",
+        "source": "stub (negative-price)",
+    }
+    jobs_plan = [
+        {"id": "batch_A", "info_bits": 40, "joules_est": 1e-17},   # admitted
+        {"id": "batch_B", "info_bits": 30, "joules_est": 1e-17},   # admitted (70 total)
+        {"id": "batch_C", "info_bits": 20, "joules_est": 1e-17},   # refused (>80-bit cap)
+    ]
+    plan = plan_soak(neg_price_window, jobs_plan, window_cap_bytes=10)
+    assert len(plan.admitted) == 2, f"(a+plan) 2 jobs should be admitted, got {len(plan.admitted)}"
+    checks += 1
+    assert len(plan.refused) == 1, f"(a+plan) 1 job should be refused, got {len(plan.refused)}"
+    checks += 1
+    assert plan.refused[0]["refused_reason"] == "bekenstein_cap_exceeded", \
+        f"(a+plan) refused reason should be bekenstein_cap_exceeded"
+    checks += 1
+    assert plan.ledger.total_info_bits == 70, f"(c+plan) ledger total should be 70"
+    checks += 1
+    assert plan.joules_label == "sample", "(doctrine) joules_label must be 'sample'"
+    checks += 1
+
+    return {"ok": True, "checks": checks}
+
+
+if __name__ == "__main__":
+    import sys
+    result = _selftest()
+    print(f"ok:{str(result['ok']).lower()} checks:{result['checks']}")
+    sys.exit(0 if result["ok"] else 1)

src/a11oy/harvest/wasted_energy_harvest.py

@@ -0,0 +1,329 @@
+#!/usr/bin/env python3
+"""wasted_energy_harvest.py — jack into FREE, no-key feeds that expose WASTED energy
+(negative-price / curtailed-renewable windows) and emit an honest harvest posture.
+
+Doctrine (binding):
+  - NO free-energy / over-unity. This HARVESTS already-wasted grid energy (power the grid
+    is paying to offload because renewable supply exceeds demand). It does not create energy.
+  - All feeds here are FREE and PUBLIC (no token). Open data, legal to ingest (idea/expression).
+  - Energy figures stay SAMPLE/ESTIMATE until a real on-box meter (NVML) feeds joules.
+    This module produces a PRICE/SURPLUS POSTURE signal, not a joule measurement.
+  - The posture only GATES when to do batch work; it never asserts physical harvest.
+
+Free feeds jacked (probed live 2026-06-13, all responded):
+  - aWATTar DE/AT wholesale price (api.awattar.de|at /v1/marketdata) — negative price = wasted
+  - CAISO OASIS LMP (oasis.caiso.com/oasisapi) — US California public
+  - Energy-Charts / Fraunhofer (api.energy-charts.info) — renewable share of load (WHY it's negative)
+  - UK Carbon Intensity (api.carbonintensity.org.uk) — low-carbon surplus index
+  - Open-Meteo (api.open-meteo.com) — wind/solar weather = FORECAST of future surplus
+
+Posture levels (worst→best for harvesting):
+  expensive < normal < cheap < curtailed-renewable < negative-price
+The daemon floods Bekenstein-gated batch work when posture >= cheap; hardest when negative-price.
+"""
+from __future__ import annotations
+import json
+import urllib.request
+import datetime
+from dataclasses import dataclass, field, asdict
+from typing import Optional
+
+UA = {"User-Agent": "szl-wasted-energy-harvest/1.0 (+https://a11oy.net)"}
+TIMEOUT = 12
+
+# Posture ordering (higher index = more wasted energy available to soak)
+POSTURE_RANK = {
+    "expensive": 0,
+    "normal": 1,
+    "cheap": 2,
+    "curtailed-renewable": 3,
+    "negative-price": 4,
+}
+
+
+def _get_json(url: str) -> Optional[object]:
+    """Best-effort GET → JSON. Returns None on any failure (honest: feed unreachable
+    or returned non-JSON, e.g. a rate-limit/empty body). Never raises."""
+    try:
+        req = urllib.request.Request(url, headers=UA)
+        with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
+            body = r.read().decode("utf-8", "replace").strip()
+        if not body:
+            return None
+        return json.loads(body)
+    except Exception:
+        return None
+
+
+@dataclass
+class FeedReading:
+    feed: str
+    reachable: bool
+    measured: bool                 # True only if a REAL value was returned (not a fallback)
+    value: Optional[float] = None  # price (EUR/MWh) or share (%) depending on feed
+    unit: str = ""
+    note: str = ""
+
+
+@dataclass
+class HarvestPosture:
+    posture: str                   # one of POSTURE_RANK keys
+    rank: int
+    wasted_energy_available: bool  # True iff posture >= cheap
+    soak_hard: bool                # True iff negative-price (flood the batch sponge)
+    drivers: list = field(default_factory=list)   # human-readable reasons
+    readings: list = field(default_factory=list)  # list[FeedReading]
+    measured_any: bool = False     # at least one real feed responded
+    timestamp_utc: str = ""
+    citation: str = "FREE no-key feeds: aWATTar, CAISO OASIS, Energy-Charts/Fraunhofer (price+renshare+grid-frequency), UK Carbon Intensity, Open-Meteo forecast; Energinet (DK, intermittent) candidate"
+    doctrine: str = "harvests wasted grid energy; no free-energy claim; joules stay SAMPLE until on-box NVML meter"
+
+
+# ---- individual free jacks -------------------------------------------------
+
+def jack_awattar(country: str = "de") -> tuple[FeedReading, list[float]]:
+    """aWATTar wholesale price. Negative marketprice = the grid is PAYING to offload (wasted)."""
+    base = "https://api.awattar.de" if country == "de" else "https://api.awattar.at"
+    d = _get_json(f"{base}/v1/marketdata")
+    if not d or "data" not in d:
+        return FeedReading(f"awattar_{country}", False, False, note="unreachable"), []
+    prices = [row["marketprice"] for row in d["data"]]
+    now = prices[0] if prices else None
+    return (
+        FeedReading(f"awattar_{country}", True, True, now, "EUR/MWh",
+                    f"min_next={min(prices):.2f} neg_windows={sum(1 for p in prices if p < 0)}/{len(prices)}"),
+        prices,
+    )
+
+
+def jack_energy_charts_renshare(country: str = "de") -> FeedReading:
+    """Fraunhofer renewable share of load. High share = surplus renewables = WHY price goes negative."""
+    d = _get_json(f"https://api.energy-charts.info/ren_share?country={country}")
+    if not d or not isinstance(d, list) or not d:
+        return FeedReading("energy_charts_ren_share", False, False, note="unreachable")
+    data = d[0].get("data") if isinstance(d[0], dict) else None
+    if not data:
+        return FeedReading("energy_charts_ren_share", True, False, note="no data array")
+    cur = data[0]
+    return FeedReading("energy_charts_ren_share", True, True, float(cur), "% of load",
+                       f"max_today={max(x for x in data if x is not None):.1f}%")
+
+
+def jack_uk_carbon() -> FeedReading:
+    """UK Carbon Intensity (free). 'low' index = clean surplus on the GB grid."""
+    d = _get_json("https://api.carbonintensity.org.uk/intensity")
+    if not d or "data" not in d or not d["data"]:
+        return FeedReading("uk_carbon_intensity", False, False, note="unreachable")
+    intensity = d["data"][0].get("intensity", {})
+    return FeedReading("uk_carbon_intensity", True, True,
+                       float(intensity.get("actual") or intensity.get("forecast") or 0),
+                       "gCO2/kWh", f"index={intensity.get('index')}")
+
+
+def jack_grid_frequency(country: str = "de") -> FeedReading:
+    """Energy-Charts grid frequency (Hz). The PUREST oversupply tell: when supply
+    exceeds demand the frequency drifts ABOVE 50.00 Hz (under-frequency <50 = deficit).
+    A sustained reading >50.00 corroborates a curtailed/negative-price surplus window."""
+    d = _get_json(f"https://api.energy-charts.info/frequency?country={country}")
+    if not d or "data" not in d and "frequency" not in d:
+        # energy-charts returns {unix_seconds:[...], data:[...]} or similar; be defensive
+        freq = None
+        if isinstance(d, dict):
+            for k in ("data", "frequency", "values"):
+                v = d.get(k)
+                if isinstance(v, list) and v:
+                    freq = v[-1]
+                    break
+        if freq is None:
+            return FeedReading("grid_frequency", bool(d), False, note="reachable, no parse")
+        return FeedReading("grid_frequency", True, True, float(freq), "Hz",
+                           f"{'surplus(>50)' if float(freq) >= 50.0 else 'deficit(<50)'}")
+    arr = d.get("data") or d.get("frequency") or []
+    if not arr:
+        return FeedReading("grid_frequency", True, False, note="empty")
+    freq = float(arr[-1])
+    return FeedReading("grid_frequency", True, True, freq, "Hz",
+                       f"{'surplus(>=50)' if freq >= 50.0 else 'deficit(<50)'}")
+
+
+def jack_open_meteo_forecast(lat: float = 52.5, lon: float = 13.4) -> FeedReading:
+    """Open-Meteo (free, no key): wind speed @100m + shortwave radiation forecast.
+    High wind+sun in coming hours = a FUTURE surplus/negative-price window we can
+    PRE-SCHEDULE batch work into before the price even drops. Returns a 0-100ish
+    'surplus_outlook' score (normalized wind+solar) for the next 6h."""
+    d = _get_json(
+        f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}"
+        f"&hourly=wind_speed_100m,shortwave_radiation&forecast_days=1")
+    if not d or "hourly" not in d:
+        return FeedReading("open_meteo_forecast", False, False, note="unreachable")
+    h = d["hourly"]
+    ws = [x for x in (h.get("wind_speed_100m") or [])[:6] if x is not None]
+    sr = [x for x in (h.get("shortwave_radiation") or [])[:6] if x is not None]
+    if not ws:
+        return FeedReading("open_meteo_forecast", True, False, note="no wind data")
+    # crude normalized outlook: wind (km/h, cap 60) + solar (W/m2, cap 800)
+    wind_score = min(sum(ws) / len(ws), 60) / 60 * 50
+    solar_score = (min(sum(sr) / len(sr), 800) / 800 * 50) if sr else 0
+    outlook = round(wind_score + solar_score, 1)
+    return FeedReading("open_meteo_forecast", True, True, outlook, "surplus_outlook_0_100",
+                       f"next6h wind~{sum(ws)/len(ws):.0f}km/h solar~{(sum(sr)/len(sr)) if sr else 0:.0f}W/m2")
+
+
+# Energy-Charts covers 40+ European countries/zones from ONE no-key endpoint.
+WORLD_ZONES = ["de", "fr", "es", "it", "pl", "nl", "be", "ch", "at", "cz",
+               "dk", "no", "se", "fi", "pt", "gr", "ro", "hu", "sk", "ie"]
+
+
+def scan_world_renshare(zones: Optional[list] = None, cap: int = 8) -> dict:
+    """FOLLOW-THE-WIND scanner: find which country has the highest renewable share
+    of load right now (the deepest surplus = best place to route batch work).
+    Free, no key (Energy-Charts). Caps the number of zones probed per call to be
+    polite to the free endpoint. Returns {zone: share} for reachable zones +
+    the best zone."""
+    zones = (zones or WORLD_ZONES)[:cap]
+    shares: dict = {}
+    for z in zones:
+        d = _get_json(f"https://api.energy-charts.info/ren_share?country={z}")
+        if isinstance(d, list) and d and isinstance(d[0], dict):
+            data = d[0].get("data")
+            if data:
+                shares[z] = round(float(data[0]), 1)
+    best = max(shares, key=shares.get) if shares else None
+    return {"shares": shares, "best_zone": best,
+            "best_share": shares.get(best) if best else None,
+            "reachable": len(shares)}
+
+
+def jack_elexon_uk() -> FeedReading:
+    """UK Elexon BMRS live fuel mix (MW), no key. Nuclear baseload + wind + negative
+    interconnector exports (INT* < 0 = UK dumping surplus abroad = wasted-energy tell)."""
+    d = _get_json("https://data.elexon.co.uk/bmrs/api/v1/datasets/FUELINST?format=json")
+    rows = d.get("data") if isinstance(d, dict) else (d if isinstance(d, list) else None)
+    if not rows:
+        return FeedReading("elexon_uk_fuelmix", bool(d), False, note="reachable, no parse")
+    last = {}
+    for r in rows[-40:]:
+        last[r.get("fuelType")] = r.get("generation")
+    neg_exports = sum(v for k, v in last.items() if k and k.startswith("INT") and isinstance(v, (int, float)) and v < 0)
+    nuclear = last.get("NUCLEAR")
+    return FeedReading("elexon_uk_fuelmix", True, True, nuclear, "MW (nuclear)",
+                       f"neg_interconnector_exports={neg_exports}MW (surplus dumped abroad)")
+
+
+def jack_caiso() -> FeedReading:
+    """CAISO OASIS reachability (US California public LMP). Probe-only here (zip payload)."""
+    try:
+        req = urllib.request.Request(
+            "https://oasis.caiso.com/oasisapi/SingleZip?queryname=PRC_LMP&version=1",
+            headers=UA)
+        with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
+            ok = r.status == 200
+        return FeedReading("caiso_oasis", ok, False, note="reachable (zip payload; parse on-box)")
+    except Exception:
+        return FeedReading("caiso_oasis", False, False, note="unreachable")
+
+
+# ---- aggregator ------------------------------------------------------------
+
+def current_harvest_posture() -> HarvestPosture:
+    """Fuse the free feeds into one honest wasted-energy posture."""
+    readings: list[FeedReading] = []
+    drivers: list[str] = []
+
+    aw, prices = jack_awattar("de")
+    readings.append(aw)
+    ren = jack_energy_charts_renshare("de")
+    readings.append(ren)
+    readings.append(jack_uk_carbon())
+    freq = jack_grid_frequency("de")
+    readings.append(freq)
+    forecast = jack_open_meteo_forecast()
+    readings.append(forecast)
+    readings.append(jack_caiso())
+
+    measured_any = any(r.measured for r in readings)
+
+    # Decide posture from the strongest REAL signal we have.
+    posture = "normal"
+    if aw.measured and prices:
+        cur = prices[0]
+        min_next = min(prices)
+        if cur < 0 or min_next < 0:
+            posture = "negative-price"
+            drivers.append(f"aWATTar negative: now={cur:.2f}, min_next={min_next:.2f} EUR/MWh — grid paying to offload")
+        elif cur < 30:
+            posture = "cheap"
+            drivers.append(f"aWATTar cheap: {cur:.2f} EUR/MWh")
+        else:
+            posture = "normal"
+            drivers.append(f"aWATTar normal: {cur:.2f} EUR/MWh")
+
+    # Renewable surplus can PROMOTE cheap→curtailed-renewable (real curtailment driver).
+    if ren.measured and ren.value is not None and ren.value >= 75 and posture in ("cheap", "normal"):
+        if posture != "negative-price":
+            posture = "curtailed-renewable"
+        drivers.append(f"renewable share {ren.value:.1f}% of load — surplus wind/solar")
+    elif ren.measured and ren.value is not None:
+        drivers.append(f"renewable share {ren.value:.1f}% of load")
+
+    # Grid frequency corroboration: sustained >50 Hz = real-time oversupply.
+    if freq.measured and freq.value is not None and freq.value >= 50.0:
+        drivers.append(f"grid frequency {freq.value:.3f} Hz (>=50 = live oversupply)")
+
+    # Forecast: high surplus outlook = a soak window is coming even if price is normal now.
+    if forecast.measured and forecast.value is not None and forecast.value >= 50:
+        drivers.append(f"surplus outlook {forecast.value:.0f}/100 next 6h (pre-schedule next soak)")
+
+    rank = POSTURE_RANK[posture]
+    return HarvestPosture(
+        posture=posture,
+        rank=rank,
+        wasted_energy_available=rank >= POSTURE_RANK["cheap"],
+        soak_hard=(posture == "negative-price"),
+        drivers=drivers,
+        readings=[asdict(r) for r in readings],
+        measured_any=measured_any,
+        timestamp_utc=datetime.datetime.now(datetime.timezone.utc).isoformat(),
+    )
+
+
+def harvest_provenance() -> dict:
+    """Receipt-shaped provenance fields for the energy receipt."""
+    p = current_harvest_posture()
+    return {
+        "energy_source": "free-public-grid-feeds",
+        "posture": p.posture,
+        "wasted_energy_available": p.wasted_energy_available,
+        "soak_hard": p.soak_hard,
+        "price_measured": p.measured_any,   # price/posture is real; joules remain SAMPLE off-box
+        "joules_label": "sample",           # NEVER measured until on-box NVML
+        "drivers": p.drivers,
+        "citation": p.citation,
+    }
+
+
+if __name__ == "__main__":
+    import sys
+    p = current_harvest_posture()
+    checks = 0
+    print("=== WASTED-ENERGY HARVEST — live free-feed probe ===")
+    for r in p.readings:
+        flag = "OK " if r["reachable"] else "DOWN"
+        meas = "MEASURED" if r["measured"] else "sample/probe"
+        print(f"  [{flag}] {r['feed']:26} {meas:13} val={r['value']} {r['unit']}  {r['note']}")
+        checks += 1
+    print(f"\n  POSTURE: {p.posture}  (rank {p.rank}/4)")
+    print(f"  wasted_energy_available: {p.wasted_energy_available}")
+    print(f"  soak_hard (flood batch sponge): {p.soak_hard}")
+    print("  drivers:")
+    for d in p.drivers:
+        print(f"    - {d}")
+    # honest checks
+    assert p.posture in POSTURE_RANK, "posture must be a known level"; checks += 1
+    assert p.measured_any, "at least one free feed must be live"; checks += 1
+    prov = harvest_provenance()
+    assert prov["joules_label"] == "sample", "joules MUST stay sample off-box (doctrine)"; checks += 1
+    assert prov["energy_source"] == "free-public-grid-feeds"; checks += 1
+    print(f"\n  provenance (receipt fields): {json.dumps(prov, indent=2)[:400]}...")
+    print(f"\nok:true checks:{checks}")
+    sys.exit(0)

szl_anatomy_loop.py

@@ -0,0 +1,504 @@
+# SPDX-License-Identifier: Apache-2.0
+# © 2026 Lutar, Stephen P. — SZL Holdings · ORCID 0009-0001-0110-4173
+"""
+szl_anatomy_loop.py — the live CLOSED-LOOP energy-circulation read endpoint.
+
+This module is the ORCHESTRATOR. It does NOT reimplement any organ — it wires the
+already-shipped anatomy organs into ONE Ayni-balanced circulation loop and exposes
+its live state:
+
+  GET /api/<ns>/v1/anatomy/loop  -> the live circulation state of the loop.
+
+THE LOOP (one cycle):
+
+  harvest INTAKE  read the live wasted-energy posture (attempt local
+                  http://127.0.0.1 harvest/posture, then public
+                  https://a11oy.net/api/a11oy/v1/harvest/posture; if both are
+                  unreachable, degrade HONESTLY to a clearly-labeled SAMPLE
+                  snapshot — we NEVER fabricate a measured number)
+    -> SAMAY      lungs: SOAK the available wasted-WORK window (a breath/window),
+                  carrying soaked WORK + receipts — NOT electrons.
+    -> KALLPA     metabolize the soaked window into bounded work_credits, capped
+                  by the Bekenstein software bound (bytes*8) and floored by the
+                  Landauer minimum (kT ln 2 per bit) so a credit is always a
+                  bounded, honest unit of WORK — never free energy.
+    -> heart/pulse a DSSE beat on the HEART σ-bus + BLOOD Merkle chain
+                  (szl_heart_blood, #332), wrapping a provenance receipt.
+    -> YARQA      irrigation canal: disperse the beat to the organs
+                  WAQAYCHAQ (guard/store), KAMAY (act/animate), RIKUY (observe).
+    -> EnergyReservoir store the metabolized work_credits (a tank, not a battery
+                  of electrons — it holds proven WORK + its receipt).
+    -> provenance receipt  bind the cycle into the tamper-evident chain
+                  (szl_energy_provenance, #331).
+    -> validate   verify the provenance chain + heart beats (offline-checkable).
+    -> Ayni F11   balance the books: intake == output + stored + proven. The loop
+                  is reciprocal, NEVER net-positive. (Ayni = direct reciprocity,
+                  Axelrod & Hamilton 1981; the F11 ledger primitive.)
+    -> repeat.
+
+DOCTRINE (v11 — NON-NEGOTIABLE; CI enforces via doctrine grep + overclaim guard):
+  - joules_label is ALWAYS "sample" unless a real MEASURED source is present
+    (on-box NVML etc.). Default is sample. This module runs off-box, so the
+    default and the offline path are both sample. We do NOT invent measured
+    numbers.
+  - The loop carries soaked-WORK + receipts, NOT electrons. It is an information/
+    work circulation, not a power line.
+  - NO free-energy / perpetual / over-unity language anywhere. A cycle can only
+    metabolize what it soaked; the books MUST balance.
+  - Ayni MUST balance: intake == output + stored + proven, never net-positive.
+  - Organs are EXPERIMENTAL tier — never claimed proven.
+  - sovereign stays False unless running on own metal (never asserted here).
+  - Λ is Conjecture 1 — never a theorem, never "proven trust".
+
+Additive + import-safe + crash-proof: every organ call is wrapped in try/except so
+this module can NEVER take down the app, and it self-tests with NO network and
+without serve.py running. Pure stdlib + whatever a11oy already imports (FastAPI).
+
+It WRAPS — never rewrites — the shipped organ modules:
+  - a11oy_harvest_endpoints (#harvest posture)   -> INTAKE
+  - szl_energy_provenance   (#331 receipt chain) -> receipt + validate
+  - szl_heart_blood         (#332 heart/pulse)   -> the beat
+when importable; otherwise it falls back to the live HTTP surfaces, and finally to
+an honest SAMPLE snapshot — always labeled.
+"""
+import json
+import math
+import urllib.request
+from datetime import datetime, timezone
+
+# ---------------------------------------------------------------------------
+# Doctrine constants (v11). These are the honest, fixed labels + physics floors.
+# ---------------------------------------------------------------------------
+DOCTRINE = "v11"
+SAMPLE_LABEL = "sample"        # default — off-box, no real power meter wired
+MEASURED_LABEL = "measured"    # ONLY when a real metered source is present
+
+# Physics floors/ceilings for a bounded WORK credit (NOT free energy):
+#   Bekenstein software bound  : a window of N bytes carries at most N*8 bits.
+#   Landauer minimum           : erasing one bit costs at least kT ln 2 joules.
+# These BOUND a credit; they never manufacture energy.
+_BOLTZMANN_K = 1.380649e-23          # J/K (CODATA)
+_ROOM_T_KELVIN = 300.0               # K — a labeled SAMPLE ambient, not metered
+LANDAUER_FLOOR_J = _BOLTZMANN_K * _ROOM_T_KELVIN * math.log(2)  # ~2.87e-21 J/bit
+
+# The three YARQA dispersal organs. EXPERIMENTAL tier — never claimed proven.
+_ORGAN_SPECS = (
+    ("WAQAYCHAQ", "guard/store — disperse the beat to the reservoir-guard organ (experimental)"),
+    ("KAMAY",     "act/animate — disperse the beat to the actuation organ (experimental)"),
+    ("RIKUY",     "observe — disperse the beat to the observation/telemetry organ (experimental)"),
+)
+
+# Candidate live posture surfaces, in attempt order: local box first, then public.
+_POSTURE_URLS = (
+    "http://127.0.0.1/api/a11oy/v1/harvest/posture",
+    "http://127.0.0.1:8000/api/a11oy/v1/harvest/posture",
+    "https://a11oy.net/api/a11oy/v1/harvest/posture",
+)
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# ---------------------------------------------------------------------------
+# INTAKE — read the live harvest posture, degrade HONESTLY to a SAMPLE snapshot.
+# Never fabricate a measured number; the SAMPLE snapshot is clearly labeled.
+# ---------------------------------------------------------------------------
+def _sample_posture() -> dict:
+    """An honest, clearly-labeled SAMPLE intake snapshot (no live feed reached).
+
+    Every field is labeled sample; no number here is claimed as metered. This is
+    the doctrine-clean degrade path when neither the local box nor the public
+    surface is reachable.
+    """
+    return {
+        "ok": False,
+        "posture": "sample",
+        "grid_price_eur_mwh": None,        # unknown off-box — NOT fabricated
+        "wasted_energy_available": False,  # conservative: assume nothing to soak
+        "joules_label": SAMPLE_LABEL,
+        "source": "SAMPLE snapshot (no live harvest feed reachable — doctrine v11)",
+        "measured_any": False,
+    }
+
+
+def _try_in_process_posture():
+    """Prefer the in-process harvest organ when it is importable (no network)."""
+    try:
+        from a11oy_harvest_endpoints import handle_posture  # shipped organ
+        p = handle_posture()
+        if isinstance(p, dict):
+            return p
+    except Exception:
+        return None
+    return None
+
+
+def _try_http_posture(timeout: float = 1.5):
+    """Attempt the live HTTP posture surfaces (local box, then public)."""
+    for url in _POSTURE_URLS:
+        try:
+            req = urllib.request.Request(url, headers={"User-Agent": "a11oy-anatomy-loop"})
+            with urllib.request.urlopen(req, timeout=timeout) as resp:  # nosec - read-only GET
+                body = resp.read().decode("utf-8", "replace")
+            data = json.loads(body)
+            if isinstance(data, dict):
+                data.setdefault("source", url)
+                return data
+        except Exception:
+            continue
+    return None
+
+
+def _read_intake() -> dict:
+    """INTAKE: live posture if reachable, else an honest SAMPLE snapshot.
+
+    Returns a normalized intake dict. joules_label is 'sample' by DEFAULT and
+    stays sample unless a REAL on-box power meter is present. Doctrine v11: a
+    live wasted-energy FEED reading (measured_any) is NOT a power measurement —
+    the harvest organ itself notes 'joules_label is always sample off-box; MEASURED
+    requires on-box NVML'. So we only flip to measured when the source explicitly
+    reports an on-box meter (metered_onbox), which never exists off-box. We NEVER
+    upgrade a sample into a measurement, and we NEVER invent numbers.
+    """
+    raw = _try_in_process_posture() or _try_http_posture()
+    if not isinstance(raw, dict):
+        return _sample_posture()
+
+    # A live FEED reading is informational only; it is NOT an on-box power meter.
+    feed_measured_any = bool(raw.get("measured_any", False))
+    # Doctrine v11: joules are 'sample' off-box. Only a real on-box meter
+    # (explicit metered_onbox flag) may yield 'measured' — absent off-box.
+    metered_onbox = bool(raw.get("metered_onbox", False))
+    joules_label = MEASURED_LABEL if metered_onbox else SAMPLE_LABEL
+    # grid price is only carried through if the live feed actually supplied one;
+    # absence -> None (never a fabricated figure).
+    grid_price = raw.get("grid_price_eur_mwh", None)
+    return {
+        "ok": bool(raw.get("ok", False)),
+        "posture": raw.get("posture", "unknown"),
+        "grid_price_eur_mwh": grid_price,
+        "wasted_energy_available": bool(raw.get("wasted_energy_available", False)),
+        "joules_label": joules_label,
+        "source": raw.get("source", "live harvest posture"),
+        "feed_measured_any": feed_measured_any,
+        "metered_onbox": metered_onbox,
+    }
+
+
+# ---------------------------------------------------------------------------
+# SAMAY (lungs/soak) -> KALLPA (metabolize to bounded work_credits).
+# ---------------------------------------------------------------------------
+def _samay_soak(intake: dict) -> dict:
+    """SAMAY: soak the available wasted-WORK window — a breath, not electrons.
+
+    The soaked window is a small, bounded number of bytes of WORK opportunity. We
+    only soak when the posture says wasted energy is available; otherwise the
+    window is zero (we never soak what is not there — that would be free energy).
+    """
+    available = bool(intake.get("wasted_energy_available", False))
+    window_bytes = 256 if available else 0   # a bounded SAMPLE soak window
+    return {"window_bytes": window_bytes, "soaked": available}
+
+
+def _kallpa_metabolize(soak: dict) -> dict:
+    """KALLPA: metabolize the soaked window into BOUNDED work_credits.
+
+    work_credits = bits of the soaked window, where bits are capped by the
+    Bekenstein software bound (window_bytes * 8) — a credit can never exceed what
+    the window can carry. The Landauer floor (kT ln 2 per bit) is the honest lower
+    bound on the work each bit represents. This BOUNDS the credit on both sides;
+    it never manufactures energy. joules figures stay SAMPLE.
+    """
+    window_bytes = int(soak.get("window_bytes", 0))
+    bekenstein_cap_bits = window_bytes * 8                 # ceiling — F19/TH6 bound
+    work_credits = bekenstein_cap_bits                     # bounded by the cap
+    landauer_floor_j = round(work_credits * LANDAUER_FLOOR_J, 30)  # SAMPLE lower bound
+    return {
+        "work_credits": work_credits,
+        "bekenstein_cap_bits": bekenstein_cap_bits,
+        "landauer_floor_joules": landauer_floor_j,
+        "joules_label": SAMPLE_LABEL,
+        "bound_note": "credit bounded above by Bekenstein bytes*8, floored by Landauer kT ln2/bit",
+    }
+
+
+# ---------------------------------------------------------------------------
+# heart/pulse beat — wrap szl_heart_blood (#332) when importable; else a local
+# byte-shaped beat so the loop + self-test run standalone with no network.
+# ---------------------------------------------------------------------------
+def _heart_beat(metab: dict) -> dict:
+    """Emit ONE DSSE beat carrying the cycle's receipt (work + receipt, not electrons)."""
+    payload = json.dumps(
+        {"work_credits": metab.get("work_credits", 0), "joules_label": SAMPLE_LABEL},
+        sort_keys=True, separators=(",", ":"),
+    )
+    try:
+        from szl_heart_blood import emit_beat  # shipped organ (#332)
+        beat = emit_beat(output=payload, energy_source="curtailed-sample", joules_est=0.0)
+        if isinstance(beat, dict) and beat.get("beat_id"):
+            return {"beat_id": beat.get("beat_id"), "beat_hash": beat.get("beat_hash", ""),
+                    "source": "szl_heart_blood (#332)"}
+    except Exception:
+        pass
+    # Local fallback beat — tamper-evident digest over the payload, NO real key.
+    import hashlib
+    h = hashlib.sha256(payload.encode("utf-8")).hexdigest()
+    return {"beat_id": f"beat-local-{h[:8]}", "beat_hash": h,
+            "source": "local fallback (szl_heart_blood not importable)"}
+
+
+# ---------------------------------------------------------------------------
+# YARQA — disperse the beat to the organs. EXPERIMENTAL tier (never proven).
+# ---------------------------------------------------------------------------
+def _yarqa_disperse(beat: dict, flowing: bool) -> list:
+    """YARQA: the irrigation canal. Disperse the beat to each organ.
+
+    Every organ is tagged EXPERIMENTAL — we make NO proven claim about any organ.
+    flowing reflects whether this cycle actually soaked + beat (i.e. there was
+    wasted work to circulate); when nothing was soaked, the canal is idle (honest).
+    """
+    organs = []
+    for name, role in _ORGAN_SPECS:
+        organs.append({
+            "name": name,
+            "role": role,
+            "flowing": bool(flowing),
+            "note": f"EXPERIMENTAL tier — carries the beat {beat.get('beat_id','')}, not electrons; "
+                    f"never claimed proven",
+        })
+    return organs
+
+
+# ---------------------------------------------------------------------------
+# EnergyReservoir — store the metabolized work_credits (a tank of WORK+receipt).
+# ---------------------------------------------------------------------------
+class EnergyReservoir:
+    """Holds metabolized WORK credits + their receipt. NOT a battery of electrons.
+
+    It is process-local and resets on restart. It only ever holds what was
+    actually metabolized this run — it can never report more than was soaked.
+    """
+
+    def __init__(self) -> None:
+        self._work_credits = 0
+
+    def store(self, work_credits: int) -> dict:
+        self._work_credits += max(0, int(work_credits))
+        return {
+            "work_credits": self._work_credits,
+            "joules_label": SAMPLE_LABEL,
+            "stored": True,
+            "note": "tank of soaked WORK + receipts (NOT stored electrons); SAMPLE figures",
+        }
+
+
+_RESERVOIR = EnergyReservoir()
+
+
+# ---------------------------------------------------------------------------
+# provenance receipt + validate — wrap szl_energy_provenance (#331) when present.
+# ---------------------------------------------------------------------------
+def _provenance_receipt(metab: dict) -> dict:
+    """Bind this cycle into the tamper-evident provenance chain; return receipt id."""
+    payload = json.dumps(
+        {"work_credits": metab.get("work_credits", 0), "joules_label": SAMPLE_LABEL},
+        sort_keys=True, separators=(",", ":"),
+    )
+    try:
+        from szl_energy_provenance import append_receipt  # shipped organ (#331)
+        entry = append_receipt(output=payload, energy_source="curtailed-sample", joules_est=0.0)
+        if isinstance(entry, dict) and entry.get("receipt_hash"):
+            return {"last_receipt_id": entry.get("receipt_hash"),
+                    "validated": True, "source": "szl_energy_provenance (#331)"}
+    except Exception:
+        pass
+    import hashlib
+    rid = hashlib.sha256(payload.encode("utf-8")).hexdigest()
+    return {"last_receipt_id": rid, "validated": True,
+            "source": "local fallback (szl_energy_provenance not importable)"}
+
+
+# ---------------------------------------------------------------------------
+# Ayni F11 — balance the books. intake == output + stored + proven; never net+.
+# ---------------------------------------------------------------------------
+def _ayni_balance(intake_credits: int, output_credits: int, stored_credits: int,
+                  proven_credits: int) -> dict:
+    """F11 reciprocity ledger: the loop must balance, never run net-positive.
+
+    The invariant is intake == output + stored + proven for THIS cycle's credits.
+    A cycle can only disperse/store/prove what it soaked. balanced is True exactly
+    when the books reconcile; net-positive is structurally impossible here because
+    output + stored + proven are derived from the same soaked intake.
+    """
+    settled = int(output_credits) + int(stored_credits) + int(proven_credits)
+    # Each soaked credit is dispersed (output), tanked (stored) and proven once;
+    # by construction they each equal intake, so the reciprocal balance holds when
+    # intake matches each leg. balanced means no leg exceeds intake (no net gain).
+    balanced = (int(output_credits) == int(intake_credits)
+                and int(stored_credits) == int(intake_credits)
+                and int(proven_credits) == int(intake_credits))
+    return {
+        "balanced": bool(balanced),
+        "intake": int(intake_credits),
+        "output": int(output_credits),
+        "stored": int(stored_credits),
+        "proven": int(proven_credits),
+        "settled": settled,
+        "note": "Ayni reciprocity (F11): intake == output == stored == proven per cycle; "
+                "reciprocal, never net-positive; Λ = Conjecture 1, not a theorem",
+    }
+
+
+# ---------------------------------------------------------------------------
+# The loop: run ONE circulation cycle and return its live state. Crash-proof.
+# ---------------------------------------------------------------------------
+def run_loop(ns: str = "a11oy") -> dict:
+    """Run one closed-loop circulation cycle and return its honest live state.
+
+    Wrapped end-to-end in try/except: any organ fault degrades to an honest,
+    doctrine-clean response — it can NEVER raise into the app.
+    """
+    try:
+        intake = _read_intake()
+        soak = _samay_soak(intake)
+        metab = _kallpa_metabolize(soak)
+        beat = _heart_beat(metab)
+        flowing = bool(soak.get("soaked", False))
+        organs = _yarqa_disperse(beat, flowing)
+
+        credits = int(metab.get("work_credits", 0))
+        reservoir = _RESERVOIR.store(credits)
+        prov = _provenance_receipt(metab)
+
+        # Per-cycle Ayni: each soaked credit is dispersed, stored and proven once.
+        cycle_stored = credits
+        ayni = _ayni_balance(
+            intake_credits=credits,
+            output_credits=credits,     # dispersed via YARQA
+            stored_credits=cycle_stored,
+            proven_credits=credits,     # proven via the receipt
+        )
+
+        # joules_label is the loop-level honesty: measured ONLY if intake measured.
+        joules_label = MEASURED_LABEL if intake.get("joules_label") == MEASURED_LABEL else SAMPLE_LABEL
+
+        beats_last_cycle = 1 if flowing or beat.get("beat_id") else 0
+
+        return {
+            "ok": True,
+            "kind": "anatomy-circulation-loop",
+            "ns": ns,
+            "doctrine": DOCTRINE,
+            "intake": {
+                "grid_price_eur_mwh": intake.get("grid_price_eur_mwh"),
+                "posture": intake.get("posture"),
+                "wasted_energy_available": bool(intake.get("wasted_energy_available", False)),
+                "joules_label": intake.get("joules_label", SAMPLE_LABEL),
+            },
+            "organs": organs,
+            "beats_last_cycle": int(beats_last_cycle),
+            "reservoir": {
+                "work_credits": reservoir.get("work_credits", 0),
+                "joules_label": SAMPLE_LABEL,
+                "stored": bool(reservoir.get("stored", False)),
+            },
+            "last_receipt_id": prov.get("last_receipt_id", ""),
+            "ayni": {
+                "balanced": bool(ayni.get("balanced", False)),
+                "intake": ayni.get("intake", 0),
+                "output": ayni.get("output", 0),
+                "stored": ayni.get("stored", 0),
+                "note": ayni.get("note", ""),
+            },
+            "joules_label": joules_label,
+            "honesty": (
+                "carries soaked WORK + receipts, NOT electrons; joules are SAMPLE off-box "
+                "(no power meter wired); organs are EXPERIMENTAL (never proven); Ayni balances "
+                "(reciprocal, never net-positive); no free-energy claim; Λ = Conjecture 1; "
+                "sovereign stays false unless on own metal; degrades to a labeled SAMPLE "
+                "snapshot when no live feed is reachable — never fabricated."
+            ),
+            "stages": {
+                "intake_source": intake.get("source", ""),
+                "samay_soaked": bool(soak.get("soaked", False)),
+                "kallpa_bekenstein_cap_bits": metab.get("bekenstein_cap_bits", 0),
+                "kallpa_landauer_floor_joules": metab.get("landauer_floor_joules", 0.0),
+                "heart_beat": beat,
+                "provenance_source": prov.get("source", ""),
+                "validated": bool(prov.get("validated", False)),
+            },
+            "computed_at": _now(),
+        }
+    except Exception as exc:  # never raise into the app — honest degrade
+        return {
+            "ok": False,
+            "kind": "anatomy-circulation-loop",
+            "ns": ns,
+            "doctrine": DOCTRINE,
+            "intake": {"grid_price_eur_mwh": None, "posture": "sample",
+                       "wasted_energy_available": False, "joules_label": SAMPLE_LABEL},
+            "organs": [
+                {"name": n, "role": r, "flowing": False,
+                 "note": "EXPERIMENTAL tier — never claimed proven"}
+                for n, r in _ORGAN_SPECS
+            ],
+            "beats_last_cycle": 0,
+            "reservoir": {"work_credits": 0, "joules_label": SAMPLE_LABEL, "stored": False},
+            "last_receipt_id": "",
+            "ayni": {"balanced": True, "intake": 0, "output": 0, "stored": 0,
+                     "note": "empty cycle balances trivially: 0 == 0 + 0 + 0"},
+            "joules_label": SAMPLE_LABEL,
+            "honesty": f"degraded honestly ({type(exc).__name__}); SAMPLE only; no fabricated numbers",
+            "computed_at": _now(),
+        }
+
+
+# ---------------------------------------------------------------------------
+# HTTP handler + registration (matches szl_energy_provenance / szl_heart_blood).
+# ---------------------------------------------------------------------------
+def _h_loop(req):
+    from starlette.responses import JSONResponse
+    return JSONResponse(run_loop())
+
+
+def register(app, ns="a11oy"):
+    """Wire the loop read endpoint onto the app under /api/<ns>/v1/anatomy/loop.
+
+    Additive. Uses FastAPI's add_api_route when available (so it resolves before
+    the SPA catch-all, matching the other szl_* modules); falls back to a Starlette
+    route append for a bare Starlette app. The handler closes over `ns`.
+    """
+    base = f"/api/{ns}/v1/anatomy"
+
+    def _loop_handler(req=None):
+        from starlette.responses import JSONResponse
+        return JSONResponse(run_loop(ns=ns))
+
+    handlers = [
+        (f"{base}/loop", _loop_handler),
+    ]
+    add_api_route = getattr(app, "add_api_route", None)
+    for path, fn in handlers:
+        if callable(add_api_route):
+            app.add_api_route(path, fn, methods=["GET"])
+        else:
+            from starlette.routing import Route
+            app.router.routes.append(Route(path, fn))
+    return [p for p, _ in handlers]
+
+
+def _selftest() -> dict:
+    """No-server self-test: run the loop offline and assert the doctrine invariants."""
+    out = run_loop()
+    assert out["joules_label"] == SAMPLE_LABEL, out
+    assert out["ayni"]["balanced"] is True, out
+    assert "kind" in out and out["kind"] == "anatomy-circulation-loop"
+    for organ in out["organs"]:
+        assert "experimental" in organ["note"].lower(), organ
+    return {"ok": True, "joules_label": out["joules_label"],
+            "ayni_balanced": out["ayni"]["balanced"], "organs": len(out["organs"])}
+
+
+if __name__ == "__main__":
+    print(json.dumps(_selftest(), indent=2))