diff --git a/arborist/substrate/prometheus.py b/arborist/substrate/prometheus.py new file mode 100644 index 0000000..9820ca6 --- /dev/null +++ b/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", +] diff --git a/tests/test_prometheus.py b/tests/test_prometheus.py new file mode 100644 index 0000000..68202af --- /dev/null +++ b/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) diff --git a/tests/test_prometheus_sigma.py b/tests/test_prometheus_sigma.py deleted file mode 100644 index 93e849a..0000000 --- a/tests/test_prometheus_sigma.py +++ /dev/null @@ -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.")