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.
This commit is contained in:
russell@unturf.com 2026-05-10 16:35:54 -04:00
parent f5dbfabed5
commit f625cac20c
No known key found for this signature in database
3 changed files with 1390 additions and 217 deletions

View file

@ -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",
]

608
tests/test_prometheus.py Normal file
View file

@ -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)

View file

@ -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.")