diff --git a/CHANGELOG.md b/CHANGELOG.md index e900dd1..9f60bfb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -84,6 +84,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added (v0.3.0 candidate) +- **Executive band package** (`substrate.executive`) — the resistance band made + operational as a decision engine. Two named lenses on one utilization value — + `LoadZone` (the load lens: IDLE/RECREATION/WORK/PEAKING/WARNING/DANGER on the + symmetric φ-conjugate ladder) and `CyclePhase` (the cycle lens: ASCENDING/PIVOT/ + PAST_PIVOT over the 24-step work span) — with `classify_load_zone` / + `classify_cycle_phase`. Levels are geometric; consequences are temporal (no + spike-tolerance field — that belongs to the `SustainedLoadTracker`). A + structurally-validated `BandProfile` (R1–R5: ordering, φ-anchors, conjugate sum, + symmetry, RESISTANCE tighten-only), the `Quantity`/`Cycle`/`ResourceKind` + discriminators + `setpoint_for` (RESISTANCE vs WORK bands; GROWTH rejected), and + `order_index` / `negentropy` (the order-from-disorder emergence metric, `1 −` + normalised Shannon entropy + its trend). Lifted into the top-level `substrate` + namespace. 29 conformance tests; pyright clean; pylint 10.00. + - **Multi-scale observation Protocol** ([`spec/multi-scale.md`](spec/multi-scale.md)) — pluggable `SubstrateScope` Protocol with default `cell` / `node` / `org` triple + operator-extensible registry. Hyphenated package diff --git a/python/src/substrate/__init__.py b/python/src/substrate/__init__.py index a4424e5..7fd0702 100644 --- a/python/src/substrate/__init__.py +++ b/python/src/substrate/__init__.py @@ -58,6 +58,17 @@ NetPotentialGainVerdict, RaiseOnNegativeGate, ) +from substrate.executive import ( + BandProfile, + CyclePhase, + LoadZone, + Quantity, + classify_cycle_phase, + classify_load_zone, + negentropy, + order_index, + setpoint_for, +) from substrate.resistance_band import ( DEFAULT_CONFIG, LOWER_BOUND, @@ -147,6 +158,16 @@ "assess", "classify", "recommend_scaling_factor", + # Executive band (the band as a decision engine) + "BandProfile", + "CyclePhase", + "LoadZone", + "Quantity", + "classify_cycle_phase", + "classify_load_zone", + "negentropy", + "order_index", + "setpoint_for", # Version "__version__", ] diff --git a/python/src/substrate/executive/__init__.py b/python/src/substrate/executive/__init__.py new file mode 100644 index 0000000..36d1c65 --- /dev/null +++ b/python/src/substrate/executive/__init__.py @@ -0,0 +1,70 @@ +"""The executive layer — the band as a decision engine. + +The resistance band, made operational as the substrate's executive function: the +corrected symmetric ladder (geometric levels, temporal consequences), the +quantity/scale discipline that keeps "what percentage is this and what does it +mean" precise, and the order metric that reads emergence from a distribution. + +Curated exports — import the names, not deep module paths. +""" +from __future__ import annotations + +from substrate.executive.band import ( + BAND_TOLERANCE, + DEFAULT_BAND_PROFILE, + TWO_THIRDS, + BandProfile, + BandProfileInvalid, + CyclePhase, + LoadZone, + classify_cycle_phase, + classify_load_zone, + validate_band_profile, + zone_to_legacy, +) +from substrate.executive.negentropy import ( + NegentropyDirection, + NegentropyReport, + negentropy, + order_index, +) +from substrate.executive.observed_graph import ( + EntityRollup, + ExtractionReport, + NpgEdge, + detect_extraction, +) +from substrate.executive.quantities import ( + Cycle, + GrowthNotADecisionBand, + Quantity, + ResourceKind, + setpoint_for, +) + +__all__ = [ + "BAND_TOLERANCE", + "DEFAULT_BAND_PROFILE", + "TWO_THIRDS", + "BandProfile", + "BandProfileInvalid", + "Cycle", + "CyclePhase", + "EntityRollup", + "ExtractionReport", + "GrowthNotADecisionBand", + "LoadZone", + "NegentropyDirection", + "NegentropyReport", + "NpgEdge", + "Quantity", + "ResourceKind", + "classify_cycle_phase", + "classify_load_zone", + "detect_extraction", + "negentropy", + "order_index", + "setpoint_for", + "validate_band_profile", + "zone_to_legacy", +] diff --git a/python/src/substrate/executive/band.py b/python/src/substrate/executive/band.py new file mode 100644 index 0000000..fbbe859 --- /dev/null +++ b/python/src/substrate/executive/band.py @@ -0,0 +1,228 @@ +"""The band — two lenses on one resistance value. + +The corrected **symmetric ladder**, anchored on the φ-conjugates and the thirds, +mirror-symmetric about the 0.50 pivot: + + 1/3 (0.3333) · 1/φ² (0.3820) · 0.50 · 1/φ (0.6180) · 2/3 (0.6667) + +Two named lenses on the same utilization value (never a name collision): + +- :class:`LoadZone` — the LOAD lens: "how loaded am I right now." +- :class:`CyclePhase` — the CYCLE lens: position on the 24-step work-span cycle. + +**Levels are geometric, consequences are temporal.** :func:`classify_load_zone` +only says *where* the instantaneous reading sits. Whether you are *actually* +pivoting or taking damage is a sustained-vs-spike call owned by the +``SustainedLoadTracker`` (see :mod:`substrate.sustained_load`) — never a geometric +window here. A transient peak to 0.62 is healthy (intervals); 0.62 *sustained* is +overload. There is **no spike-tolerance field** on the profile: the temporal +tolerance is the tracker's, so there is nothing to desync. + +The φ anchors are reused from :mod:`substrate.resistance_band` (one source of +constants); the legacy :class:`substrate.resistance_band.ZoneClassification` is the +wire-compatible projection target. +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from typing import Final + +from substrate.executive.quantities import Quantity, ResourceKind +from substrate.resistance_band import ( + LOWER_BOUND, + PHI_CONJUGATE, + UPPER_BOUND, + WORK_ZONE_UPPER, + ZoneClassification, +) + +#: ``2/3`` — the danger line; sustained above it = damage/debt. The thirds +#: ``1/3 + 2/3 = 1`` are a conjugate pair, mirror of the φ-conjugates. +TWO_THIRDS: Final[float] = 2.0 / 3.0 + +#: Structural tolerance for the φ-anchor / symmetry rules (R2–R4). The structure +#: is invariant; the exact numbers within are free (framework-with-tolerance). +BAND_TOLERANCE: Final[float] = 0.05 + + +class LoadZone(str, Enum): + """The LOAD lens — how loaded an entity is right now.""" + + IDLE = "idle" # < 1/3 sustained → decay/atrophy + RECREATION = "recreation" # 1/3 … 1/φ² light/enjoyable; resistance setpoint + WORK = "work" # 1/φ² … 0.50 the only SUSTAINABLE cruise + PEAKING = "peaking" # 0.50 … 1/φ growth — TRANSIENT peaks build + WARNING = "warning" # 1/φ … 2/3 winded; mirror of RECREATION + DANGER = "danger" # > 2/3 sustained → breakdown/damage/debt + + +class CyclePhase(str, Enum): + """The CYCLE lens — position on the 24-step work-and-growth span.""" + + ASCENDING = "ascending" # below the 0.50 half-period pivot (building) + PIVOT = "pivot" # at the pivot (position 12) — half-period reversal + PAST_PIVOT = "past_pivot" # above the pivot (the cycle's reverse half) + + +class BandProfileInvalid(ValueError): + """Raised when a :class:`BandProfile` violates a structural rule (R1–R5).""" + + def __init__(self, rule: str, detail: str) -> None: + super().__init__(f"{rule}: {detail}") + self.rule = rule + self.detail = detail + + +@dataclass(frozen=True, slots=True) +class BandProfile: + """The geometric levels, parameterizable but structurally validated. + + Defaults are the φ anchors. A caller may shift the numbers (per-resource + tuning) but not the *structure* — :func:`validate_band_profile` enforces the + ordering, the φ-anchors, the conjugate sum, the symmetry, and (for a + RESISTANCE profile) tighten-only. There is no ``spike_tolerance`` — the + temporal tolerance belongs to the tracker. + """ + + idle_ceiling: float = LOWER_BOUND # 1/3 — IDLE strictly below + recreation_ceiling: float = UPPER_BOUND # 1/φ² — RECREATION top / WORK bottom + pivot: float = WORK_ZONE_UPPER # 0.50 — WORK top / PEAKING bottom + growth_ceiling: float = PHI_CONJUGATE # 1/φ — PEAKING top / WARNING bottom + danger_line: float = TWO_THIRDS # 2/3 — WARNING top; above = damage + quantity: Quantity | None = None # if RESISTANCE, R5 applies + resource: ResourceKind = field(default=ResourceKind.GENERIC) + + def __post_init__(self) -> None: + validate_band_profile(self) + + +def validate_band_profile(p: BandProfile) -> None: + """Raise :class:`BandProfileInvalid` unless all structural rules hold.""" + tol = BAND_TOLERANCE + # R1 — ordering. + if not ( + 0.0 < p.idle_ceiling < p.recreation_ceiling < p.pivot + < p.growth_ceiling < p.danger_line < 1.0 + ): + raise BandProfileInvalid( + "R1", + "require 0 < idle < recreation < pivot < growth < danger < 1; got " + f"{p.idle_ceiling}/{p.recreation_ceiling}/{p.pivot}/" + f"{p.growth_ceiling}/{p.danger_line}", + ) + # R2 — φ anchors. + if abs(p.recreation_ceiling - UPPER_BOUND) > tol: + raise BandProfileInvalid( + "R2", f"recreation_ceiling must be within {tol} of 1/φ² ({UPPER_BOUND:.4f})" + ) + if abs(p.growth_ceiling - PHI_CONJUGATE) > tol: + raise BandProfileInvalid( + "R2", f"growth_ceiling must be within {tol} of 1/φ ({PHI_CONJUGATE:.4f})" + ) + # R3 — conjugate sum (1/φ² + 1/φ = 1). + if abs(p.recreation_ceiling + p.growth_ceiling - 1.0) > tol: + raise BandProfileInvalid( + "R3", "recreation_ceiling + growth_ceiling must be within " + f"{tol} of 1.0 (φ-conjugate sum)" + ) + # R4 — symmetry about the pivot (and the thirds conjugate 1/3 + 2/3 = 1). + midpoint = (p.recreation_ceiling + p.growth_ceiling) / 2.0 + if abs(p.pivot - midpoint) > tol: + raise BandProfileInvalid( + "R4", f"pivot must be within {tol} of the φ-anchor midpoint {midpoint:.4f}" + ) + if abs(p.idle_ceiling + p.danger_line - 1.0) > tol: + raise BandProfileInvalid( + "R4", "idle_ceiling + danger_line must be within " + f"{tol} of 1.0 (thirds conjugate)" + ) + # R5 — resistance tighten-only (no widening to escape challenge). + if p.quantity is Quantity.RESISTANCE: + if p.idle_ceiling < LOWER_BOUND - 1e-9: + raise BandProfileInvalid( + "R5", "RESISTANCE: idle_ceiling may only TIGHTEN (≥ 1/3)" + ) + if p.recreation_ceiling > UPPER_BOUND + 1e-9: + raise BandProfileInvalid( + "R5", "RESISTANCE: recreation_ceiling may only TIGHTEN (≤ 1/φ²)" + ) + + +#: The canonical φ-anchored profile. +DEFAULT_BAND_PROFILE: Final[BandProfile] = BandProfile() + + +def classify_load_zone( + u: float, profile: BandProfile = DEFAULT_BAND_PROFILE +) -> LoadZone: + """Classify a utilization reading into a :class:`LoadZone`. + + Each zone is ``(prev_ceiling, ceiling]`` (upper-inclusive) except + ``IDLE = [0, idle_ceiling)`` — so ``1/3`` belongs to RECREATION. This is the + geometric *where*; SPIKE-vs-SUSTAINED (the consequences) is the tracker's. + """ + if u < profile.idle_ceiling: + return LoadZone.IDLE + if u <= profile.recreation_ceiling: + return LoadZone.RECREATION + if u <= profile.pivot: + return LoadZone.WORK + if u <= profile.growth_ceiling: + return LoadZone.PEAKING + if u <= profile.danger_line: + return LoadZone.WARNING + return LoadZone.DANGER + + +def classify_cycle_phase( + u: float, profile: BandProfile = DEFAULT_BAND_PROFILE +) -> CyclePhase: + """Classify a reading's position relative to the 0.50 half-period pivot. + + The work-and-growth span ``[recreation_ceiling, growth_ceiling]`` ≈ + ``[0.38, 0.62]`` is the 24-position cycle (12 steps each side of the pivot); + the PIVOT window is half a step wide. + """ + span = profile.growth_ceiling - profile.recreation_ceiling + half_step = (span / 24.0) / 2.0 + if u < profile.pivot - half_step: + return CyclePhase.ASCENDING + if u > profile.pivot + half_step: + return CyclePhase.PAST_PIVOT + return CyclePhase.PIVOT + + +_ZONE_TO_LEGACY: Final[dict[LoadZone, ZoneClassification]] = { + LoadZone.IDLE: ZoneClassification.UNDER_LOADED, + LoadZone.RECREATION: ZoneClassification.CALIBRATION, + LoadZone.WORK: ZoneClassification.WORKING, + LoadZone.PEAKING: ZoneClassification.WORKING, + LoadZone.WARNING: ZoneClassification.PEAKING, + LoadZone.DANGER: ZoneClassification.DEBT, +} + + +def zone_to_legacy(zone: LoadZone) -> ZoneClassification: + """Project a :class:`LoadZone` to the legacy 5-band classification. + + Serialization / wire compatibility for consumers on the older layered-zone + enum. The projection is exact and lossy-by-design (the new PEAKING growth zone + maps to legacy WORKING; new WARNING → legacy PEAKING; new DANGER → legacy DEBT). + """ + return _ZONE_TO_LEGACY[zone] + + +__all__ = [ + "BAND_TOLERANCE", + "DEFAULT_BAND_PROFILE", + "TWO_THIRDS", + "BandProfile", + "BandProfileInvalid", + "CyclePhase", + "LoadZone", + "classify_cycle_phase", + "classify_load_zone", + "validate_band_profile", + "zone_to_legacy", +] diff --git a/python/src/substrate/executive/negentropy.py b/python/src/substrate/executive/negentropy.py new file mode 100644 index 0000000..daee815 --- /dev/null +++ b/python/src/substrate/executive/negentropy.py @@ -0,0 +1,134 @@ +"""Negentropy / order metric — order-from-disorder, the emergence tell. + +The substrate's central claim is *emergent order*: local iteration converges to +coherent structure (closed cycles, mode agreement) rather than scattering into +disorder. This module makes that mechanical and measurable. + +* :func:`order_index` — the INSTANTANEOUS order in a distribution, ``1 - + normalised Shannon entropy``. A distribution concentrated in one category + (everything aligned) is maximally ordered (``1.0``); a uniform distribution + (maximal disagreement) is maximally disordered (``0.0``). The mechanical inverse + of entropy — no tuning, just information theory. +* :func:`negentropy` — order OVER TIME. Rising order is *emergence* (the + negentropic direction); falling order is *decay* (entropy winning). The trend + + its rate is the negentropy the system is producing or losing. + +Feeds naturally from any substrate distribution — a mode distribution, a vote +tally, a zone distribution — sampled over a window. + +Pure logic +========== + +* No DAO, no LLM, no network. Deterministic. +* Frozen dataclasses with slots throughout. +""" +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from math import log2 +from typing import Final, Sequence + + +def order_index(counts: Sequence[int]) -> float: + """Instantaneous order of a distribution — ``1 - H/H_max`` in ``[0, 1]``. + + ``counts`` are the per-category frequencies (zero counts ignored). One + populated category → maximal order (``1.0``); a uniform spread over ``N`` + categories → maximal disorder (``0.0``). Raises for an empty / all-zero + distribution (no order is defined over nothing). + """ + positive = [c for c in counts if c > 0] + for c in counts: + if c < 0: + raise ValueError(f"counts must be >= 0; got {c!r}") + total = sum(positive) + if total == 0: + raise ValueError("at least one positive count is required") + n = len(positive) + if n <= 1: + return 1.0 # a single occupied category is perfectly ordered + entropy = -sum( + (c / total) * log2(c / total) for c in positive + ) + return 1.0 - entropy / log2(n) + + +class NegentropyDirection(str, Enum): + """Which way order is moving over the window.""" + + EMERGING = "emerging" # order rising — the substrate is self-organising + STABLE = "stable" # order holding + DECAYING = "decaying" # order falling — entropy winning + + +#: Below this absolute order-delta the change reads as noise, not a direction. +_ORDER_DEADBAND: Final[float] = 0.02 + + +@dataclass(frozen=True, slots=True) +class NegentropyReport: + """Order at the latest reading + its direction over the window.""" + + current_order: float + earlier_order: float + order_delta: float + direction: NegentropyDirection + sample_count: int + rationale: str + + +def negentropy(order_history: Sequence[float]) -> NegentropyReport: + """Classify order's movement over a window of :func:`order_index` values. + + Compares the recent half's mean order to the earlier half's: rising past the + dead-band is EMERGING (negentropic — order from disorder), falling is + DECAYING, otherwise STABLE. A single reading is trivially STABLE. + + Raises for an empty history or an out-of-range order value. + """ + if not order_history: + raise ValueError("order_history must be non-empty") + for v in order_history: + if not 0.0 <= v <= 1.0: + raise ValueError(f"order values must be in [0, 1]; got {v!r}") + current = float(order_history[-1]) + if len(order_history) == 1: + return NegentropyReport( + current_order=current, + earlier_order=current, + order_delta=0.0, + direction=NegentropyDirection.STABLE, + sample_count=1, + rationale=f"single reading order={current:.3f}; trend undefined → stable", + ) + mid = len(order_history) // 2 + earlier = sum(order_history[:mid]) / mid + recent = sum(order_history[mid:]) / (len(order_history) - mid) + delta = recent - earlier + if delta > _ORDER_DEADBAND: + direction = NegentropyDirection.EMERGING + elif delta < -_ORDER_DEADBAND: + direction = NegentropyDirection.DECAYING + else: + direction = NegentropyDirection.STABLE + rationale = ( + f"order {earlier:.3f}→{recent:.3f} (Δ{delta:+.3f}) over " + f"{len(order_history)} readings → {direction.value}" + ) + return NegentropyReport( + current_order=current, + earlier_order=earlier, + order_delta=delta, + direction=direction, + sample_count=len(order_history), + rationale=rationale, + ) + + +__all__ = [ + "NegentropyDirection", + "NegentropyReport", + "negentropy", + "order_index", +] diff --git a/python/src/substrate/executive/observed_graph.py b/python/src/substrate/executive/observed_graph.py new file mode 100644 index 0000000..8336cb4 --- /dev/null +++ b/python/src/substrate/executive/observed_graph.py @@ -0,0 +1,146 @@ +"""NPG calculus on the OBSERVED entity graph — extraction detection. + +A deliberation engine runs the substrate calculus over *candidate actions the +system might take*. This module runs the same calculus over +an *observed graph of relationships between other entities* — the perception / +ontology pipeline fuses multi-modal observation into one entity graph, and here +each relationship edge carries a **net-potential-gain sign** (does the actor raise +or lower the target's potential — the *work* done on them) and a **cycle** (a +sustained long-cycle relationship vs a one-off short-cycle hit). + +Reading the graph for one signature finds the predator; reading it for the +complement finds the unprotected (the care side). The signature of extraction +an entity whose **short-cycle takings exceed its long-cycle givings** +— it lowers others' potential in one-off hits while contributing no sustained +support. The substrate's own runaway-power / parasitism tell, applied to observed +data (a shell company draining a subsidiary; a person extracting from dependents). + +Pure logic +========== + +* No DAO, no LLM, no network. Deterministic. Frozen dataclasses with slots. +* Reuses the executive :class:`Cycle` enum (SHORT = one-off, LONG = sustained) — + no shadow enum. +""" +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass +from typing import Dict, Sequence, Tuple + +from substrate.executive.quantities import Cycle + + +@dataclass(frozen=True, slots=True) +class NpgEdge: + """One observed relationship: the actor's NPG effect on the target. + + ``npg_delta`` is the work done on ``target_entity_id`` — positive raises that + entity's potential (support), negative lowers it (a taking). ``cycle`` marks + a one-off (SHORT) vs a sustained (LONG) relationship. + """ + + source_entity_id: str + target_entity_id: str + npg_delta: float + cycle: Cycle + relation: str = "" + + def __post_init__(self) -> None: + if not self.source_entity_id: + raise ValueError("source_entity_id must be non-empty") + if not self.target_entity_id: + raise ValueError("target_entity_id must be non-empty") + + +@dataclass(frozen=True, slots=True) +class EntityRollup: # pylint: disable=too-many-instance-attributes + """One actor entity's aggregated effect on the entities it touches.""" + + entity_id: str + short_cycle_taken: float # Σ harm done in one-off (SHORT) edges + long_cycle_given: float # Σ support given in sustained (LONG) edges + net_potential_caused: float # Σ all npg_delta (signed) + extraction_margin: float # short_cycle_taken − long_cycle_given + is_extractive: bool + is_supportive: bool + edge_count: int + + +@dataclass(frozen=True, slots=True) +class ExtractionReport: + """The graph read for predators and for sustained supporters.""" + + rollups: Tuple[EntityRollup, ...] # all actors, worst-extractor first + extractive: Tuple[EntityRollup, ...] # short-cycle takings > long-cycle giving + supportive: Tuple[EntityRollup, ...] # net long-cycle givers + rationale: str + + +def detect_extraction( + edges: Sequence[NpgEdge], + *, + extraction_threshold: float = 0.0, +) -> ExtractionReport: + """Find extractive (predatory) and supportive entities on the observed graph. + + For each ACTOR (edge source): the harm it does in one-off SHORT-cycle edges is + its ``short_cycle_taken``; the support it gives in sustained LONG-cycle edges is + its ``long_cycle_given``. The ``extraction_margin`` is the difference — an + entity is **extractive** when that margin exceeds ``extraction_threshold`` + (it predates short-cycle while contributing no sustained long-cycle support), + and **supportive** when it is a net long-cycle giver. Actors are returned + worst-extractor-first. + """ + by_actor: Dict[str, list[NpgEdge]] = defaultdict(list) + for edge in edges: + by_actor[edge.source_entity_id].append(edge) + + rollups: list[EntityRollup] = [] + for actor, actor_edges in by_actor.items(): + short_taken = sum( + -e.npg_delta + for e in actor_edges + if e.cycle is Cycle.SHORT and e.npg_delta < 0.0 + ) + long_given = sum( + e.npg_delta + for e in actor_edges + if e.cycle is Cycle.LONG and e.npg_delta > 0.0 + ) + net = sum(e.npg_delta for e in actor_edges) + margin = short_taken - long_given + rollups.append( + EntityRollup( + entity_id=actor, + short_cycle_taken=short_taken, + long_cycle_given=long_given, + net_potential_caused=net, + extraction_margin=margin, + is_extractive=margin > extraction_threshold, + is_supportive=long_given > short_taken, + edge_count=len(actor_edges), + ) + ) + + rollups.sort(key=lambda r: (-r.extraction_margin, r.entity_id)) + extractive = tuple(r for r in rollups if r.is_extractive) + supportive = tuple(r for r in rollups if r.is_supportive) + rationale = ( + f"{len(rollups)} actors over {len(edges)} edges: " + f"{len(extractive)} extractive, {len(supportive)} supportive" + ) + return ExtractionReport( + rollups=tuple(rollups), + extractive=extractive, + supportive=supportive, + rationale=rationale, + ) + + +__all__ = [ + "EntityRollup", + "ExtractionReport", + "NpgEdge", + "detect_extraction", +] diff --git a/python/src/substrate/executive/quantities.py b/python/src/substrate/executive/quantities.py new file mode 100644 index 0000000..296c4e2 --- /dev/null +++ b/python/src/substrate/executive/quantities.py @@ -0,0 +1,91 @@ +"""Quantity / Cycle / ResourceKind + setpoints. + +The three discriminators that make "what percentage is this, and what does it +mean" precise — the discipline that stops a band-shaped number being read as the +wrong band: + +- **Quantity** — *what kind of band* a utilization reading lives in. + ``RESISTANCE`` (a challenge held at the 1/3–1/φ² setpoint), ``WORK`` (the + sustainable cruise in the work zone ≤ 0.50), ``GROWTH`` (φ-stepped transient + peaks — NOT a sustained band; routed through growth-step assessment). +- **Cycle** — ``SHORT`` (latency / random / multitask — a TIGHT band, the + queueing hockey-stick) vs ``LONG`` (throughput / sequential / focus — runs + HOT). The same percentage means different things on each. +- **ResourceKind** — selects per-resource band defaults (CPU vs disk-capacity vs + memory vs network), because a "protected reserve" and a hard-fail ceiling + differ by resource. + +``setpoint_for`` returns the ``(low, high)`` target band for a quantity from a +:class:`~substrate.executive.band.BandProfile`. Pure logic; the band module owns +the levels, this module owns the *meaning*. +""" +from __future__ import annotations + +from enum import Enum +from typing import TYPE_CHECKING, Tuple + +if TYPE_CHECKING: # avoid an import cycle — band imports nothing from here + from substrate.executive.band import BandProfile + + +class Quantity(str, Enum): + """What kind of band a utilization reading lives in.""" + + RESISTANCE = "resistance" # held at the 1/3–1/φ² imposed-challenge setpoint + WORK = "work" # the sustainable cruise (work zone, ≤ pivot) + GROWTH = "growth" # φ-stepped transient peaks — not a decision band + + +class Cycle(str, Enum): + """Latency vs throughput orientation of the work.""" + + SHORT = "short" # latency / random / multitask — TIGHT band + LONG = "long" # throughput / sequential / focus — runs HOT + + +class ResourceKind(str, Enum): + """The resource a utilization reading measures (selects band defaults).""" + + GENERIC = "generic" + CPU = "cpu" + DISK_CAPACITY = "disk_capacity" # non-transferable; tight, hard-fail near full + DISK_IO = "disk_io" + MEMORY = "memory" # hard-fail + NETWORK = "network" # microbursts + + +class GrowthNotADecisionBand(ValueError): + """Raised when ``GROWTH`` is used where a decision band is required. + + GROWTH is φ-stepped transient peaks assessed by the growth-step path, not a + sustained band the decision join operates over. + """ + + +def setpoint_for(quantity: Quantity, profile: "BandProfile") -> Tuple[float, float]: + """Return the ``(low, high)`` target band for ``quantity`` from ``profile``. + + - ``RESISTANCE`` → ``(idle_ceiling, recreation_ceiling)`` ≈ ``(1/3, 1/φ²)``: + the imposed-challenge setpoint (held, not exceeded). + - ``WORK`` → ``(recreation_ceiling, pivot)`` ≈ ``(1/φ², 0.50)``: the only + indefinitely-sustainable cruise (the work zone). + - ``GROWTH`` → raises :class:`GrowthNotADecisionBand` (transient peaks are + assessed by the growth-step path, not a sustained setpoint band). + """ + if quantity is Quantity.RESISTANCE: + return (profile.idle_ceiling, profile.recreation_ceiling) + if quantity is Quantity.WORK: + return (profile.recreation_ceiling, profile.pivot) + raise GrowthNotADecisionBand( + "GROWTH has no sustained setpoint band — use the growth-step path " + "(φ-stepped transient peaks + consolidation), not setpoint_for" + ) + + +__all__ = [ + "Cycle", + "GrowthNotADecisionBand", + "Quantity", + "ResourceKind", + "setpoint_for", +] diff --git a/python/tests/test_executive_band.py b/python/tests/test_executive_band.py new file mode 100644 index 0000000..6a1c2b0 --- /dev/null +++ b/python/tests/test_executive_band.py @@ -0,0 +1,119 @@ +"""Conformance tests for the executive band — LoadZone / CyclePhase / BandProfile. + +Covers: +- classify_load_zone across the six zones + boundary inclusivity (1/3 is RECREATION) +- classify_cycle_phase: pivot window derived from the 24-step cycle, not invented +- BandProfile structural validation R1-R5 (ordering, φ-anchors, conjugate sum, + symmetry, RESISTANCE tighten-only) +- setpoint_for: RESISTANCE vs WORK bands; GROWTH rejected +- zone_to_legacy projection to the 5-band ZoneClassification +""" +from __future__ import annotations + +import pytest + +from substrate.executive.band import ( + BandProfile, + BandProfileInvalid, + CyclePhase, + LoadZone, + classify_cycle_phase, + classify_load_zone, + zone_to_legacy, +) +from substrate.executive.quantities import ( + GrowthNotADecisionBand, + Quantity, + setpoint_for, +) +from substrate.resistance_band import ( + LOWER_BOUND, + UPPER_BOUND, + WORK_ZONE_UPPER, + ZoneClassification, +) + + +class TestClassifyLoadZone: + def test_idle_below_third(self) -> None: + assert classify_load_zone(0.2) is LoadZone.IDLE + + def test_third_is_recreation_inclusive(self) -> None: + assert classify_load_zone(LOWER_BOUND) is LoadZone.RECREATION + + def test_work_zone(self) -> None: + assert classify_load_zone(0.44) is LoadZone.WORK + + def test_upper_bound_is_recreation_top(self) -> None: + # 1/φ² is the inclusive TOP of RECREATION; WORK starts just above. + assert classify_load_zone(UPPER_BOUND) is LoadZone.RECREATION + assert classify_load_zone(0.39) is LoadZone.WORK + + def test_peaking(self) -> None: + assert classify_load_zone(0.55) is LoadZone.PEAKING + + def test_warning(self) -> None: + assert classify_load_zone(0.64) is LoadZone.WARNING + + def test_danger_above_two_thirds(self) -> None: + assert classify_load_zone(0.8) is LoadZone.DANGER + + +class TestClassifyCyclePhase: + def test_pivot_at_half(self) -> None: + assert classify_cycle_phase(0.5) is CyclePhase.PIVOT + + def test_ascending_below(self) -> None: + assert classify_cycle_phase(0.42) is CyclePhase.ASCENDING + + def test_past_pivot_above(self) -> None: + assert classify_cycle_phase(0.58) is CyclePhase.PAST_PIVOT + + +class TestBandProfileValidation: + def test_default_is_valid(self) -> None: + BandProfile() # no raise + + def test_ordering_violation_rejected(self) -> None: + with pytest.raises(BandProfileInvalid) as exc: + BandProfile(idle_ceiling=0.6, recreation_ceiling=0.5) + assert exc.value.rule == "R1" + + def test_phi_anchor_violation_rejected(self) -> None: + with pytest.raises(BandProfileInvalid): + BandProfile(recreation_ceiling=0.30) # too far from 1/φ² + + def test_resistance_tighten_only(self) -> None: + # widening the resistance band to escape challenge is rejected (R5). + with pytest.raises(BandProfileInvalid) as exc: + BandProfile( + idle_ceiling=0.30, # < 1/3 = looser + quantity=Quantity.RESISTANCE, + ) + assert exc.value.rule in ("R1", "R5") + + +class TestSetpointFor: + def test_resistance_setpoint(self) -> None: + low, high = setpoint_for(Quantity.RESISTANCE, BandProfile()) + assert low == pytest.approx(LOWER_BOUND) + assert high == pytest.approx(UPPER_BOUND) + + def test_work_setpoint(self) -> None: + low, high = setpoint_for(Quantity.WORK, BandProfile()) + assert low == pytest.approx(UPPER_BOUND) + assert high == pytest.approx(WORK_ZONE_UPPER) + + def test_growth_rejected(self) -> None: + with pytest.raises(GrowthNotADecisionBand): + setpoint_for(Quantity.GROWTH, BandProfile()) + + +class TestZoneToLegacy: + def test_projection(self) -> None: + assert zone_to_legacy(LoadZone.IDLE) is ZoneClassification.UNDER_LOADED + assert zone_to_legacy(LoadZone.RECREATION) is ZoneClassification.CALIBRATION + assert zone_to_legacy(LoadZone.WORK) is ZoneClassification.WORKING + assert zone_to_legacy(LoadZone.PEAKING) is ZoneClassification.WORKING + assert zone_to_legacy(LoadZone.WARNING) is ZoneClassification.PEAKING + assert zone_to_legacy(LoadZone.DANGER) is ZoneClassification.DEBT diff --git a/python/tests/test_executive_negentropy.py b/python/tests/test_executive_negentropy.py new file mode 100644 index 0000000..96c8048 --- /dev/null +++ b/python/tests/test_executive_negentropy.py @@ -0,0 +1,52 @@ +"""Conformance tests for the negentropy / order metric.""" +from __future__ import annotations + +import pytest + +from substrate.executive.negentropy import ( + NegentropyDirection, + negentropy, + order_index, +) + + +class TestOrderIndex: + def test_single_category_is_maximal_order(self) -> None: + assert order_index([10, 0, 0]) == 1.0 + + def test_uniform_is_maximal_disorder(self) -> None: + assert order_index([5, 5, 5]) == pytest.approx(0.0) + + def test_concentrated_beats_spread(self) -> None: + assert order_index([8, 1, 1]) > order_index([4, 3, 3]) + + def test_zero_counts_ignored(self) -> None: + assert order_index([5, 5, 0, 0]) == order_index([5, 5]) + + def test_empty_rejected(self) -> None: + with pytest.raises(ValueError, match="positive count"): + order_index([0, 0]) + + def test_negative_rejected(self) -> None: + with pytest.raises(ValueError, match="counts"): + order_index([5, -1]) + + +class TestNegentropy: + def test_empty_rejected(self) -> None: + with pytest.raises(ValueError, match="non-empty"): + negentropy([]) + + def test_single_reading_is_stable(self) -> None: + assert negentropy([0.5]).direction is NegentropyDirection.STABLE + + def test_rising_order_is_emerging(self) -> None: + report = negentropy([0.1, 0.2, 0.4, 0.6, 0.8]) + assert report.direction is NegentropyDirection.EMERGING + assert report.order_delta > 0 + + def test_falling_order_is_decaying(self) -> None: + assert negentropy([0.9, 0.7, 0.5, 0.3, 0.1]).direction is NegentropyDirection.DECAYING + + def test_flat_within_deadband_is_stable(self) -> None: + assert negentropy([0.50, 0.51, 0.49, 0.50, 0.51]).direction is NegentropyDirection.STABLE diff --git a/python/tests/test_executive_observed_graph.py b/python/tests/test_executive_observed_graph.py new file mode 100644 index 0000000..d4cac98 --- /dev/null +++ b/python/tests/test_executive_observed_graph.py @@ -0,0 +1,90 @@ +"""Tests for the NPG calculus on the observed entity graph (WS-8).""" +from __future__ import annotations + +import pytest + +from substrate.executive.observed_graph import ( + NpgEdge, + detect_extraction, +) +from substrate.executive.quantities import Cycle + + +class TestNpgEdgeValidation: + def test_empty_source_rejected(self) -> None: + with pytest.raises(ValueError, match="source_entity_id"): + NpgEdge("", "t", -0.5, Cycle.SHORT) + + def test_empty_target_rejected(self) -> None: + with pytest.raises(ValueError, match="target_entity_id"): + NpgEdge("s", "", -0.5, Cycle.SHORT) + + +class TestDetectExtraction: + def test_empty_graph(self) -> None: + report = detect_extraction([]) + assert not report.rollups + assert not report.extractive + assert not report.supportive + + def test_predator_flagged_extractive(self) -> None: + edges = [ + NpgEdge("pred", "v1", -0.6, Cycle.SHORT), + NpgEdge("pred", "v2", -0.4, Cycle.SHORT), + ] + report = detect_extraction(edges) + assert len(report.extractive) == 1 + pred = report.extractive[0] + assert pred.entity_id == "pred" + assert pred.short_cycle_taken == pytest.approx(1.0) + assert pred.long_cycle_given == 0.0 + assert pred.is_extractive is True + assert pred.is_supportive is False + + def test_mentor_flagged_supportive(self) -> None: + edges = [ + NpgEdge("mentor", "student", 0.8, Cycle.LONG), + NpgEdge("mentor", "x", -0.1, Cycle.SHORT), + ] + report = detect_extraction(edges) + assert len(report.supportive) == 1 + mentor = report.supportive[0] + assert mentor.long_cycle_given == pytest.approx(0.8) + assert mentor.is_supportive is True + assert mentor.is_extractive is False + + def test_short_cycle_giving_is_not_long_support(self) -> None: + # A short-cycle gift does not count as sustained long-cycle support. + edges = [NpgEdge("a", "b", 0.5, Cycle.SHORT)] + rollup = detect_extraction(edges).rollups[0] + assert rollup.long_cycle_given == 0.0 + assert rollup.short_cycle_taken == 0.0 # it's a gift, not a taking + + def test_long_cycle_taking_not_counted_as_short_extraction(self) -> None: + # Sustained correction (long-cycle negative) is not one-off extraction. + edges = [NpgEdge("a", "b", -0.5, Cycle.LONG)] + rollup = detect_extraction(edges).rollups[0] + assert rollup.short_cycle_taken == 0.0 + + def test_ranking_worst_extractor_first(self) -> None: + edges = [ + NpgEdge("mild", "v", -0.2, Cycle.SHORT), + NpgEdge("severe", "v", -0.9, Cycle.SHORT), + ] + report = detect_extraction(edges) + assert report.rollups[0].entity_id == "severe" + assert report.rollups[1].entity_id == "mild" + + def test_threshold_gates_extractive(self) -> None: + edges = [NpgEdge("a", "b", -0.3, Cycle.SHORT)] + assert not detect_extraction(edges, extraction_threshold=0.5).extractive + assert len(detect_extraction(edges, extraction_threshold=0.0).extractive) == 1 + + def test_net_potential_caused_is_signed_sum(self) -> None: + edges = [ + NpgEdge("a", "b", 0.5, Cycle.LONG), + NpgEdge("a", "c", -0.2, Cycle.SHORT), + ] + rollup = detect_extraction(edges).rollups[0] + assert rollup.net_potential_caused == pytest.approx(0.3) + assert rollup.edge_count == 2