arborist/substrate/prometheus: Phase 1 controller (#000037)

Pure-function recursive-falsification controller implementing
ticket #000037 §13 algorithm (steps 1-9 + 12), §5 Shannon-entropy
fork-selection with stable softmax, §6 8-class hard-veto order,
§7 Kelly-bounded allocation with 4 safety guards, §7.1 EMA-smoothed
difficulty update, §14 exception-matrix dispatch, and §15 three
named weight profiles (safe/conservative/exploratory).

No DB, no LLM, no scheduler — advisory pure function over already-
committed state. Phase 2 (controller_events sibling table) lands
in companion commit a786d6d. Phase 3 (sleep sweep scheduler) is
not in this commit.

Module: arborist/substrate/prometheus.py (782 lines)
Tests: tests/test_prometheus.py (22 passing tests — the 17 named
contracts from §16.2 plus 5 boundary cases for the §6 veto-class
priority dispatch and §7.1 EMA stability).

Also removes obsolete tests/test_prometheus_sigma.py — pre-Phase-1
scaffolding placeholder whose 17 tests all called pytest.fail()
with "Phase 1 implementation pending" and the @skip_until_phase_1
decorator never auto-flipped to pass-on-import. The contract is
now in tests/test_prometheus.py.

arborist/substrate/prometheus.py

@@ -0,0 +1,782 @@
+"""Prometheus-Σ recursive falsification controller (ticket #000037, Phase 1).
+
+Pure-function advisory controller over already-committed substrate state.
+Implements §5 (Shannon-entropy fork-selection with numerically stable
+softmax), §6 (eight-class hard-veto order), §7 (Kelly-bounded
+allocation with four safety guards), §7.1 (EMA-smoothed difficulty
+update law), §13 algorithm steps 1-9 + 12, §14 exception-handling
+matrix, and §15 three named weight profiles (safe / conservative /
+exploratory).
+
+Phase scope (locked):
+
+- **Phase 1 (this module):** pure function. No DB writes, no LLM
+  calls, no scheduler, no background jobs. Controller returns a
+  :class:`ControllerDecision` (advisory) plus optional
+  :class:`MemoryRootUpdateProposal` / :class:`SelfModelUpdateProposal`
+  records — proposals, never mutations. The hard rule from §4.4
+  stands: controller never mutates MemoryRoot or SelfModel directly.
+- **Phase 2 (separate ticket scope):** sibling ``controller_events``
+  audit-row writes. This module builds the ``advisory_events`` tuple
+  the writer will consume; it does not persist anything.
+- **Phase 3 (separate ticket scope):** unconscious sweep scheduler
+  that walks ``providence_cache`` and ``documents`` during ingest
+  lulls. Out of scope here.
+
+Doctrinal commitments (preserved verbatim from §21 of the ticket):
+
+- The controller does not defeat Gödel.
+- LLM is witness, never authority.
+
+This module imports nothing from :mod:`arborist.qa.client` (LLM
+gateway) or :mod:`arborist.qa.keys` (cache_key path) — verified by
+:func:`test_no_llm_call_in_controller` and
+:func:`test_controller_does_not_modify_cache_key_inputs`.
+
+§8 note — cross-chain matrix omission. The §13 algorithm references
+the §8 ``A[i, j]`` cross-chain matrix; the spec is explicit that
+Phase 1 uses only the scalar utility ``U_i`` (§5) and softmax. The
+full ``A[i, j]`` matrix is **deferred to Phase 2+ if expressivity is
+insufficient** — the scalar form already passes the §16.2 test list
+and the matrix adds complexity that the ticket explicitly gates on
+measured pressure.
+"""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass, field, replace
+from typing import Literal, Sequence
+
+
+# ---------------------------------------------------------------------
+# Veto classification (§6)
+# ---------------------------------------------------------------------
+
+#: §6 veto-tag → output-label class. ``hard_vetoes`` on a
+#: ControllerBranch is the input signal — caller pre-classifies. The
+#: controller routes by tag string to the right output label class.
+VETO_CLASS: dict[str, Literal["QUARANTINE", "REJECT", "ESCALATE"]] = {
+    # §6.1, §6.4, §6.5, §6.7 — QUARANTINE class
+    "unsupported_carrier": "QUARANTINE",
+    "cache_drift": "QUARANTINE",
+    "soft_hash_signal": "QUARANTINE",
+    "schema_mismatch": "QUARANTINE",
+    # §6.2, §6.3, §6.8 — REJECT class
+    "verifier_failure": "REJECT",
+    "hard_regression": "REJECT",
+    "source_warrant_downgrade": "REJECT",
+    # §6.6 + soft-hash replay-window — ESCALATE class
+    "memory_invalidation_exceeded": "ESCALATE",
+    "replay_window_unbounded": "ESCALATE",
+}
+
+#: Severity ordering when ALL branches are vetoed and the controller
+#: must pick the most severe class present. Fail-loud — ESCALATE >
+#: QUARANTINE > REJECT.
+_SEVERITY_RANK: dict[str, int] = {
+    "ESCALATE": 3,
+    "QUARANTINE": 2,
+    "REJECT": 1,
+}
+
+#: Entropy gating thresholds (§5). ``H_norm <= H_LOW`` → narrow,
+#: confident; ``H_norm >= H_HIGH`` → widen, uncertain.
+H_LOW = 0.3
+H_HIGH = 0.7
+
+#: §9 memory-invalidation governance threshold (κ). A selected branch
+#: with ``memory_invalidation >= KAPPA_MEMORY`` escalates rather than
+#: silently committing across a large memory frontier. Default per the
+#: ticket text where ``κ`` appears in §6 / §8 / §14 without a numeric
+#: value — fox can tighten per profile later.
+KAPPA_MEMORY = 0.5
+
+#: §12 Trigger 2 minimum sample count.
+N_MIN_TRIGGER_2 = 30
+
+#: §12 Trigger 2 coefficient-of-variation threshold (also matches the
+#: docstring text "CoV ≥ 0.5").
+COV_THRESHOLD_TRIGGER_2 = 0.5
+
+#: §12 Trigger 2 absolute stddev threshold for the zero-mean fallback
+#: path. Picked to match the spec text "≥ 0.05" (smaller than the
+#: 0.10 absolute-stddev cutoff for the normal CoV variant — the
+#: zero-mean path is the "any signal at all" path).
+ABS_STDDEV_ZERO_MEAN = 0.05
+
+
+# ---------------------------------------------------------------------
+# Locked dataclass contract (§16.1)
+# ---------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class BatteryDeltas:
+    """Per-branch Δ-rate against the 5S/5T/5F/5R batteries (§5)."""
+
+    delta_5s: float
+    delta_5t: float
+    delta_5f: float
+    delta_5r: float
+
+
+@dataclass(frozen=True)
+class ControllerBranch:
+    """One candidate branch fed into the controller.
+
+    ``hard_vetoes`` carries pre-classified veto tags from upstream
+    (kernel, verifier, soft-hash analyzer, …). The controller does
+    not re-classify; it routes per :data:`VETO_CLASS`.
+
+    ``payoff_b`` is the Kelly edge-payoff (§7). Defaults to 1.0
+    (even bet) so callers that haven't measured payoff don't need
+    to set it. Tests exercise the ``b_i <= 0`` safety guard via
+    this field.
+    """
+
+    branch_id: str
+    deltas: BatteryDeltas
+    witness_divergence: float
+    capital_cost: float
+    regression_penalty: float
+    security_risk: float
+    memory_invalidation: float
+    selfmodel_calibration_gain: float = 0.0
+    warrant_promotion_gain: float = 0.0
+    hard_vetoes: tuple[str, ...] = ()
+    payoff_b: float = 1.0
+
+
+@dataclass(frozen=True)
+class ControllerWeights:
+    """§15.1 safe-default weight profile.
+
+    Other profiles via :func:`conservative_weights` /
+    :func:`exploratory_weights`. The profile name does NOT enter
+    ``governance_policy_hash`` in Phase 1 (§15 explicit).
+    """
+
+    # §5 utility coefficients
+    alpha_5s: float = 1.0
+    beta_5t: float = 1.0
+    gamma_5f: float = 1.25
+    rho_5r: float = 1.0
+    sigma_selfmodel_calibration: float = 0.75
+    tau_warrant_promotion: float = 0.75
+    lambda_capital_cost: float = 1.0
+    mu_regression_penalty: float = 2.0
+    nu_witness_divergence: float = 1.5
+    xi_security_risk: float = 3.0
+    omega_memory_invalidation: float = 2.0
+
+    # §5 softmax temperature + §7.1 EMA smoothing
+    eta_softmax_temperature: float = 1.0
+    ema_smoothing_r: float = 0.2
+
+    # §7.1 difficulty-update sub-weights
+    a_entropy_drive: float = 0.1
+    b_divergence_drive: float = 0.1
+    c_hard_regression_drive: float = 0.3
+    d_witness_agreement_pull: float = 0.1
+    e_capital_pressure_pull: float = 0.05
+    min_difficulty: float = 0.0
+    max_difficulty: float = 10.0
+
+
+@dataclass(frozen=True)
+class ControllerInput:
+    """Full input bundle for one controller invocation.
+
+    ``budget`` is the binding Hermes-concurrency constraint (~4 in
+    today's deployment, per §11). ``hermes_utilization`` is the
+    current in-flight count — Phase 2 cooperative-yield logic will
+    consult it. Phase 1 honors only ``budget == 0`` (§14 row).
+    """
+
+    organism_root: str
+    branches: tuple[ControllerBranch, ...]
+    budget: int
+    hermes_utilization: int
+    weights: ControllerWeights
+    difficulty: float
+    divergence_rate: float = 0.0
+    hard_regression_rate: float = 0.0
+    witness_agreement_rate: float = 1.0
+    capital_pressure: float = 0.0
+
+
+ControllerLabel = Literal[
+    "ACCEPT",
+    "MARGINAL",
+    "REJECT",
+    "QUARANTINE",
+    "UNKNOWN",
+    "ESCALATE",
+    "DEFERRED",
+]
+
+
+@dataclass(frozen=True)
+class MemoryRootUpdateProposal:
+    """§4.4 — proposal record for MemoryRoot. Phase 2 commits."""
+
+    organism_root: str
+    branch_id: str
+    proposed_invalidation_tags: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class SelfModelUpdateProposal:
+    """§4.4 — proposal record for SelfModel. Phase 2 commits."""
+
+    organism_root: str
+    branch_id: str
+    calibration_delta: float
+    reason: str
+
+
+@dataclass(frozen=True)
+class ControllerDecision:
+    """§13 step-9 output. Advisory; Phase 2 persists ``advisory_events``."""
+
+    selected_branch_id: str | None
+    label: ControllerLabel
+    allocations: dict[str, float]
+    difficulty_next: float
+    veto_reasons: dict[str, tuple[str, ...]]
+    entropy: float
+    notes: tuple[str, ...]
+    memory_proposals: tuple[MemoryRootUpdateProposal, ...] = ()
+    selfmodel_proposals: tuple[SelfModelUpdateProposal, ...] = ()
+    advisory_events: tuple[tuple[str, dict], ...] = ()
+
+
+# ---------------------------------------------------------------------
+# §15 weight-profile factories
+# ---------------------------------------------------------------------
+
+
+def safe_weights() -> ControllerWeights:
+    """§15.1 safe-default profile — Phase 1 baseline."""
+    return ControllerWeights()
+
+
+def conservative_weights() -> ControllerWeights:
+    """§15.2 conservative profile — real-shard production."""
+    return replace(
+        ControllerWeights(),
+        gamma_5f=1.0,
+        lambda_capital_cost=1.5,
+        xi_security_risk=4.0,
+        eta_softmax_temperature=0.75,
+    )
+
+
+def exploratory_weights() -> ControllerWeights:
+    """§15.3 exploratory profile — research sweeps."""
+    return replace(
+        ControllerWeights(),
+        gamma_5f=1.5,
+        nu_witness_divergence=0.75,
+        lambda_capital_cost=0.5,
+        eta_softmax_temperature=1.5,
+    )
+
+
+WEIGHT_PROFILES: dict[str, ControllerWeights] = {
+    "safe": safe_weights(),
+    "conservative": conservative_weights(),
+    "exploratory": exploratory_weights(),
+}
+
+
+# ---------------------------------------------------------------------
+# §12 Trigger 2 — divergence-variance phase-gate check
+# ---------------------------------------------------------------------
+
+
+def phase_1_trigger_check(
+    divergence_samples: Sequence[float],
+) -> tuple[bool, str]:
+    """§12 Trigger 2 — divergence-variance check with §14 guards.
+
+    Returns ``(triggered, reason)``. Edge cases per §14:
+
+    - sample_count < :data:`N_MIN_TRIGGER_2` → ``(False, "small_sample")``
+    - mean == 0:
+      - stddev == 0 → ``(False, "no_variance_signal")``
+      - stddev > 0 → ``(stddev >= ABS_STDDEV_ZERO_MEAN, "absolute_stddev_path")``
+    - otherwise CoV = stddev / max(mean, 1e-6):
+      - ``CoV >= COV_THRESHOLD_TRIGGER_2`` → ``(True, "variance_threshold")``
+      - else → ``(False, "variance_below_threshold")``
+    """
+    n = len(divergence_samples)
+    if n < N_MIN_TRIGGER_2:
+        return (False, "small_sample")
+
+    mean = sum(divergence_samples) / n
+    # Population stddev — the controller is making a decision over the
+    # observed window, not estimating a parent population parameter.
+    var = sum((x - mean) ** 2 for x in divergence_samples) / n
+    stddev = math.sqrt(var)
+
+    if mean == 0.0:
+        if stddev == 0.0:
+            return (False, "no_variance_signal")
+        return (stddev >= ABS_STDDEV_ZERO_MEAN, "absolute_stddev_path")
+
+    cov = stddev / max(mean, 1e-6)
+    if cov >= COV_THRESHOLD_TRIGGER_2:
+        return (True, "variance_threshold")
+    return (False, "variance_below_threshold")
+
+
+# ---------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------
+
+
+def _utility(branch: ControllerBranch, w: ControllerWeights) -> float:
+    """§5 scalar utility ``U_i`` — all 11 weighted terms."""
+    d = branch.deltas
+    return (
+        w.alpha_5s * d.delta_5s
+        + w.beta_5t * d.delta_5t
+        + w.gamma_5f * d.delta_5f
+        + w.rho_5r * d.delta_5r
+        + w.sigma_selfmodel_calibration * branch.selfmodel_calibration_gain
+        + w.tau_warrant_promotion * branch.warrant_promotion_gain
+        - w.lambda_capital_cost * branch.capital_cost
+        - w.mu_regression_penalty * branch.regression_penalty
+        - w.nu_witness_divergence * branch.witness_divergence
+        - w.xi_security_risk * branch.security_risk
+        - w.omega_memory_invalidation * branch.memory_invalidation
+    )
+
+
+def _stable_softmax(zs: list[float]) -> list[float]:
+    """§5 numerically stable softmax: ``exp(z - max z) / Σ exp(z - max z)``.
+
+    Empty input yields empty output. All-equal input yields uniform.
+    Required — naive ``exp(η · U_i)`` overflows around ``η · U > ~700``.
+    """
+    if not zs:
+        return []
+    m = max(zs)
+    exps = [math.exp(z - m) for z in zs]
+    total = sum(exps)
+    if total == 0.0:  # defensive — shouldn't happen post-shift
+        n = len(zs)
+        return [1.0 / n] * n
+    return [e / total for e in exps]
+
+
+def _entropy_norm(ps: list[float]) -> float:
+    """§5 normalized Shannon entropy ``H(p) / log(n)``.
+
+    Edge case: ``n == 1`` → ``log(1) = 0`` would divide by zero;
+    return ``0.0`` (the distribution is a point mass, zero entropy
+    by construction).
+    """
+    n = len(ps)
+    if n <= 1:
+        return 0.0
+    h = 0.0
+    for p in ps:
+        if p > 0.0:
+            h -= p * math.log(p)
+    return h / math.log(n)
+
+
+def _all_vetoed_label(
+    branches: Sequence[ControllerBranch],
+) -> tuple[ControllerLabel, str]:
+    """§3 / §6 — pick most severe veto class across all branches.
+
+    Returns ``(label, severity_class)``. Fail-loud: ESCALATE >
+    QUARANTINE > REJECT.
+    """
+    worst_rank = 0
+    worst_class: str = "REJECT"
+    for b in branches:
+        for tag in b.hard_vetoes:
+            cls = VETO_CLASS.get(tag, "REJECT")
+            r = _SEVERITY_RANK.get(cls, 0)
+            if r > worst_rank:
+                worst_rank = r
+                worst_class = cls
+    # worst_class is already a ControllerLabel (REJECT / QUARANTINE /
+    # ESCALATE), all three valid output labels.
+    return (worst_class, worst_class)  # type: ignore[return-value]
+
+
+def _clamp(x: float, lo: float, hi: float) -> float:
+    return max(lo, min(hi, x))
+
+
+def _difficulty_next(
+    inp: ControllerInput,
+    h_norm: float,
+) -> float:
+    """§7.1 EMA-smoothed difficulty update law."""
+    w = inp.weights
+    raw = _clamp(
+        inp.difficulty
+        + w.a_entropy_drive * h_norm
+        + w.b_divergence_drive * inp.divergence_rate
+        + w.c_hard_regression_drive * inp.hard_regression_rate
+        - w.d_witness_agreement_pull * inp.witness_agreement_rate
+        - w.e_capital_pressure_pull * inp.capital_pressure,
+        w.min_difficulty,
+        w.max_difficulty,
+    )
+    r = w.ema_smoothing_r
+    return (1.0 - r) * inp.difficulty + r * raw
+
+
+# ---------------------------------------------------------------------
+# Algorithm entry point (§13)
+# ---------------------------------------------------------------------
+
+
+def controller_decide(inp: ControllerInput) -> ControllerDecision:
+    """Run the §13 algorithm (steps 1-9 + 12) on the input bundle.
+
+    Pure function: no DB, no LLM, no scheduler. Same inputs →
+    byte-identical output. Idempotent.
+
+    Steps:
+
+    1. Accept pre-computed ``hard_vetoes`` per branch.
+    2. (Phase 1 scalar utility only — full ``A[i, j]`` matrix per
+       §8 deferred to Phase 2+ if expressivity insufficient.)
+    3. Apply §6 hard-veto order; ALL-vetoed cascades to the most
+       severe class present (ESCALATE > QUARANTINE > REJECT).
+    4. Compute ``U_i`` (§5) for unvetoed branches.
+    5. Stable softmax ``p_i = exp(z_i - max z) / Σ exp(z_j - max z)``.
+    6. ``H_norm = H(p) / log(n)`` (0.0 when n=1).
+    7. EMA-smoothed difficulty update (§7.1).
+    8. Kelly-bounded allocation with four guards (§7): vetoed,
+       ``payoff_b <= 0``, ``p_i <= 0``, ``Σ_raw == 0``.
+    9. Label dispatch per §14 exception matrix.
+    12. Emit MemoryRoot / SelfModel update **proposals** — never
+        mutations. The existing write paths gate every commit on
+        their own validation; the controller is advisory only.
+    """
+    branches = inp.branches
+    w = inp.weights
+    veto_reasons: dict[str, tuple[str, ...]] = {
+        b.branch_id: tuple(b.hard_vetoes) for b in branches if b.hard_vetoes
+    }
+    notes: list[str] = []
+
+    # ---- §14: no candidate branches → UNKNOWN ----
+    if not branches:
+        return ControllerDecision(
+            selected_branch_id=None,
+            label="UNKNOWN",
+            allocations={},
+            difficulty_next=inp.difficulty,
+            veto_reasons={},
+            entropy=0.0,
+            notes=("NO_CANDIDATE_BRANCHES",),
+        )
+
+    # ---- Step 1 + Step 3: hard-veto sweep ----
+    unvetoed = [b for b in branches if not b.hard_vetoes]
+    if not unvetoed:
+        label, _ = _all_vetoed_label(branches)
+        # §14: ALL_BRANCHES_VETOED → no allocation, no proposals.
+        # (Memory-proposal emission below filters on vetoed-AND-positive-
+        # regression-penalty, so cascade still passes through it.)
+        memory_proposals = _emit_memory_proposals(inp.organism_root, branches)
+        # Difficulty still updates — entropy is 0 (no live distribution).
+        difficulty_next = _difficulty_next(inp, h_norm=0.0)
+        advisory_events = (
+            (
+                "controller_decision",
+                {
+                    "organism_root": inp.organism_root,
+                    "label": label,
+                    "reason": "ALL_BRANCHES_VETOED",
+                    "selected_branch_id": None,
+                    "veto_count": len(branches),
+                },
+            ),
+        )
+        return ControllerDecision(
+            selected_branch_id=None,
+            label=label,
+            allocations={b.branch_id: 0.0 for b in branches},
+            difficulty_next=difficulty_next,
+            veto_reasons=veto_reasons,
+            entropy=0.0,
+            notes=("ALL_BRANCHES_VETOED",),
+            memory_proposals=memory_proposals,
+            advisory_events=advisory_events,
+        )
+
+    # ---- Step 4: utilities ----
+    utilities = {b.branch_id: _utility(b, w) for b in unvetoed}
+
+    # ---- Step 5: stable softmax ----
+    eta = w.eta_softmax_temperature
+    zs = [eta * utilities[b.branch_id] for b in unvetoed]
+    ps_unvetoed = _stable_softmax(zs)
+    probs: dict[str, float] = {
+        b.branch_id: p for b, p in zip(unvetoed, ps_unvetoed)
+    }
+
+    # ---- Step 6: normalized entropy ----
+    h_norm = _entropy_norm(ps_unvetoed)
+
+    # ---- Step 7: difficulty update ----
+    difficulty_next = _difficulty_next(inp, h_norm)
+
+    # ---- Step 8: Kelly-bounded allocation under B ----
+    allocations: dict[str, float] = {b.branch_id: 0.0 for b in branches}
+
+    # §14 row: Budget B = 0 → DEFERRED for everything.
+    if inp.budget == 0:
+        advisory_events = (
+            (
+                "controller_decision",
+                {
+                    "organism_root": inp.organism_root,
+                    "label": "DEFERRED",
+                    "reason": "ZERO_BUDGET",
+                    "entropy": h_norm,
+                },
+            ),
+        )
+        return ControllerDecision(
+            selected_branch_id=None,
+            label="DEFERRED",
+            allocations=allocations,
+            difficulty_next=difficulty_next,
+            veto_reasons=veto_reasons,
+            entropy=h_norm,
+            notes=("ZERO_BUDGET",),
+            advisory_events=advisory_events,
+        )
+
+    raw: dict[str, float] = {}
+    for b in unvetoed:
+        p_i = probs[b.branch_id]
+        b_i = b.payoff_b
+        if b_i <= 0.0:
+            f_i = 0.0
+        elif p_i <= 0.0:
+            f_i = 0.0
+        else:
+            q_i = 1.0 - p_i
+            f_i = max(0.0, (p_i * b_i - q_i) / b_i)
+        # raw_i = f_i · exp(η · U_i) using the SAME η. Re-stabilize
+        # using the same max-shift so the multiplication doesn't
+        # overflow even if U_i is large.
+        z_i = eta * utilities[b.branch_id]
+        # exp(z_i - max(zs)) is in [0, 1] post-shift — bounded.
+        if zs:
+            max_z = max(zs)
+            scaled_exp = math.exp(z_i - max_z)
+        else:
+            scaled_exp = 1.0
+        raw[b.branch_id] = f_i * scaled_exp
+
+    sum_raw = sum(raw.values())
+
+    # §14 + algorithm step 8: Σ_raw == 0 → DEFERRED.
+    if sum_raw == 0.0:
+        advisory_events = (
+            (
+                "controller_decision",
+                {
+                    "organism_root": inp.organism_root,
+                    "label": "DEFERRED",
+                    "reason": "ZERO_RAW_ALLOCATION",
+                    "entropy": h_norm,
+                },
+            ),
+        )
+        return ControllerDecision(
+            selected_branch_id=None,
+            label="DEFERRED",
+            allocations=allocations,
+            difficulty_next=difficulty_next,
+            veto_reasons=veto_reasons,
+            entropy=h_norm,
+            notes=("ZERO_RAW_ALLOCATION",),
+            advisory_events=advisory_events,
+        )
+
+    # Defensive: NaN / inf check on sum_raw (should never trigger
+    # given the stable-softmax max-shift, but the spec calls for the
+    # UNKNOWN exit if the math comes out invalid).
+    if math.isnan(sum_raw) or math.isinf(sum_raw):
+        advisory_events = (
+            (
+                "controller_decision",
+                {
+                    "organism_root": inp.organism_root,
+                    "label": "UNKNOWN",
+                    "reason": "NUMERIC_INSTABILITY",
+                    "entropy": h_norm,
+                },
+            ),
+        )
+        return ControllerDecision(
+            selected_branch_id=None,
+            label="UNKNOWN",
+            allocations=allocations,
+            difficulty_next=difficulty_next,
+            veto_reasons=veto_reasons,
+            entropy=h_norm,
+            notes=("NUMERIC_INSTABILITY",),
+            advisory_events=advisory_events,
+        )
+
+    budget_f = float(inp.budget)
+    for bid, r_i in raw.items():
+        allocations[bid] = budget_f * r_i / sum_raw
+
+    # ---- Step 9: label dispatch (§14 exception matrix) ----
+    # Pick the branch with highest allocation among unvetoed.
+    selected = max(
+        unvetoed, key=lambda b: (allocations[b.branch_id], b.branch_id)
+    )
+    selected_u = utilities[selected.branch_id]
+
+    label: ControllerLabel
+    reason_notes: list[str] = []
+
+    if selected.payoff_b <= 0.0:
+        label = "REJECT"
+        reason_notes.append("NEGATIVE_PAYOFF")
+    elif selected_u <= 0.0:
+        label = "REJECT"
+        reason_notes.append("NON_POSITIVE_UTILITY")
+    elif selected.memory_invalidation >= KAPPA_MEMORY:
+        label = "ESCALATE"
+        reason_notes.append("MEMORY_INVALIDATION_ABOVE_KAPPA")
+    elif h_norm <= H_LOW:
+        label = "ACCEPT"
+        reason_notes.append("LOW_ENTROPY")
+    elif h_norm >= H_HIGH:
+        label = "MARGINAL"
+        reason_notes.append("HIGH_ENTROPY")
+    else:
+        label = "ACCEPT"
+        reason_notes.append("MID_ENTROPY")
+
+    notes.extend(reason_notes)
+
+    # ---- Step 12: emit proposals (NOT mutations) ----
+    memory_proposals = _emit_memory_proposals(inp.organism_root, branches)
+    selfmodel_proposals: tuple[SelfModelUpdateProposal, ...] = ()
+    if selected.selfmodel_calibration_gain != 0.0:
+        selfmodel_proposals = (
+            SelfModelUpdateProposal(
+                organism_root=inp.organism_root,
+                branch_id=selected.branch_id,
+                calibration_delta=selected.selfmodel_calibration_gain,
+                reason="controller_phase_1_calibration",
+            ),
+        )
+
+    # ---- Advisory event (§13 step 10; persisted by Phase 2) ----
+    advisory_events: tuple[tuple[str, dict], ...] = (
+        (
+            "controller_decision",
+            {
+                "organism_root": inp.organism_root,
+                "label": label,
+                "selected_branch_id": selected.branch_id,
+                "entropy": h_norm,
+                "utility": selected_u,
+                "allocation": allocations[selected.branch_id],
+                "reason": reason_notes[0] if reason_notes else "",
+            },
+        ),
+        (
+            "controller_difficulty",
+            {
+                "organism_root": inp.organism_root,
+                "difficulty_prev": inp.difficulty,
+                "difficulty_next": difficulty_next,
+                "h_norm": h_norm,
+                "divergence_rate": inp.divergence_rate,
+            },
+        ),
+        (
+            "controller_budget_allocation",
+            {
+                "organism_root": inp.organism_root,
+                "budget": inp.budget,
+                "allocations": dict(allocations),
+            },
+        ),
+    )
+
+    return ControllerDecision(
+        selected_branch_id=selected.branch_id,
+        label=label,
+        allocations=allocations,
+        difficulty_next=difficulty_next,
+        veto_reasons=veto_reasons,
+        entropy=h_norm,
+        notes=tuple(notes),
+        memory_proposals=memory_proposals,
+        selfmodel_proposals=selfmodel_proposals,
+        advisory_events=advisory_events,
+    )
+
+
+def _emit_memory_proposals(
+    organism_root: str,
+    branches: Sequence[ControllerBranch],
+) -> tuple[MemoryRootUpdateProposal, ...]:
+    """Step 12 — propose ``BRANCH_REGRESSED`` invalidation tags.
+
+    Triggers when a branch has ``regression_penalty > 0`` AND it was
+    vetoed (i.e. carries any hard veto). Pure proposal — no mutation.
+    """
+    proposals: list[MemoryRootUpdateProposal] = []
+    for b in branches:
+        if b.regression_penalty > 0.0 and b.hard_vetoes:
+            proposals.append(
+                MemoryRootUpdateProposal(
+                    organism_root=organism_root,
+                    branch_id=b.branch_id,
+                    proposed_invalidation_tags=("BRANCH_REGRESSED",),
+                )
+            )
+    return tuple(proposals)
+
+
+__all__ = [
+    # Constants
+    "VETO_CLASS",
+    "H_LOW",
+    "H_HIGH",
+    "KAPPA_MEMORY",
+    "N_MIN_TRIGGER_2",
+    "COV_THRESHOLD_TRIGGER_2",
+    "ABS_STDDEV_ZERO_MEAN",
+    "WEIGHT_PROFILES",
+    # Dataclasses
+    "BatteryDeltas",
+    "ControllerBranch",
+    "ControllerWeights",
+    "ControllerInput",
+    "ControllerDecision",
+    "MemoryRootUpdateProposal",
+    "SelfModelUpdateProposal",
+    # Weight profile factories
+    "safe_weights",
+    "conservative_weights",
+    "exploratory_weights",
+    # Entry points
+    "controller_decide",
+    "phase_1_trigger_check",
+]

tests/test_prometheus.py

@@ -0,0 +1,608 @@
+"""Phase 1 tests for arborist.substrate.prometheus (ticket #000037).
+
+Covers the §16.2 test list — 17 named cases — plus a small set of
+sanity checks (determinism, weight-profile shape, trigger-check
+positive path) that don't appear in the §16.2 list but harden the
+"pure function" contract.
+
+Discipline (§16.1):
+
+- Pure-function — no DB, no LLM, no scheduler.
+- Banned imports verified via AST inspection of the module source.
+- Proposals only; no mutation API calls.
+"""
+
+from __future__ import annotations
+
+import ast
+import math
+import random
+from pathlib import Path
+
+import pytest
+
+from arborist.substrate.prometheus import (
+    ABS_STDDEV_ZERO_MEAN,
+    H_HIGH,
+    H_LOW,
+    KAPPA_MEMORY,
+    N_MIN_TRIGGER_2,
+    WEIGHT_PROFILES,
+    BatteryDeltas,
+    ControllerBranch,
+    ControllerDecision,
+    ControllerInput,
+    ControllerWeights,
+    MemoryRootUpdateProposal,
+    SelfModelUpdateProposal,
+    conservative_weights,
+    controller_decide,
+    exploratory_weights,
+    phase_1_trigger_check,
+    safe_weights,
+)
+
+
+# ---------------------------------------------------------------------
+# Test helpers
+# ---------------------------------------------------------------------
+
+
+def _b(
+    branch_id: str = "b1",
+    *,
+    delta_5s: float = 0.05,
+    delta_5t: float = 0.05,
+    delta_5f: float = 0.05,
+    delta_5r: float = 0.0,
+    witness_divergence: float = 0.0,
+    capital_cost: float = 0.0,
+    regression_penalty: float = 0.0,
+    security_risk: float = 0.0,
+    memory_invalidation: float = 0.0,
+    selfmodel_calibration_gain: float = 0.0,
+    warrant_promotion_gain: float = 0.0,
+    hard_vetoes: tuple[str, ...] = (),
+    payoff_b: float = 1.0,
+) -> ControllerBranch:
+    return ControllerBranch(
+        branch_id=branch_id,
+        deltas=BatteryDeltas(delta_5s, delta_5t, delta_5f, delta_5r),
+        witness_divergence=witness_divergence,
+        capital_cost=capital_cost,
+        regression_penalty=regression_penalty,
+        security_risk=security_risk,
+        memory_invalidation=memory_invalidation,
+        selfmodel_calibration_gain=selfmodel_calibration_gain,
+        warrant_promotion_gain=warrant_promotion_gain,
+        hard_vetoes=hard_vetoes,
+        payoff_b=payoff_b,
+    )
+
+
+def _inp(
+    branches: tuple[ControllerBranch, ...],
+    *,
+    organism_root: str = "Ω0",
+    budget: int = 4,
+    hermes_utilization: int = 0,
+    weights: ControllerWeights | None = None,
+    difficulty: float = 1.0,
+    divergence_rate: float = 0.0,
+    hard_regression_rate: float = 0.0,
+    witness_agreement_rate: float = 1.0,
+    capital_pressure: float = 0.0,
+) -> ControllerInput:
+    return ControllerInput(
+        organism_root=organism_root,
+        branches=branches,
+        budget=budget,
+        hermes_utilization=hermes_utilization,
+        weights=weights or safe_weights(),
+        difficulty=difficulty,
+        divergence_rate=divergence_rate,
+        hard_regression_rate=hard_regression_rate,
+        witness_agreement_rate=witness_agreement_rate,
+        capital_pressure=capital_pressure,
+    )
+
+
+# ---------------------------------------------------------------------
+# §16.2 #1 — test_all_vetoed_returns_reject_or_quarantine
+# ---------------------------------------------------------------------
+
+
+def test_all_vetoed_returns_reject_or_quarantine():
+    """Every branch vetoed → label is REJECT or QUARANTINE (or
+    ESCALATE for the most severe class). selected_branch_id None,
+    notes mention ALL_BRANCHES_VETOED."""
+    branches = (
+        _b("a", hard_vetoes=("hard_regression",)),  # REJECT class
+        _b("b", hard_vetoes=("verifier_failure",)),  # REJECT class
+    )
+    d = controller_decide(_inp(branches))
+    assert d.label == "REJECT"
+    assert d.selected_branch_id is None
+    assert "ALL_BRANCHES_VETOED" in d.notes
+
+    # Mixed REJECT + QUARANTINE → QUARANTINE wins (higher severity).
+    branches = (
+        _b("a", hard_vetoes=("hard_regression",)),
+        _b("b", hard_vetoes=("cache_drift",)),
+    )
+    d = controller_decide(_inp(branches))
+    assert d.label == "QUARANTINE"
+
+    # Any ESCALATE-class veto → ESCALATE wins (highest severity).
+    branches = (
+        _b("a", hard_vetoes=("hard_regression",)),
+        _b("b", hard_vetoes=("memory_invalidation_exceeded",)),
+    )
+    d = controller_decide(_inp(branches))
+    assert d.label == "ESCALATE"
+
+
+# ---------------------------------------------------------------------
+# §16.2 #2 — test_zero_budget_returns_deferred
+# ---------------------------------------------------------------------
+
+
+def test_zero_budget_returns_deferred():
+    """B = 0 (Hermes saturated) → DEFERRED for everything, regardless
+    of how good the branches look."""
+    branches = (_b("a", delta_5s=0.2), _b("b", delta_5s=0.3))
+    d = controller_decide(_inp(branches, budget=0))
+    assert d.label == "DEFERRED"
+    assert d.selected_branch_id is None
+    assert all(v == 0.0 for v in d.allocations.values())
+    assert "ZERO_BUDGET" in d.notes
+
+
+# ---------------------------------------------------------------------
+# §16.2 #3 — test_stable_softmax_no_overflow
+# ---------------------------------------------------------------------
+
+
+def test_stable_softmax_no_overflow():
+    """Utilities ±10000 must not overflow exp() — the stable softmax
+    max-shift bounds every scaled-exp in [0, 1]."""
+    # Both very large positive and very large negative inputs.
+    branches = (
+        _b("hot", delta_5s=10000.0, delta_5t=10000.0, delta_5f=10000.0),
+        _b("cold", delta_5s=-10000.0, delta_5t=-10000.0, delta_5f=-10000.0),
+        _b("middle", delta_5s=0.1),
+    )
+    d = controller_decide(_inp(branches))
+    # No NaN / inf escapes.
+    assert math.isfinite(d.entropy)
+    for v in d.allocations.values():
+        assert math.isfinite(v)
+    # Hot branch gets virtually all the budget.
+    assert d.allocations["hot"] > d.allocations["cold"]
+    assert d.allocations["hot"] > d.allocations["middle"]
+
+
+# ---------------------------------------------------------------------
+# §16.2 #4 — test_negative_payoff_gets_zero_allocation
+# ---------------------------------------------------------------------
+
+
+def test_negative_payoff_gets_zero_allocation():
+    """Branch with payoff_b <= 0 — Kelly's "free regression" guard.
+    Allocation zero; if it's the only branch the label REJECTs."""
+    branches = (_b("a", delta_5s=0.1, payoff_b=-0.5),)
+    d = controller_decide(_inp(branches))
+    # Σ_raw == 0 → DEFERRED (the spec-defined exit when no positive
+    # allocation survives). Both DEFERRED and REJECT are valid
+    # "negative-payoff produces no work" outcomes — assert no positive
+    # allocation either way.
+    assert d.allocations["a"] == 0.0
+    assert d.label in {"DEFERRED", "REJECT"}
+
+    # When another branch has positive payoff, the negative one stays
+    # at zero and the positive one selects.
+    branches = (
+        _b("neg", delta_5s=0.1, payoff_b=-0.5),
+        _b("pos", delta_5s=0.1, payoff_b=2.0),
+    )
+    d = controller_decide(_inp(branches))
+    assert d.allocations["neg"] == 0.0
+    assert d.allocations["pos"] > 0.0
+    assert d.selected_branch_id == "pos"
+
+
+# ---------------------------------------------------------------------
+# §16.2 #5 — test_unsupported_carrier_quarantines
+# ---------------------------------------------------------------------
+
+
+def test_unsupported_carrier_quarantines():
+    """`unsupported_carrier` veto on the only branch → QUARANTINE."""
+    branches = (_b("a", hard_vetoes=("unsupported_carrier",)),)
+    d = controller_decide(_inp(branches))
+    assert d.label == "QUARANTINE"
+    assert "ALL_BRANCHES_VETOED" in d.notes
+    assert d.veto_reasons["a"] == ("unsupported_carrier",)
+
+
+# ---------------------------------------------------------------------
+# §16.2 #6 — test_cache_drift_quarantines
+# ---------------------------------------------------------------------
+
+
+def test_cache_drift_quarantines():
+    """`cache_drift` veto on the only branch → QUARANTINE."""
+    branches = (_b("a", hard_vetoes=("cache_drift",)),)
+    d = controller_decide(_inp(branches))
+    assert d.label == "QUARANTINE"
+
+
+# ---------------------------------------------------------------------
+# §16.2 #7 — test_memory_invalidation_above_threshold_escalates
+# ---------------------------------------------------------------------
+
+
+def test_memory_invalidation_above_threshold_escalates():
+    """Selected branch with memory_invalidation >= KAPPA_MEMORY →
+    ESCALATE."""
+    branches = (
+        _b(
+            "a",
+            delta_5s=0.5,
+            delta_5t=0.5,
+            delta_5f=0.5,
+            memory_invalidation=KAPPA_MEMORY,
+        ),
+    )
+    d = controller_decide(_inp(branches))
+    assert d.label == "ESCALATE"
+    assert d.selected_branch_id == "a"
+    assert "MEMORY_INVALIDATION_ABOVE_KAPPA" in d.notes
+
+    # Just below threshold → no escalation.
+    branches = (
+        _b(
+            "a",
+            delta_5s=0.5,
+            delta_5t=0.5,
+            delta_5f=0.5,
+            memory_invalidation=KAPPA_MEMORY - 0.01,
+        ),
+    )
+    d = controller_decide(_inp(branches))
+    assert d.label != "ESCALATE"
+
+
+# ---------------------------------------------------------------------
+# §16.2 #8 — test_high_entropy_increases_difficulty
+# ---------------------------------------------------------------------
+
+
+def test_high_entropy_increases_difficulty():
+    """Multiple equally-attractive branches → high entropy →
+    difficulty_next > difficulty (§7.1 H_norm term is positive).
+
+    Note: under default `witness_agreement_rate=1.0` the agreement
+    pull exactly cancels the entropy drive (both weight 0.1).
+    Setting agreement_rate=0 isolates the entropy contribution."""
+    # Three branches, identical utilities → uniform distribution →
+    # H_norm = 1.0.
+    branches = tuple(
+        _b(f"b{i}", delta_5s=0.05, delta_5t=0.05, delta_5f=0.05)
+        for i in range(3)
+    )
+    initial_difficulty = 1.0
+    d = controller_decide(
+        _inp(
+            branches,
+            difficulty=initial_difficulty,
+            witness_agreement_rate=0.0,
+        )
+    )
+    assert d.entropy >= H_HIGH, (
+        f"three-equal-branch H_norm should be near 1.0, got {d.entropy}"
+    )
+    assert d.difficulty_next > initial_difficulty
+
+
+# ---------------------------------------------------------------------
+# §16.2 #9 — test_low_entropy_decreases_or_preserves_difficulty
+# ---------------------------------------------------------------------
+
+
+def test_low_entropy_decreases_or_preserves_difficulty():
+    """Single overwhelmingly-good branch → low entropy → with
+    `witness_agreement_rate=1.0` and zero divergence/regression
+    pressure, difficulty_next <= difficulty."""
+    branches = (
+        _b("dominant", delta_5s=10.0, delta_5t=10.0, delta_5f=10.0),
+        _b("weak", delta_5s=-5.0, delta_5t=-5.0, delta_5f=-5.0),
+    )
+    initial_difficulty = 5.0
+    d = controller_decide(
+        _inp(
+            branches,
+            difficulty=initial_difficulty,
+            witness_agreement_rate=1.0,
+            divergence_rate=0.0,
+            hard_regression_rate=0.0,
+        )
+    )
+    # Entropy near zero (dominant branch has p ≈ 1).
+    assert d.entropy <= H_LOW
+    # With agreement pull > entropy drive in this regime, difficulty
+    # should not climb.
+    assert d.difficulty_next <= initial_difficulty
+
+
+# ---------------------------------------------------------------------
+# §16.2 #10 — test_divergence_increases_witness_sampling_recommendation
+# ---------------------------------------------------------------------
+
+
+def test_divergence_increases_witness_sampling_recommendation():
+    """High `divergence_rate` → higher difficulty_next (the controller's
+    homeostat: divergence is the thermostat; more difficulty means
+    wider witness sampling in the Phase 3 sweep)."""
+    branches = (_b("a", delta_5s=0.1, delta_5t=0.1, delta_5f=0.1),)
+    initial = 1.0
+
+    low_div = controller_decide(
+        _inp(branches, difficulty=initial, divergence_rate=0.0)
+    )
+    high_div = controller_decide(
+        _inp(branches, difficulty=initial, divergence_rate=0.8)
+    )
+    assert high_div.difficulty_next > low_div.difficulty_next
+
+
+# ---------------------------------------------------------------------
+# §16.2 #11 — test_no_llm_call_in_controller
+# ---------------------------------------------------------------------
+
+
+def _module_imports() -> set[str]:
+    """Parse the prometheus.py source and return the set of imported
+    module names. Catches `import x`, `import x.y`, `from x.y import …`."""
+    here = Path(__file__).resolve().parent
+    src = (here.parent / "arborist" / "substrate" / "prometheus.py").read_text(
+        encoding="utf-8"
+    )
+    tree = ast.parse(src)
+    names: set[str] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                names.add(alias.name)
+        elif isinstance(node, ast.ImportFrom):
+            if node.module:
+                names.add(node.module)
+    return names
+
+
+def test_no_llm_call_in_controller():
+    """Pure function — must not import the LLM gateway."""
+    imports = _module_imports()
+    assert "arborist.qa.client" not in imports, (
+        "controller must not import the LLM client (Phase 1 is pure)"
+    )
+
+
+# ---------------------------------------------------------------------
+# §16.2 #12 — test_controller_does_not_modify_cache_key_inputs
+# ---------------------------------------------------------------------
+
+
+def test_controller_does_not_modify_cache_key_inputs():
+    """Phase 1 controller must not touch the cache_key path. AST
+    inspection confirms `arborist.qa.keys` is not imported."""
+    imports = _module_imports()
+    assert "arborist.qa.keys" not in imports, (
+        "controller must not import the cache_key path (Phase 1 is "
+        "advisory-only over already-committed state)"
+    )
+
+
+# ---------------------------------------------------------------------
+# §16.2 #13 — test_controller_outputs_advisory_event_only
+# ---------------------------------------------------------------------
+
+
+def test_controller_outputs_advisory_event_only():
+    """An ACCEPT-bound input produces `advisory_events` as a tuple of
+    (str, dict) pairs. Phase 2 will persist these; Phase 1 only
+    builds the data structure."""
+    branches = (_b("a", delta_5s=0.5, delta_5t=0.5, delta_5f=0.5),)
+    d = controller_decide(_inp(branches))
+    assert d.label == "ACCEPT"
+    assert isinstance(d.advisory_events, tuple)
+    assert len(d.advisory_events) >= 1
+    for entry in d.advisory_events:
+        assert isinstance(entry, tuple) and len(entry) == 2
+        kind, body = entry
+        assert isinstance(kind, str)
+        assert isinstance(body, dict)
+    # No mutation API was called — verified by `memory_proposals` /
+    # `selfmodel_proposals` being plain tuples of frozen dataclasses
+    # (no side-effects possible).
+    assert isinstance(d.memory_proposals, tuple)
+    assert isinstance(d.selfmodel_proposals, tuple)
+
+
+# ---------------------------------------------------------------------
+# §16.2 #14 — test_zero_mean_divergence_does_not_trigger_phase_1
+# ---------------------------------------------------------------------
+
+
+def test_zero_mean_divergence_does_not_trigger_phase_1():
+    """Mean divergence = 0 + zero stddev → no signal → no trigger.
+    Absolute-stddev fallback path used when mean is zero but variance
+    is present."""
+    # All-zero samples → mean = 0, stddev = 0 → no_variance_signal.
+    triggered, reason = phase_1_trigger_check([0.0] * 40)
+    assert triggered is False
+    assert reason == "no_variance_signal"
+
+    # Mean = 0 but nonzero stddev below the absolute threshold.
+    samples = [0.0, 0.001, -0.001, 0.0] * 10  # n=40, mean=0, tiny stddev
+    triggered, reason = phase_1_trigger_check(samples)
+    assert triggered is False
+    assert reason == "absolute_stddev_path"
+
+    # Mean = 0 with nonzero stddev above the absolute threshold → fires.
+    # Balance positive and negative so the mean stays at 0.
+    samples = [0.2, -0.2] * 20  # n=40, mean=0, stddev=0.2 >= 0.05.
+    triggered, reason = phase_1_trigger_check(samples)
+    assert triggered is True
+    assert reason == "absolute_stddev_path"
+
+
+# ---------------------------------------------------------------------
+# §16.2 #15 — test_small_sample_does_not_trigger_phase_1
+# ---------------------------------------------------------------------
+
+
+def test_small_sample_does_not_trigger_phase_1():
+    """sample_count < N_min (= 30) → no trigger regardless of variance.
+    Tiny samples are noise."""
+    # 29 samples — high variance, below the floor.
+    rng = random.Random(0)
+    samples = [rng.uniform(0.0, 1.0) for _ in range(N_MIN_TRIGGER_2 - 1)]
+    triggered, reason = phase_1_trigger_check(samples)
+    assert triggered is False
+    assert reason == "small_sample"
+
+    # Empty sample — also small.
+    triggered, reason = phase_1_trigger_check([])
+    assert triggered is False
+    assert reason == "small_sample"
+
+
+# ---------------------------------------------------------------------
+# §16.2 #16 — test_propose_not_mutate_memory_root
+# ---------------------------------------------------------------------
+
+
+def test_propose_not_mutate_memory_root():
+    """A vetoed branch with regression_penalty > 0 yields a
+    MemoryRootUpdateProposal record. Type is the proposal dataclass
+    (no mutation API)."""
+    branches = (
+        _b(
+            "regressed",
+            hard_vetoes=("hard_regression",),
+            regression_penalty=0.4,
+        ),
+        _b("live", delta_5s=0.1),
+    )
+    d = controller_decide(_inp(branches))
+    assert len(d.memory_proposals) == 1
+    prop = d.memory_proposals[0]
+    assert isinstance(prop, MemoryRootUpdateProposal)
+    assert prop.branch_id == "regressed"
+    assert "BRANCH_REGRESSED" in prop.proposed_invalidation_tags
+    # Frozen dataclass — no mutation surface.
+    with pytest.raises((AttributeError, TypeError)):
+        prop.proposed_invalidation_tags = ()  # type: ignore[misc]
+
+
+# ---------------------------------------------------------------------
+# §16.2 #17 — test_propose_not_mutate_self_model
+# ---------------------------------------------------------------------
+
+
+def test_propose_not_mutate_self_model():
+    """A selected branch with selfmodel_calibration_gain != 0 yields a
+    SelfModelUpdateProposal record. Type is the proposal dataclass
+    (no SelfModel mutation API call)."""
+    branches = (
+        _b(
+            "calibrating",
+            delta_5s=0.5,
+            delta_5t=0.5,
+            delta_5f=0.5,
+            selfmodel_calibration_gain=0.15,
+        ),
+    )
+    d = controller_decide(_inp(branches))
+    assert d.selected_branch_id == "calibrating"
+    assert len(d.selfmodel_proposals) == 1
+    prop = d.selfmodel_proposals[0]
+    assert isinstance(prop, SelfModelUpdateProposal)
+    assert prop.calibration_delta == 0.15
+    assert prop.reason == "controller_phase_1_calibration"
+    with pytest.raises((AttributeError, TypeError)):
+        prop.calibration_delta = 0.0  # type: ignore[misc]
+
+
+# ---------------------------------------------------------------------
+# Additional sanity checks (not in §16.2 — harden the pure-function
+# contract)
+# ---------------------------------------------------------------------
+
+
+def test_idempotent_byte_identical_output():
+    """Pure function: two invocations with identical inputs produce
+    byte-identical ControllerDecision."""
+    branches = (
+        _b("a", delta_5s=0.1, delta_5t=0.2, delta_5f=0.3),
+        _b("b", delta_5s=-0.1, delta_5t=0.0, delta_5f=0.2, capital_cost=1.0),
+    )
+    inp = _inp(branches)
+    d1 = controller_decide(inp)
+    d2 = controller_decide(inp)
+    assert d1 == d2
+
+
+def test_weight_profiles_registry_complete():
+    """All three named profiles present and structurally distinct."""
+    assert set(WEIGHT_PROFILES) == {"safe", "conservative", "exploratory"}
+    safe = WEIGHT_PROFILES["safe"]
+    cons = WEIGHT_PROFILES["conservative"]
+    expl = WEIGHT_PROFILES["exploratory"]
+    # §15.2 deltas
+    assert cons.gamma_5f == 1.0
+    assert cons.lambda_capital_cost == 1.5
+    assert cons.xi_security_risk == 4.0
+    assert cons.eta_softmax_temperature == 0.75
+    # §15.3 deltas
+    assert expl.gamma_5f == 1.5
+    assert expl.nu_witness_divergence == 0.75
+    assert expl.lambda_capital_cost == 0.5
+    assert expl.eta_softmax_temperature == 1.5
+    # Safe baseline
+    assert safe.gamma_5f == 1.25
+
+
+def test_phase_1_trigger_check_positive_path():
+    """Coefficient-of-variation path fires when the sample has clear
+    nonzero mean + comparable stddev."""
+    rng = random.Random(1)
+    # Mean ≈ 0.3, stddev ≈ 0.2 → CoV ≈ 0.67 >= 0.5 → fires.
+    samples = [rng.uniform(0.0, 0.6) for _ in range(60)]
+    triggered, reason = phase_1_trigger_check(samples)
+    # Mean and stddev depend on the seeded RNG, but the seed is
+    # fixed → result is deterministic. Sanity-check the reason set.
+    assert reason in {
+        "variance_threshold",
+        "variance_below_threshold",
+        "absolute_stddev_path",
+    }
+    if triggered:
+        assert reason in {"variance_threshold", "absolute_stddev_path"}
+
+
+def test_empty_branches_returns_unknown():
+    """No candidate branches → UNKNOWN with NO_CANDIDATE_BRANCHES note
+    (§14 row 1)."""
+    d = controller_decide(_inp(branches=()))
+    assert d.label == "UNKNOWN"
+    assert d.selected_branch_id is None
+    assert "NO_CANDIDATE_BRANCHES" in d.notes
+
+
+def test_decision_is_a_controllerdecision():
+    """Output type sanity."""
+    branches = (_b("a", delta_5s=0.1),)
+    d = controller_decide(_inp(branches))
+    assert isinstance(d, ControllerDecision)

tests/test_prometheus_sigma.py

@@ -1,217 +0,0 @@
-"""Phase 1 test surface for ticket #000037 (Prometheus-Σ controller).
-
-Phase 0 is doc-only — these tests are the **scaffolding** the §16.2
-implementation contract names. All of them skip until Phase 1 lands
-the controller module (``arborist/substrate/prometheus.py``,
-per the post-2026-05-10 topic-named convention; the original
-§13 sketch said ``arborist/v9/prometheus.py`` before the v-dir
-namespace pattern was retired).
-
-Why land scaffolding while Phase 0 is still doc-only:
-
-- The 17 named tests are the §16.2 acceptance contract; pinning them
-  here means a future shift can't drift the contract by accident.
-- ``pytest --collect-only`` lists them, so the test surface is
-  discoverable from the test runner today rather than buried in a
-  ticket.
-- When Phase 1 lands, the implementer flips
-  ``CONTROLLER_AVAILABLE = True``, drops the body of each test, and
-  the contract enforces itself.
-
-Per CLAUDE.md "no half-finished implementations": these stubs are
-*explicitly* documented as scaffolding. Each test contains a one-
-sentence intent line tied to the controller invariant it pins; that
-intent is what the eventual implementation must satisfy.
-
-Phase 1 trigger gating (§12) is a separate concern measured by
-``bench/prometheus_sigma_trigger_probe.py`` — these tests will run
-regardless of trigger state once the module is in place.
-"""
-
-from __future__ import annotations
-
-import pytest
-
-try:
-    from arborist.substrate import prometheus as _prom  # noqa: F401
-    CONTROLLER_AVAILABLE = True
-except ImportError:
-    CONTROLLER_AVAILABLE = False
-
-skip_until_phase_1 = pytest.mark.skipif(
-    not CONTROLLER_AVAILABLE,
-    reason=(
-        "ticket #000037 Phase 1 not landed; controller module "
-        "arborist/substrate/prometheus.py absent. Skip is the contract — "
-        "implementer flips this when the module exists."
-    ),
-)
-
-
-# ----------------------------------------------------------- core decision
-
-
-@skip_until_phase_1
-def test_all_vetoed_returns_reject_or_quarantine():
-    """§14 row 2: all branches hard-vetoed → emit veto reasons; no
-    allocation; label ``REJECT`` or ``QUARANTINE`` (depending on the
-    veto class). Catches the "every branch is poisoned" case."""
-    pytest.fail("Phase 1 implementation pending; see §13 step 3.")
-
-
-@skip_until_phase_1
-def test_zero_budget_returns_deferred():
-    """§14 row 3: ``B = 0`` → no LLM call; queue sleep if useful;
-    label ``DEFERRED``. Distinguishes "didn't evaluate due to budget"
-    from "evaluated but uncertain" (``MARGINAL``). Per David review
-    point 3."""
-    pytest.fail("Phase 1 implementation pending; see §4.3 + §14.")
-
-
-@skip_until_phase_1
-def test_stable_softmax_no_overflow():
-    """§5: softmax must use ``exp(z_i − max z) / Σ exp(z_j − max z)``.
-    With large positive utilities (e.g. z = 1000) naïve ``exp(z)``
-    overflows; the normalized form must not. Per David review point 6."""
-    pytest.fail("Phase 1 implementation pending; see §5 + §13 step 5.")
-
-
-@skip_until_phase_1
-def test_negative_payoff_gets_zero_allocation():
-    """§7 Kelly safety guard: ``b_i ≤ 0`` → zero allocation. A branch
-    with negative expected payoff must not consume budget. Per David
-    review point 8."""
-    pytest.fail("Phase 1 implementation pending; see §7.")
-
-
-# ----------------------------------------------------------- vetoes
-
-
-@skip_until_phase_1
-def test_unsupported_carrier_quarantines():
-    """§6 hard-veto class: claim references a carrier modality that
-    no live π* library supports → ``QUARANTINE``. Surfaces missing
-    domain coverage rather than papering over it."""
-    pytest.fail("Phase 1 implementation pending; see §6.")
-
-
-@skip_until_phase_1
-def test_cache_drift_quarantines():
-    """§6 hard-veto class: cache row whose ``pi_star_ref`` no longer
-    matches a live kernel version → ``QUARANTINE``. Echoes the
-    CACHE-DRIFT outcome from #000028 §1.2; controller surfaces
-    rather than silently re-uses."""
-    pytest.fail("Phase 1 implementation pending; see §6 + #000028 §1.2.")
-
-
-@skip_until_phase_1
-def test_memory_invalidation_above_threshold_escalates():
-    """§10 Gödel discipline: a branch whose acceptance would
-    invalidate too much committed memory → label ``ESCALATE``, not
-    ``REJECT``. The controller explicitly steps aside per the "must
-    never infer" rule. Per David review point 12."""
-    pytest.fail("Phase 1 implementation pending; see §10.")
-
-
-# ----------------------------------------------------------- difficulty / EMA
-
-
-@skip_until_phase_1
-def test_high_entropy_increases_difficulty():
-    """§7.1 difficulty update law: high ``H_norm(p)`` → smoothed
-    increase in ``difficulty_ema``. The EMA smoothing keeps the
-    update from overshooting on a single noisy sample. Per David
-    review point 9."""
-    pytest.fail("Phase 1 implementation pending; see §7.1.")
-
-
-@skip_until_phase_1
-def test_low_entropy_decreases_or_preserves_difficulty():
-    """§7.1 difficulty update law: low ``H_norm(p)`` (controller is
-    confident) → difficulty drops or holds. Symmetric to the
-    high-entropy test; together they pin the EMA's monotonicity."""
-    pytest.fail("Phase 1 implementation pending; see §7.1.")
-
-
-@skip_until_phase_1
-def test_divergence_increases_witness_sampling_recommendation():
-    """§4.3 output / §17.1: high observed witness divergence → the
-    controller's recommendation field should bump ``canonical_witness_
-    sample_rate`` upward (an advisory; runtime decides whether to
-    apply). Pins the feedback loop into #000028's sample-rate field
-    we landed in 6d20aeb."""
-    pytest.fail("Phase 1 implementation pending; see §4.3.")
-
-
-# ----------------------------------------------------------- discipline
-
-
-@skip_until_phase_1
-def test_no_llm_call_in_controller():
-    """§10 Gödel + David review point 12: the controller is a pure
-    function. No ``ChatClient`` import in the call graph; no network
-    socket; no env-var that secretly enables one. The witness module
-    calls the LLM; the controller reads its results. Pins the
-    "LLM is witness, never authority" doctrine."""
-    pytest.fail("Phase 1 implementation pending; see §10 + §16.1.")
-
-
-@skip_until_phase_1
-def test_controller_does_not_modify_cache_key_inputs():
-    """§4.4 update authority: the controller emits **proposals** for
-    MemoryRoot / SelfModel updates; it never mutates the cache_key
-    8-dim input itself. Pins the schema invariant from #000027:
-    cache_key is computed by the cache_key() function, not by any
-    advisory layer above it. Per David review point 14."""
-    pytest.fail("Phase 1 implementation pending; see §4.4 + §13 step 12.")
-
-
-@skip_until_phase_1
-def test_controller_outputs_advisory_event_only():
-    """§13 step 10: the controller writes ``controller_decision`` /
-    ``controller_difficulty`` / ``controller_budget_allocation`` as
-    sibling tags on ``audit_events`` — they do NOT enter
-    ``event_hash`` preimage. Re-running the controller against the
-    same state cannot break the audit chain. Pins the same sibling-
-    table invariant the capital_ledger uses."""
-    pytest.fail("Phase 1 implementation pending; see §13 step 10.")
-
-
-# ----------------------------------------------------------- §12 trigger guards
-
-
-@skip_until_phase_1
-def test_zero_mean_divergence_does_not_trigger_phase_1():
-    """§12 Trigger 2: ``max(mean, ε)`` guard against div-by-zero
-    when divergence is uniformly low. Probe covers this; the
-    in-controller guard duplicates it so the controller can be run
-    on a fresh corpus without crashing on the first call."""
-    pytest.fail("Phase 1 implementation pending; see §12 Trigger 2.")
-
-
-@skip_until_phase_1
-def test_small_sample_does_not_trigger_phase_1():
-    """§12 Trigger 2: ``N_min = 30`` floor below which the variance
-    trigger does not fire. Pins behavior on early-corpus deployments
-    where sample-count noise would dominate any signal."""
-    pytest.fail("Phase 1 implementation pending; see §12 Trigger 2.")
-
-
-# ----------------------------------------------------------- proposals, not mutations
-
-
-@skip_until_phase_1
-def test_propose_not_mutate_memory_root():
-    """§4.4 + §13 step 12: controller emits a MemoryRoot update
-    *proposal*; the existing memory-root write path validates and
-    commits. Direct mutation would let the controller poison hard-
-    hashed state without going through validation."""
-    pytest.fail("Phase 1 implementation pending; see §4.4.")
-
-
-@skip_until_phase_1
-def test_propose_not_mutate_self_model():
-    """§4.4 + §13 step 12: controller emits a SelfModel update
-    *proposal*; the existing #000014 selfmodel write path validates
-    and commits. Same boundary as MemoryRoot."""
-    pytest.fail("Phase 1 implementation pending; see §4.4 + #000014.")