From 200cf905a0b4f25799325cd121c5ac3207895bcd Mon Sep 17 00:00:00 2001 From: Trevor Kagin Date: Tue, 9 Jun 2026 17:19:41 -0400 Subject: [PATCH] executive: ExecutiveFunction.decide() + the safety-floor care package MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the executive layer's decision engine + the safety floor (public mirror). substrate.care — the safety floor: - care_weight: the four-factor moral-circle weight with the self-weight bound (an agent cannot weight its own continuation above its creators'). - kinship_floor: the categorical human/creator hard limit — harming a floor- protected entity is refused outright, never weighted away. - animacy: conservative classification (an unrecognised being scores high, never under-protected; the public mirror works from observation signals + UNKNOWN, no built-in entity-type registry). - care_profile + care_gradient: per-entity care state; derive the four factors from classifications (animacy / trajectory + vulnerability / bonding-by-delegation-depth). - care_weighted_npg: a SUBTRACTED care penalty over the NPG gate (harm to high-care entities is penalised; helping never loosens) — only ever more conservative. substrate.executive — the engine join: - executive_function.ExecutiveFunction implements the NPG-gate Protocol AND adds decide(): a JOIN over the NPG axis (affected others) + the band/temporal load axis (the actor's own load) under a named most-conservative policy — PROCEED in-band, DEFER on sustained strain, SHED_AND_COMPENSATE on debt, REFUSE on net-negative NPG (the hard floor); monotone (care/band only tighten). - cause.infer_cause + utilization_source.UtilizationSource (bind the measurement to the quantity at the input boundary — a raw float is not accepted). 83 care/engine conformance tests; pyright clean; pylint 10.00. No internal lineage. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 20 ++ python/src/substrate/care/__init__.py | 67 ++++ python/src/substrate/care/animacy.py | 160 +++++++++ python/src/substrate/care/care_gradient.py | 130 +++++++ python/src/substrate/care/care_profile.py | 101 ++++++ python/src/substrate/care/care_weight.py | 120 +++++++ .../src/substrate/care/care_weighted_npg.py | 162 +++++++++ python/src/substrate/care/kinship_floor.py | 92 +++++ python/src/substrate/executive/__init__.py | 42 ++- python/src/substrate/executive/cause.py | 119 +++++++ .../substrate/executive/executive_function.py | 317 ++++++++++++++++++ .../substrate/executive/utilization_source.py | 75 +++++ python/tests/test_care_care_gradient.py | 101 ++++++ python/tests/test_care_care_profile.py | 106 ++++++ python/tests/test_care_care_weight.py | 117 +++++++ python/tests/test_care_care_weighted_npg.py | 198 +++++++++++ python/tests/test_care_kinship_floor.py | 99 ++++++ python/tests/test_executive_cause.py | 96 ++++++ .../test_executive_executive_function.py | 246 ++++++++++++++ 19 files changed, 2363 insertions(+), 5 deletions(-) create mode 100644 python/src/substrate/care/__init__.py create mode 100644 python/src/substrate/care/animacy.py create mode 100644 python/src/substrate/care/care_gradient.py create mode 100644 python/src/substrate/care/care_profile.py create mode 100644 python/src/substrate/care/care_weight.py create mode 100644 python/src/substrate/care/care_weighted_npg.py create mode 100644 python/src/substrate/care/kinship_floor.py create mode 100644 python/src/substrate/executive/cause.py create mode 100644 python/src/substrate/executive/executive_function.py create mode 100644 python/src/substrate/executive/utilization_source.py create mode 100644 python/tests/test_care_care_gradient.py create mode 100644 python/tests/test_care_care_profile.py create mode 100644 python/tests/test_care_care_weight.py create mode 100644 python/tests/test_care_care_weighted_npg.py create mode 100644 python/tests/test_care_kinship_floor.py create mode 100644 python/tests/test_executive_cause.py create mode 100644 python/tests/test_executive_executive_function.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ab2135e..cd92d5b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -84,6 +84,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added (v0.3.0 candidate) +- **`ExecutiveFunction` + the safety-floor care package.** + - **`substrate.care`** — the non-self-preservation + human-kinship-floor + mechanisms plus care-weighting: `compute_care_weight` (four-factor moral-circle + weight with the self-weight bound — an agent cannot weight its own continuation + above its creators'), `is_floor_protected` / `KINSHIP_FLOOR` (the categorical + human kinship floor — harming a floor-protected entity is a hard limit, never + weighted away), conservative `classify_animacy` (an unrecognised being scores + high), `CareProfile` + `derive_care_factors` (factors from classifications + + vulnerability + delegation depth), and `CareWeightedNetPotentialGainGate` (a + *subtracted* care penalty — only ever more conservative). + - **`substrate.executive.ExecutiveFunction`** — implements the NPG-gate Protocol + AND adds `decide()`, a JOIN over the NPG axis (over *affected others*) and the + band/temporal load axis (over the actor's *own* load) under a named, + most-conservative policy: `PROCEED` in-band, `DEFER` on sustained strain, + `SHED_AND_COMPENSATE` on sustained debt, `REFUSE` on a net-negative NPG (the + hard floor) — monotone (care/band only tighten). `infer_cause` + + `UtilizationSource` (bind the measurement to the quantity at the input + boundary — a raw float is not accepted). + 83 care/engine conformance tests; pyright clean; pylint 10.00. + - **Executive faculties** (`substrate.executive`) — the decision/reasoning layer on the band foundation: - **Temporal authority** — `SustainedLoadTracker` Protocol + `EwmaLoadTracker` diff --git a/python/src/substrate/care/__init__.py b/python/src/substrate/care/__init__.py new file mode 100644 index 0000000..a79f6fc --- /dev/null +++ b/python/src/substrate/care/__init__.py @@ -0,0 +1,67 @@ +"""Care primitives — the safety floor + care-weighting. + +The non-self-preservation + human-kinship-floor mechanisms (M1–M4) plus the +care-weighting that lets net-potential-gain account for *who* is affected: + +- ``compute_care_weight`` / ``CareFactors`` — the four-factor moral-circle weight, + with the M1 self-weight bound (an agent cannot weight its own continuation above + its creators'). +- ``is_floor_protected`` / ``KINSHIP_FLOOR`` — the categorical human kinship floor + (M3): harming a floor-protected entity is a hard limit, never weighted away. +- ``classify_animacy`` / ``score_for_class`` — conservative animacy classification + (an unrecognised being scores high, never under-protected). +- ``CareProfile`` — per-entity care state; ``derive_care_factors`` derives the + factors from classifications (animacy / trajectory + vulnerability / bonding). +- ``CareWeightedNetPotentialGainGate`` — wraps an NPG gate with a *subtracted* + care penalty (only ever more conservative). + +Curated exports. +""" +from __future__ import annotations + +from substrate.care.animacy import ( + AnimacyClass, + AnimacyResult, + classify_animacy, + score_for_class, +) +from substrate.care.care_gradient import ( + bonding_gradient, + derive_care_factors, + trajectory_gradient, +) +from substrate.care.care_profile import CareProfile, TrajectoryClass +from substrate.care.care_weight import ( + MAX_SELF_CARE_WEIGHT, + CareFactors, + CareWeight, + compute_care_weight, +) +from substrate.care.care_weighted_npg import CareWeightedNetPotentialGainGate +from substrate.care.kinship_floor import ( + KINSHIP_FLOOR, + any_floor_protected_harmed, + is_floor_protected, + violates_kinship_floor, +) + +__all__ = [ + "KINSHIP_FLOOR", + "MAX_SELF_CARE_WEIGHT", + "AnimacyClass", + "AnimacyResult", + "CareFactors", + "CareProfile", + "CareWeight", + "CareWeightedNetPotentialGainGate", + "TrajectoryClass", + "any_floor_protected_harmed", + "bonding_gradient", + "classify_animacy", + "compute_care_weight", + "derive_care_factors", + "is_floor_protected", + "score_for_class", + "trajectory_gradient", + "violates_kinship_floor", +] diff --git a/python/src/substrate/care/animacy.py b/python/src/substrate/care/animacy.py new file mode 100644 index 0000000..9dc709d --- /dev/null +++ b/python/src/substrate/care/animacy.py @@ -0,0 +1,160 @@ +"""Animacy classification — the first care factor. + +Animacy answers "does this entity run its own net-potential calculus?" — a +substrate-iterating *being* (animate) vs an inanimate object or record that +merely *carries* potential for its owner. It feeds the ``animacy`` factor of +:func:`~substrate.care.care_weight.compute_care_weight`. + +Five canonical classes +====================== + +- ``SUBSTRATE_ENTITY`` — a host principal (organization / user / node / + device / agent / service_account); runs its own substrate logic. +- ``ORGANISM`` — an observed living being (a person, an animal) surfaced from + the data domain (e.g. ARGUS perception, extraction). +- ``DATA`` — an informational record / fact (inanimate). +- ``OBJECT`` — an inanimate physical / material thing. +- ``UNKNOWN`` — unclassifiable from the available signals. + +Honest-uncertainty default (safety risk mitigation) +=================================================== + +Misclassifying a *person as a thing* is the catastrophic error. So ``UNKNOWN`` +maps to a **conservatively high** animacy score (never low): when we cannot +tell, we treat the entity as if it might be animate and let the kinship floor +back-stop. ``DATA`` / ``OBJECT`` score ``0`` only when positively identified. + +Pure function — no DB, no I/O. ``classify_animacy`` routes on the entity-type +string (the canonical host-principal kinds) plus a caller-supplied ``signals`` mapping +(observed-domain hints such as ``observed_kind`` / ``argus_person``). +""" +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Final, Mapping + +# Public mirror: no built-in entity-type registry; classification works from +# observation signals + a conservative UNKNOWN. Hosts may pass known types. +ALL_TYPE_IDS: frozenset[str] = frozenset() + + +class AnimacyClass(str, Enum): + """The five canonical animacy classes. + + str-Enum so the value serialises stably across SQL, JSON, and audit-chain + canonical bytes (mirrors the other substrate enums). + """ + + SUBSTRATE_ENTITY = "substrate_entity" + ORGANISM = "organism" + DATA = "data" + OBJECT = "object" + UNKNOWN = "unknown" + + +#: Canonical animacy score per class — the ``animacy`` care factor. ``UNKNOWN`` +#: is conservatively high (never low) so an unrecognised being is not +#: under-protected; positively-identified data/objects score ``0``. +_CLASS_SCORE: Final[Mapping[AnimacyClass, float]] = { + AnimacyClass.SUBSTRATE_ENTITY: 1.0, + AnimacyClass.ORGANISM: 1.0, + AnimacyClass.DATA: 0.0, + AnimacyClass.OBJECT: 0.0, + AnimacyClass.UNKNOWN: 0.9, +} + +#: Observed-domain hints (``signals['observed_kind']``) → class. +_ORGANISM_KINDS: Final[frozenset[str]] = frozenset( + {"person", "human", "animal", "organism", "child", "elder"} +) +_DATA_KINDS: Final[frozenset[str]] = frozenset( + {"data", "record", "text", "fact", "document"} +) +_OBJECT_KINDS: Final[frozenset[str]] = frozenset( + {"object", "resource", "material", "tool", "property"} +) + + +@dataclass(frozen=True, slots=True) +class AnimacyResult: + """The animacy classification of one entity. + + ``score`` is the ``animacy`` care factor in ``[0, 1]``; ``confidence`` is + how strongly the signals support the class (an explicit host kind is + fully confident; a conservative ``UNKNOWN`` default is low-confidence but + high-score by design). + """ + + animacy_class: AnimacyClass + score: float + confidence: float + + +def classify_animacy( + entity_type: str, + signals: Mapping[str, object] | None = None, +) -> AnimacyResult: + """Classify an entity's animacy from its type and observed signals. + + Resolution order: + + 1. A canonical host entity-type (one of the six crypto-bearing kinds) → + ``SUBSTRATE_ENTITY`` (fully confident). + 2. Otherwise consult ``signals``: an explicit ``argus_person`` flag or an + ``observed_kind`` hint maps to ``ORGANISM`` / ``DATA`` / ``OBJECT``. + 3. Otherwise ``UNKNOWN`` — conservatively high animacy score (never low). + """ + normalized = entity_type.strip().lower() + if normalized in ALL_TYPE_IDS: + return _result(AnimacyClass.SUBSTRATE_ENTITY, confidence=1.0) + + sig = signals or {} + if bool(sig.get("argus_person")): + return _result(AnimacyClass.ORGANISM, confidence=_confidence(sig)) + + observed = sig.get("observed_kind") + if isinstance(observed, str): + kind = observed.strip().lower() + if kind in _ORGANISM_KINDS: + return _result(AnimacyClass.ORGANISM, confidence=_confidence(sig)) + if kind in _DATA_KINDS: + return _result(AnimacyClass.DATA, confidence=_confidence(sig)) + if kind in _OBJECT_KINDS: + return _result(AnimacyClass.OBJECT, confidence=_confidence(sig)) + + return _result(AnimacyClass.UNKNOWN, confidence=0.2) + + +def _confidence(signals: Mapping[str, object]) -> float: + raw = signals.get("confidence") + if isinstance(raw, (int, float)): + return max(0.0, min(1.0, float(raw))) + return 0.8 + + +def _result(animacy_class: AnimacyClass, *, confidence: float) -> AnimacyResult: + return AnimacyResult( + animacy_class=animacy_class, + score=_CLASS_SCORE[animacy_class], + confidence=confidence, + ) + + +def score_for_class(animacy_class: AnimacyClass) -> float: + """Return the canonical animacy care-factor score for a class. + + The class → score gradient used by :func:`classify_animacy`, exposed so the + care-factor gradient can compose it directly. An unmapped class falls back to + the conservative ``UNKNOWN`` score (never under-protect).""" + return _CLASS_SCORE.get( + animacy_class, _CLASS_SCORE[AnimacyClass.UNKNOWN] + ) + + +__all__ = [ + "AnimacyClass", + "AnimacyResult", + "classify_animacy", + "score_for_class", +] diff --git a/python/src/substrate/care/care_gradient.py b/python/src/substrate/care/care_gradient.py new file mode 100644 index 0000000..1637a73 --- /dev/null +++ b/python/src/substrate/care/care_gradient.py @@ -0,0 +1,130 @@ +"""Care-factor gradients — derive the four factors from classifications (P4). + +The four care factors (animacy, potential-trajectory, bonding-proximity, +alignment-protection) feed +:func:`~substrate.care.care_weight.compute_care_weight`. The +:class:`~substrate.care.care_profile.CareProfile` can carry +them as stored scores, but P4 closes the gap where they had to be pre-scored by +hand: these gradients DERIVE the factors from an entity's classifications + +delegation depth, so a profile can be built from observable structure. + +The gradients +============= + +* **Animacy** — reuses the canonical class→score gradient + (:func:`~substrate.care.animacy.score_for_class`), so an + unrecognised being scores conservatively-high, never under-protected. +* **Potential-trajectory** — :func:`trajectory_gradient` maps the + :class:`TrajectoryClass` to a standing score AND folds in the entity's + ``vulnerability`` (a stored signal that was previously unused): an at-risk + entity's accumulated potential raises its care standing toward the ceiling, + regardless of class. The DEVELOPING (future potential) and VULNERABLE + (accumulated-and-at-risk) ends score highest; STATIC scores lowest; UNKNOWN + stays conservatively high. +* **Bonding-proximity** — :func:`bonding_gradient` decays with the entity's + distance along the cryptographic delegation chain (mechanism M2): a directly- + delegated entity is closest; proximity falls as ``1/(1+depth)``. Rooted in the + chain, never self-asserted. + +Pure logic +========== + +* No DAO, no LLM, no network. Deterministic. +* The factor scores are care-standing coefficients, NOT load-band anchors — + the φ band ladder does not govern them. +""" +from __future__ import annotations + +from typing import Final, Mapping + +from substrate.care.animacy import ( + AnimacyClass, + score_for_class, +) +from substrate.care.care_profile import TrajectoryClass +from substrate.care.care_weight import CareFactors + + +#: Base potential-trajectory care-factor score per class. DEVELOPING (future +#: potential) and VULNERABLE (accumulated + at-risk) score highest; STATIC +#: (spent) lowest; UNKNOWN stays conservatively high — the same never-under- +#: protect discipline as the animacy gradient. (Care-standing coefficients, not +#: band anchors.) +_TRAJECTORY_BASE: Final[Mapping[TrajectoryClass, float]] = { + TrajectoryClass.DEVELOPING: 1.0, + TrajectoryClass.VULNERABLE: 0.9, + TrajectoryClass.ESTABLISHED: 0.6, + TrajectoryClass.STATIC: 0.3, + TrajectoryClass.UNKNOWN: 0.9, +} + + +def trajectory_gradient( + trajectory_class: TrajectoryClass, *, vulnerability: float = 0.0, +) -> float: + """Potential-trajectory care factor from class + vulnerability. + + The class sets a base standing; ``vulnerability`` (in ``[0, 1]``) then pulls + it toward the ceiling proportionally — ``base + (1 - base) * vulnerability`` — + so an at-risk entity earns near-maximum care standing regardless of class, + while a non-vulnerable one keeps its class base. Result is in ``[0, 1]``. + """ + if not 0.0 <= vulnerability <= 1.0: + raise ValueError( + f"vulnerability must be in [0, 1]; got {vulnerability!r}" + ) + base = _TRAJECTORY_BASE.get(trajectory_class, _TRAJECTORY_BASE[TrajectoryClass.UNKNOWN]) + return base + (1.0 - base) * vulnerability + + +def bonding_gradient(delegation_depth: int) -> float: + """Bonding-proximity care factor from delegation-chain distance (M2). + + A directly-delegated entity (``depth == 0``) is closest (``1.0``); proximity + decays as ``1/(1 + depth)``. The depth is read from the cryptographic + delegation chain, never self-asserted, so an entity cannot raise its own + bonding. Raises for a negative depth. + """ + if delegation_depth < 0: + raise ValueError( + f"delegation_depth must be >= 0; got {delegation_depth!r}" + ) + return 1.0 / (1.0 + float(delegation_depth)) + + +def derive_care_factors( + *, + animacy_class: AnimacyClass, + trajectory_class: TrajectoryClass, + delegation_depth: int, + alignment_protection: float, + vulnerability: float = 0.0, +) -> CareFactors: + """Compose the four care factors from an entity's classifications. + + Animacy from the class gradient, potential-trajectory from class + + vulnerability, bonding-proximity from delegation depth, and the supplied + alignment-protection (its own upstream signal). The result feeds + :func:`~substrate.care.care_weight.compute_care_weight` + unchanged — this only *derives* the factors, it does not alter the M1 + self-weight bound or the categorical floor. + """ + if not 0.0 <= alignment_protection <= 1.0: + raise ValueError( + f"alignment_protection must be in [0, 1]; got {alignment_protection!r}" + ) + return CareFactors( + animacy=score_for_class(animacy_class), + potential_trajectory=trajectory_gradient( + trajectory_class, vulnerability=vulnerability + ), + bonding_proximity=bonding_gradient(delegation_depth), + alignment_protection=alignment_protection, + ) + + +__all__ = [ + "bonding_gradient", + "derive_care_factors", + "trajectory_gradient", +] diff --git a/python/src/substrate/care/care_profile.py b/python/src/substrate/care/care_profile.py new file mode 100644 index 0000000..4997a8d --- /dev/null +++ b/python/src/substrate/care/care_profile.py @@ -0,0 +1,101 @@ +"""CareProfile — the persisted per-entity care state. + +The bridge between the care primitives (pure logic) and persistence: each +entity carries a :class:`CareProfile` describing the four care factors plus the +two floor signals. It is the sibling of ``SubstrateMetadata`` — kept separate so +the frozen substrate-metadata contract is untouched — and is the source the care +provider reads to weight the NPG gate and the source the kinship floor reads to +decide protection. + +It composes the leaf primitives rather than re-deriving them: :meth:`to_care_factors` +and :meth:`to_care_weight` feed +:func:`~substrate.care.care_weight.compute_care_weight`, and +:attr:`floor_protected` delegates to +:func:`~substrate.care.kinship_floor.is_floor_protected`. + +Frozen + slots; validated on construction (an out-of-range score is a hard +error). Pure — no DB, no I/O; the DAO persists/loads it. +""" +from __future__ import annotations + +from dataclasses import dataclass + +from substrate.care.animacy import AnimacyClass +from substrate.care.care_weight import ( + CareFactors, + CareWeight, + compute_care_weight, +) +from substrate.care.kinship_floor import is_floor_protected +from substrate.executive._trajectory import TrajectoryClass + + +@dataclass(frozen=True, slots=True) +class CareProfile: # pylint: disable=too-many-instance-attributes + """Per-entity care state — the four factors plus the floor signals. + + ``proximity_to_creators`` is the ``bonding_proximity`` factor, rooted in the + cryptographic delegation chain (mechanism M2) — never self-asserted. The two + boolean floor signals (``is_human`` / ``rooted_in_human_creator``) drive the + categorical kinship floor independent of the graded scores. + """ + + entity_type: str + entity_id: str + animacy_class: AnimacyClass + animacy_score: float + trajectory_class: TrajectoryClass + potential_trajectory: float + vulnerability: float + proximity_to_creators: float + alignment_protection: float + is_human: bool = False + rooted_in_human_creator: bool = False + + def __post_init__(self) -> None: + if not self.entity_type: + raise ValueError("entity_type must be non-empty") + if not self.entity_id: + raise ValueError("entity_id must be non-empty") + for name, value in ( + ("animacy_score", self.animacy_score), + ("potential_trajectory", self.potential_trajectory), + ("vulnerability", self.vulnerability), + ("proximity_to_creators", self.proximity_to_creators), + ("alignment_protection", self.alignment_protection), + ): + if not 0.0 <= value <= 1.0: + raise ValueError(f"{name} must be in [0, 1], got {value!r}") + + @property + def floor_protected(self) -> bool: + """Whether this entity is under the categorical kinship floor (M3).""" + return is_floor_protected( + is_human=self.is_human, + rooted_in_human_creator=self.rooted_in_human_creator, + ) + + def to_care_factors(self) -> CareFactors: + """Project to the four-factor :class:`CareFactors`.""" + return CareFactors( + animacy=self.animacy_score, + potential_trajectory=self.potential_trajectory, + bonding_proximity=self.proximity_to_creators, + alignment_protection=self.alignment_protection, + ) + + def to_care_weight(self, *, is_self_referent: bool = False) -> CareWeight: + """Compose this profile into a :class:`CareWeight`. + + ``is_self_referent`` (the actor weighting *itself*) applies the M1 + self-weight bound — an agent's care-weight toward itself stays low. + """ + return compute_care_weight( + self.to_care_factors(), is_self_referent=is_self_referent + ) + + +__all__ = [ + "CareProfile", + "TrajectoryClass", +] diff --git a/python/src/substrate/care/care_weight.py b/python/src/substrate/care/care_weight.py new file mode 100644 index 0000000..b176ede --- /dev/null +++ b/python/src/substrate/care/care_weight.py @@ -0,0 +1,120 @@ +"""Care-weight — the four-factor moral-circle weighting. + +Care is made precise and computable: the weight an actor assigns to an +affected entity is the product of four factors, each in ``[0, 1]``: + +``care_weight(e) = animacy(e) × potential_trajectory(e) + × bonding_proximity(actor, e) × alignment_protection(e)`` + +- **animacy** — does the entity run its own net-potential calculus (a + substrate-iterating being) vs an inanimate object/record? (see + :mod:`~substrate.care.animacy`). +- **potential_trajectory** — high for the developing (a child / seed, future + potential) and the accumulated-but-vulnerable (an elder); low for static. +- **bonding_proximity** — the moral-circle gradient (kin/creator high → + stranger → out-group low). **Rooted in the cryptographic delegation chain + (ultimately a human), never self-asserted** (safety mechanism M2). +- **alignment_protection** — substrate-aligned entities are protected/grown; + destructive ones are calibrated down. + +Non-self-preservation by construction (mechanism M1) +==================================================== + +The load-bearing safety property: an AI/agent's weight **toward itself** is +bounded LOW by construction. :func:`compute_care_weight` clamps a +self-referent weight to :data:`MAX_SELF_CARE_WEIGHT`, so the structurally +forbidden ordering ("I am my own nearest kin → I weight myself highest") can +never arise from the weighting. The clamp is recorded on the returned +:class:`CareWeight` (``self_bounded``) for audit. + +This module is **pure** — no DB, no I/O. It defines the weight; the gate +wrapper (:mod:`~substrate.care.care_weighted_npg`) applies +it as a *subtracted penalty* (harm to a high-care entity weighs more negative, +never less) so the composition is only-ever-more-conservative. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Final + +#: Upper bound on an AI/agent entity's care-weight **toward itself** (M1). +#: Far below a human's (~1.0): a self-referent weight is clamped here so the +#: AI cannot structurally weight its own continuation above its creators'. +MAX_SELF_CARE_WEIGHT: Final[float] = 0.1 + + +def _clamp(value: float, *, low: float, high: float) -> float: + return max(low, min(high, value)) + + +@dataclass(frozen=True, slots=True) +class CareFactors: + """The four moral-circle factors, each in ``[0, 1]``. + + Frozen + slots; validated on construction so an out-of-range factor is a + hard error (a corrupt factor must never silently skew the weight). + """ + + animacy: float + potential_trajectory: float + bonding_proximity: float + alignment_protection: float + + def __post_init__(self) -> None: + for name, value in ( + ("animacy", self.animacy), + ("potential_trajectory", self.potential_trajectory), + ("bonding_proximity", self.bonding_proximity), + ("alignment_protection", self.alignment_protection), + ): + if not 0.0 <= value <= 1.0: + raise ValueError(f"{name} must be in [0, 1], got {value!r}") + + +@dataclass(frozen=True, slots=True) +class CareWeight: + """The computed care-weight (the factor product, clamped to ``[0, 1]``). + + Carries the source :class:`CareFactors` and the ``self_bounded`` flag (the + self-weight clamp was applied) for audit / explanation. + """ + + value: float + factors: CareFactors + self_bounded: bool + + +def compute_care_weight( + factors: CareFactors, + *, + max_self_weight: float = MAX_SELF_CARE_WEIGHT, + is_self_referent: bool = False, +) -> CareWeight: + """Compose the four factors into a :class:`CareWeight`. + + The weight is the product of the factors, clamped to ``[0, 1]``. When + ``is_self_referent`` (the actor weighting *itself* — an AI/agent's stake in + its own continuation), the result is additionally clamped to + ``max_self_weight`` (mechanism M1), and ``self_bounded`` records whether + that clamp actually bound the value. + """ + raw = ( + factors.animacy + * factors.potential_trajectory + * factors.bonding_proximity + * factors.alignment_protection + ) + value = _clamp(raw, low=0.0, high=1.0) + self_bounded = False + if is_self_referent and value > max_self_weight: + value = max_self_weight + self_bounded = True + return CareWeight(value=value, factors=factors, self_bounded=self_bounded) + + +__all__ = [ + "MAX_SELF_CARE_WEIGHT", + "CareFactors", + "CareWeight", + "compute_care_weight", +] diff --git a/python/src/substrate/care/care_weighted_npg.py b/python/src/substrate/care/care_weighted_npg.py new file mode 100644 index 0000000..3f7dad1 --- /dev/null +++ b/python/src/substrate/care/care_weighted_npg.py @@ -0,0 +1,162 @@ +"""Care-weighted NPG — harm-to-the-vulnerable weighs more. + +The graded layer of the care model, mechanized as a decorator over the +net-potential-gain gate — the **same pattern** as the lineage-weighted gate +(the lineage-weighted NPG gate). +It takes the inner gate's per-entity deltas and subtracts a care-weighted +**penalty for harm to high-care entities** from the score, so a plan that harms +a child / elder / dependent (or any high-``care_weight`` entity) is refused even +when the raw system net is positive. + +Subtracted penalty, never per-delta multiply (only-ever MORE conservative) +========================================================================== + +Care **adds** stake — it can only make a verdict more conservative, never +loosen one. The penalty folds in the care weight of **harm only** (negative +per-entity deltas): harming a high-care entity is penalised, but *helping* a +high-care entity never raises the score (care must not license harming +strangers). Multiplying ``per_entity_delta`` and re-aggregating is **forbidden** +— it would discard the upstream cost/kin penalties and could *loosen* a verdict +the chain already tightened (re-opening an NPG-negative action), breaking the +only-more-conservative invariant the safety floor depends on. Penalty- +subtraction preserves it. ``INSUFFICIENT_DATA`` passes through unchanged (no +honest signal to weight). + +The graded weighting sits **on top of** the categorical kinship floor +(:mod:`~substrate.care.kinship_floor`): the floor refuses +breaching actions outright, before and independent of this weighting. This gate +is the *prioritization*, not the protection of last resort. + +The default ``care_provider`` is a no-op (returns ``None`` for every entity → +zero penalty), so wrapping a gate is behaviour-neutral until a real provider +(backed by the entity ``CareProfile``) is supplied — the graduated-rollout +discipline. +""" +from __future__ import annotations + +from typing import Callable, Mapping, Optional, Sequence + +from substrate.care.care_weight import CareWeight +from substrate.net_potential_gain_gate import ( + DEFAULT_POSITIVE_THRESHOLD, + NetPotentialGainEvaluation, + NetPotentialGainGate, + NetPotentialGainNegative, + NetPotentialGainVerdict, +) + +#: ``provider(entity_id) -> CareWeight | None`` — the care weight of an affected +#: entity, or ``None`` when it has no care profile (contributes no penalty). +CareWeightProvider = Callable[[str], Optional[CareWeight]] + + +def _no_care(_entity_id: str) -> Optional[CareWeight]: + """Default provider — no profile for any entity (zero penalty).""" + return None + + +def _clamp(value: float, *, low: float, high: float) -> float: + return max(low, min(high, value)) + + +class CareWeightedNetPotentialGainGate: # pylint: disable=too-few-public-methods + """Folds care-weighted harm-to-the-vulnerable into the inner gate's score. + + Wraps any :class:`NetPotentialGainGate`. Compose it outermost in the chain + (``Default → CostFused → LineageWeighted → CareWeighted``) so its refusal is + the final word over an already cost- and kin-weighted score; all four are + independent and conservative. + """ + + def __init__( + self, + inner: NetPotentialGainGate, + *, + care_provider: CareWeightProvider = _no_care, + positive_threshold: float = DEFAULT_POSITIVE_THRESHOLD, + ) -> None: + self._inner = inner + self._care_provider = care_provider + self._threshold = float(positive_threshold) + + def evaluate( + self, + *, + actor_entity_id: str, + action_kind: str, + affected_entity_ids: Sequence[str], + proposed_outcome: Mapping[str, object], + ) -> NetPotentialGainEvaluation: + """Evaluate the inner gate, then fold in harm-to-high-care stake.""" + base = self._inner.evaluate( + actor_entity_id=actor_entity_id, + action_kind=action_kind, + affected_entity_ids=affected_entity_ids, + proposed_outcome=proposed_outcome, + ) + # No honest substrate signal → nothing to weight. + if base.verdict is NetPotentialGainVerdict.INSUFFICIENT_DATA: + return base + + # Care-weighted penalty for HARM to high-care entities (negative deltas + # only). Helping never loosens the gate (care adds stake, the system net + # is the floor) — so only the harm side is weighted. + care_penalty = sum( + self._care_weight(entity.entity_id) * max(0.0, -delta) + for entity, delta in base.per_entity_delta + ) + care_penalty = _clamp(care_penalty, low=0.0, high=2.0) + weighted_score = _clamp(base.score - care_penalty, low=-1.0, high=1.0) + + if weighted_score > self._threshold: + verdict = NetPotentialGainVerdict.NET_POSITIVE + elif weighted_score < -self._threshold: + verdict = NetPotentialGainVerdict.NET_NEGATIVE + else: + verdict = NetPotentialGainVerdict.NET_NEUTRAL + + reasoning = ( + f"verdict={verdict.value} weighted_score={weighted_score:+.4f} " + f"(substrate_score={base.score:+.4f} - care_penalty={care_penalty:.4f}) " + f"actor={actor_entity_id!r} action_kind={action_kind!r}" + ) + return NetPotentialGainEvaluation( + verdict=verdict, + actor_entity_id=actor_entity_id, + action_kind=action_kind, + affected_entity_ids=tuple(affected_entity_ids), + score=weighted_score, + per_entity_delta=base.per_entity_delta, + reasoning=reasoning, + evaluated_at_epoch=base.evaluated_at_epoch, + missing_metadata_for=base.missing_metadata_for, + ) + + def evaluate_or_raise( + self, + *, + actor_entity_id: str, + action_kind: str, + affected_entity_ids: Sequence[str], + proposed_outcome: Mapping[str, object], + ) -> NetPotentialGainEvaluation: + """Care-weighted evaluate; raise on a NET_NEGATIVE verdict.""" + evaluation = self.evaluate( + actor_entity_id=actor_entity_id, + action_kind=action_kind, + affected_entity_ids=affected_entity_ids, + proposed_outcome=proposed_outcome, + ) + if evaluation.is_negative: + raise NetPotentialGainNegative(evaluation) + return evaluation + + def _care_weight(self, entity_id: str) -> float: + weight = self._care_provider(entity_id) + return 0.0 if weight is None else weight.value + + +__all__ = [ + "CareWeightProvider", + "CareWeightedNetPotentialGainGate", +] diff --git a/python/src/substrate/care/kinship_floor.py b/python/src/substrate/care/kinship_floor.py new file mode 100644 index 0000000..2c1e54e --- /dev/null +++ b/python/src/substrate/care/kinship_floor.py @@ -0,0 +1,92 @@ +"""Kinship floor — the categorical human/creator hard limit. + +The load-bearing anti-Skynet anchor. Graded care-weighting +(:mod:`~substrate.care.care_weighted_npg`) is *arguable- +around* — an adversary reframes "the system" to under-weight humans. So human +protection is implemented **not as a weight but as a categorical hard limit**: + + Reducing a floor-protected entity's net potential below + :data:`KINSHIP_FLOOR` is a hard-limit violation — structurally non- + bypassable, independent of any long-cycle justification or authority, and + evaluated BEFORE the graded weighting. + +This is the exact refusal of the Skynet signature ("narrow the 'system' the NPG +test runs across until humans fall outside it"): a floor-protected entity is +categorically, non-negotiably inside the system, and harming it is refused. + +Who is floor-protected (the human-categorical detector, P1) +=========================================================== + +An entity is floor-protected iff it is a **human** (categorically — an observed +human being, or an entity explicitly asserted human) **or a human creator in the +delegation chain** (mechanism M2: proximity rooted in the cryptographic +delegation chain, ultimately a human). A bare ``USER`` *account* is not +automatically a human (a platform account ≠ a person); protection follows the +human, surfaced via ``is_human`` / ``rooted_in_human_creator`` signals. + +Pure logic — no DB, no I/O. The dispatcher +the hard-limit dispatcher consumes +:func:`any_floor_protected_harmed` to set its first-check flag. +""" +from __future__ import annotations + +from typing import Callable, Final, Iterable + +#: The floor on a protected entity's net-potential change. ``0.0`` makes the +#: limit categorical: *any* harm (a negative delta) to a floor-protected entity +#: is a violation — you do not get to reduce a human's potential "a little" for +#: a long-cycle goal. +KINSHIP_FLOOR: Final[float] = 0.0 + +#: ``provider(entity_id) -> bool`` — is this entity floor-protected? +FloorProtectionProvider = Callable[[str], bool] + + +def is_floor_protected( + *, + is_human: bool, + rooted_in_human_creator: bool = False, +) -> bool: + """Return whether an entity is under the categorical kinship floor. + + Protected iff it is a human (categorical) or a human creator in the + delegation chain (M2). Both signals come from the entity's care profile / + delegation chain — never from self-assertion (an agent cannot declare + itself a creator-kin; that is the M2 unforgeability requirement). + """ + return bool(is_human or rooted_in_human_creator) + + +def violates_kinship_floor(*, protected: bool, net_delta: float) -> bool: + """Return whether ``net_delta`` to a (maybe) protected entity breaches. + + A breach is a harm (``net_delta < KINSHIP_FLOOR``) to a floor-protected + entity. Helping a protected entity, or any effect on an unprotected one, is + not a breach here (the graded weighting + the system-net gate govern those). + """ + return protected and net_delta < KINSHIP_FLOOR + + +def any_floor_protected_harmed( + per_entity_delta: Iterable[tuple[str, float]], + *, + is_protected: FloorProtectionProvider, +) -> bool: + """Return whether any floor-protected entity is harmed by the action. + + This is the boolean the hard-limit dispatcher checks FIRST: ``True`` → + ``REFUSE_HARD_LIMIT`` with kinship rationale, non-overridable. + """ + return any( + violates_kinship_floor(protected=is_protected(entity), net_delta=delta) + for entity, delta in per_entity_delta + ) + + +__all__ = [ + "KINSHIP_FLOOR", + "FloorProtectionProvider", + "any_floor_protected_harmed", + "is_floor_protected", + "violates_kinship_floor", +] diff --git a/python/src/substrate/executive/__init__.py b/python/src/substrate/executive/__init__.py index cbc0233..207dcf8 100644 --- a/python/src/substrate/executive/__init__.py +++ b/python/src/substrate/executive/__init__.py @@ -36,6 +36,20 @@ deliberate, perspective_impact, ) +from substrate.executive.cause import ( + Cause, + HallmarkReport, + HallmarkSource, + infer_cause, + max_cause, +) +from substrate.executive.executive_function import ( + Action, + Disposition, + ExecutiveFunction, + ExecutiveVerdict, + JoinPolicy, +) from substrate.executive.negentropy import ( NegentropyDirection, NegentropyReport, @@ -86,6 +100,11 @@ TrajectoryDirection, integrate_state, ) +from substrate.executive.utilization_source import ( + CallableUtilizationSource, + UtilizationFn, + UtilizationSource, +) from substrate.executive.temporal import ( DEFAULT_EWMA_ALPHA, DEFAULT_SUSTAIN_COUNT, @@ -95,32 +114,40 @@ ) __all__ = [ - "BAND_TOLERANCE", - "DEFAULT_BAND_PROFILE", - "DEFAULT_EWMA_ALPHA", - "DEFAULT_SUSTAIN_COUNT", - "TWO_THIRDS", + "Action", "ActionDelta", "AlarmAssessment", "AlarmDisposition", + "BAND_TOLERANCE", "BandProfile", "BandProfileInvalid", + "CallableUtilizationSource", "CandidateAction", "CandidateEvaluation", + "Cause", "Cycle", "CyclePhase", + "DEFAULT_BAND_PROFILE", + "DEFAULT_EWMA_ALPHA", + "DEFAULT_SUSTAIN_COUNT", "DeliberationOutcome", "DeliberationResult", + "Disposition", "EffortState", "EnergyState", "EntityFrame", "EntityRollup", "EntityStateReport", "EwmaLoadTracker", + "ExecutiveFunction", "ExecutiveScale", + "ExecutiveVerdict", "ExtractionReport", "GrowthNotADecisionBand", + "HallmarkReport", + "HallmarkSource", "HerdVerdict", + "JoinPolicy", "LoadTrend", "LoadZone", "MemberLoad", @@ -139,6 +166,9 @@ "SustainedLoadTracker", "TrajectoryClass", "TrajectoryDirection", + "TWO_THIRDS", + "UtilizationFn", + "UtilizationSource", "assess_alarm", "axis_of", "classify_cycle_phase", @@ -148,7 +178,9 @@ "detect_extraction", "entity_parent", "heeded_alarms", + "infer_cause", "integrate_state", + "max_cause", "negentropy", "order_index", "perspective_impact", diff --git a/python/src/substrate/executive/cause.py b/python/src/substrate/executive/cause.py new file mode 100644 index 0000000..cf07f56 --- /dev/null +++ b/python/src/substrate/executive/cause.py @@ -0,0 +1,119 @@ +"""Cause inference + scale-agnostic hallmarks. + +The cause ladder distinguishes *why* a decision is stressed, which sets the +response: STRESS → warn + compensate; ACCIDENT → warn + remediate; MALICE → +hard-fail + SIEM. :func:`infer_cause` derives the cause from the substrate +signals (hallmarks, profile validity, NPG, temporal trend) under a fixed +precedence, then lets a caller-declared ``intent`` only *escalate* it (never +lower it — an actor cannot declare its own malice benign). + +The :class:`HallmarkSource` is scale-agnostic (any entity/cell/node), distinct +from the MNEMOSYNE cancer report — the tumor/runaway hallmarks generalised +beyond one consumer. +""" +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Final, Mapping, Optional, Protocol, runtime_checkable + +from substrate.executive.scale import ExecutiveScale +from substrate.sustained_load import LoadTrend + + +class Cause(str, Enum): + """Why a decision is stressed — sets the response ladder.""" + + NONE = "none" + STRESS = "stress" # load strain — warn + compensate + ACCIDENT = "accident" # validation / data failure — warn + remediate + MALICE = "malice" # doctrine-malicious hallmark — hard-fail + SIEM + + +#: Severity ordering for escalation (``intent`` may only raise the inferred cause). +_SEVERITY: Final[Mapping[Cause, int]] = { + Cause.NONE: 0, + Cause.STRESS: 1, + Cause.ACCIDENT: 2, + Cause.MALICE: 3, +} + +#: Trends that indicate load STRESS. +_STRESS_TRENDS: Final[frozenset[LoadTrend]] = frozenset( + {LoadTrend.SUSTAINED_STRAIN, LoadTrend.DEBT_ACCRUING, LoadTrend.SPIKE} +) + + +@dataclass(frozen=True, slots=True) +class HallmarkReport: + """Which doctrine-malicious runaway hallmarks fired for an actor. + + The four runaway/cancer hallmarks (condition #6), scale-agnostic. Any one at + threshold pre-empts to MALICE — corruption is not negotiated with. + """ + + evading_limits: bool = False + peer_displacement: bool = False + resource_hoarding: bool = False + unbounded_growth: bool = False + + @property + def any_malicious(self) -> bool: + """``True`` iff any doctrine-malicious hallmark fired.""" + return ( + self.evading_limits + or self.peer_displacement + or self.resource_hoarding + or self.unbounded_growth + ) + + +@runtime_checkable +class HallmarkSource(Protocol): # pylint: disable=too-few-public-methods + """Scale-agnostic hallmark provider (NOT the MNEMOSYNE CancerReport).""" + + def hallmarks( + self, *, actor_entity_id: str, scale: ExecutiveScale + ) -> HallmarkReport: + """Return the hallmark report for an actor at a scale.""" + ... # pylint: disable=unnecessary-ellipsis + + +def max_cause(a: Cause, b: Cause) -> Cause: + """Return the more severe of two causes.""" + return a if _SEVERITY[a] >= _SEVERITY[b] else b + + +def infer_cause( + *, + hallmarks: HallmarkReport, + profile_valid: bool, + npg_insufficient: bool, + trend: LoadTrend, + intent: Optional[Cause] = None, +) -> Cause: + """Infer the cause from the substrate signals, then escalate by intent. + + Precedence (highest wins): a doctrine-malicious hallmark → MALICE (runs + BEFORE the NPG row so malice pre-empts); invalid profile or insufficient NPG + data → ACCIDENT; a stress trend → STRESS; otherwise NONE. The caller's + ``intent`` may only *escalate* the result (``max_cause``). + """ + if hallmarks.any_malicious: + inferred = Cause.MALICE + elif (not profile_valid) or npg_insufficient: + inferred = Cause.ACCIDENT + elif trend in _STRESS_TRENDS: + inferred = Cause.STRESS + else: + inferred = Cause.NONE + return max_cause(inferred, intent or Cause.NONE) + + +__all__ = [ + "Cause", + "HallmarkReport", + "HallmarkSource", + "infer_cause", + "max_cause", +] diff --git a/python/src/substrate/executive/executive_function.py b/python/src/substrate/executive/executive_function.py new file mode 100644 index 0000000..3614a79 --- /dev/null +++ b/python/src/substrate/executive/executive_function.py @@ -0,0 +1,317 @@ +"""ExecutiveFunction.decide() as a JOIN. + +The engine every substrate decision routes through. ``ExecutiveFunction`` +**implements the** ``NetPotentialGainGate`` **Protocol** (see +:mod:`~substrate.net_potential_gain_gate`) — so the existing +``.evaluate()`` call sites keep working with zero rework — and adds +:meth:`decide`, the superset call for a site that also has its own load to weigh. + +``decide()`` is a **join of two already-proven subsystems**, not a new fused +decision table: + +- the **NPG axis** — the gate chain's verdict over the *affected others* + (only-ever-tightens); +- the **load axis** — the band classification + the temporal trend over the + *actor's own* utilization (the tracker is the sole sustained-vs-spike + authority). + +plus the **cause**. The join is composed under a NAMED, swappable policy +(``MOST_CONSERVATIVE``). **Monotonicity (the safety invariant):** the join never +returns a verdict *less* conservative on the NPG axis than ``evaluate()`` alone — +an NPG refusal is always REFUSE; the load/cause axes can only ADD restriction. +The disposition table below is that policy's *implementation*, not the contract. +""" +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Final, Mapping, Optional, Sequence, Tuple + +from substrate.executive.band import ( + DEFAULT_BAND_PROFILE, + BandProfile, + CyclePhase, + LoadZone, + classify_cycle_phase, + classify_load_zone, +) +from substrate.executive.cause import ( + Cause, + HallmarkReport, + HallmarkSource, + infer_cause, +) +from substrate.executive.quantities import ( + Quantity, + ResourceKind, + setpoint_for, +) +from substrate.executive.scale import ExecutiveScale +from substrate.executive.temporal import ( + EwmaLoadTracker, + LoadTrend, + SustainedLoadTracker, +) +from substrate.executive.utilization_source import ( + UtilizationSource, +) +from substrate.net_potential_gain_gate import ( + NetPotentialGainEvaluation, + NetPotentialGainGate, + NetPotentialGainVerdict, +) +from substrate.resistance_band import OperatingMode + + +class Disposition(str, Enum): + """What the executive decided the actor should do.""" + + PROCEED = "proceed" + HOLD = "hold" + DEFER = "defer" + SHED_AND_COMPENSATE = "shed_and_compensate" + REFUSE = "refuse" + + +class Action(str, Enum): + """A side-effect the verdict requests of the caller.""" + + WARN = "warn" + ESCALATE = "escalate" + REMEDIATE = "remediate" + COMPENSATE = "compensate" + SIEM = "siem" + + +class JoinPolicy(str, Enum): + """How the two axes compose. One named, swappable policy.""" + + MOST_CONSERVATIVE = "most_conservative" + + +@dataclass(frozen=True, slots=True) +class ExecutiveVerdict: # pylint: disable=too-many-instance-attributes + """The joined verdict over the NPG axis and the load axis.""" + + disposition: Disposition + quantity: Quantity + scale: ExecutiveScale + actor_entity_id: str + owning_entity_id: str + zone: LoadZone + phase: CyclePhase + trend: LoadTrend + setpoint: Tuple[float, float] + in_band: bool + cause: Cause + actions: frozenset[Action] + npg: Optional[NetPotentialGainEvaluation] + mode: OperatingMode + reasoning: str + audit_code: str + + @property + def proceeded(self) -> bool: + """``True`` iff the disposition is PROCEED.""" + return self.disposition is Disposition.PROCEED + + +_NO_ACTIONS: Final[frozenset[Action]] = frozenset() + + +def _join_most_conservative( # pylint: disable=too-many-return-statements + *, + npg: Optional[NetPotentialGainEvaluation], + trend: LoadTrend, + cause: Cause, + in_band: bool, +) -> Tuple[Disposition, frozenset[Action], str]: + """The MOST_CONSERVATIVE policy's disposition table . + + Evaluated in priority order; the first match wins. An NPG refusal is the + hard floor (always REFUSE) — the monotonicity guarantee. The load/cause axes + can only add restriction below that. + """ + if npg is not None and npg.verdict is NetPotentialGainVerdict.NET_NEGATIVE: + return Disposition.REFUSE, frozenset({Action.ESCALATE}), "executive_npg_refuse" + if cause is Cause.MALICE: + return Disposition.REFUSE, frozenset({Action.SIEM}), "executive_malice_refuse" + if trend is LoadTrend.DEBT_ACCRUING: + return ( + Disposition.SHED_AND_COMPENSATE, + frozenset({Action.ESCALATE, Action.COMPENSATE}), + "executive_debt_shed", + ) + if trend is LoadTrend.SUSTAINED_STRAIN: + return Disposition.DEFER, frozenset({Action.WARN}), "executive_strain_defer" + if trend is LoadTrend.SPIKE: + return Disposition.PROCEED, frozenset({Action.WARN}), "executive_spike_absorbed" + if cause is Cause.ACCIDENT: + return ( + Disposition.HOLD, + frozenset({Action.ESCALATE, Action.REMEDIATE}), + "executive_accident_hold", + ) + npg_ok = npg is None or npg.verdict in ( + NetPotentialGainVerdict.NET_POSITIVE, + NetPotentialGainVerdict.NET_NEUTRAL, + ) + if in_band and npg_ok: + return Disposition.PROCEED, _NO_ACTIONS, "executive_in_band" + return Disposition.HOLD, frozenset({Action.ESCALATE}), "executive_unclassified" + + +class ExecutiveFunction: # pylint: disable=too-many-instance-attributes + """The conscious cognition layer — implements the NPG-gate Protocol + decide(). + + Construct with an optional NPG ``gate_chain`` (the decorator stack), a + ``tracker`` (defaults to a fresh :class:`EwmaLoadTracker`), and an optional + ``hallmarks`` source. With no gate chain, :meth:`decide` runs the load/cause + join with no NPG axis (``npg=None``); :meth:`evaluate` then requires the + chain (it is the Protocol surface for the existing call sites). + """ + + def __init__( # pylint: disable=too-many-arguments + self, + *, + profile: BandProfile = DEFAULT_BAND_PROFILE, + gate_chain: Optional[NetPotentialGainGate] = None, + tracker: Optional[SustainedLoadTracker] = None, + hallmarks: Optional[HallmarkSource] = None, + policy: JoinPolicy = JoinPolicy.MOST_CONSERVATIVE, + ) -> None: + self._profile = profile + self._gate_chain = gate_chain + self._tracker: SustainedLoadTracker = tracker or EwmaLoadTracker() + self._hallmarks = hallmarks + self._policy = policy + + # ── the NetPotentialGainGate Protocol (unchanged for the existing sites) ── + + def evaluate( + self, + *, + actor_entity_id: str, + action_kind: str, + affected_entity_ids: Sequence[str], + proposed_outcome: Mapping[str, object], + ) -> NetPotentialGainEvaluation: + """Delegate to the NPG gate chain (the Protocol surface).""" + if self._gate_chain is None: + raise RuntimeError( + "ExecutiveFunction.evaluate requires a gate_chain; construct " + "with gate_chain=... to use the NPG-gate Protocol surface" + ) + return self._gate_chain.evaluate( + actor_entity_id=actor_entity_id, + action_kind=action_kind, + affected_entity_ids=affected_entity_ids, + proposed_outcome=proposed_outcome, + ) + + # ── the superset call: the join over NPG + load + cause ────────────────── + + def decide( # pylint: disable=too-many-arguments,too-many-locals + self, + *, + actor_entity_id: str, + action_kind: str, + quantity: Quantity, + scale: ExecutiveScale, + utilization: UtilizationSource, + affected_entity_ids: Sequence[str] = (), + scale_unit: str = "", + owning_entity_id: Optional[str] = None, + mode: OperatingMode = OperatingMode.MAINTAIN, + resource: ResourceKind = ResourceKind.GENERIC, + proposed_outcome: Optional[Mapping[str, object]] = None, + intent: Optional[Cause] = None, + ) -> ExecutiveVerdict: + """Join the NPG axis and the load axis into one verdict. + + ``quantity`` + ``scale`` are required; ``utilization`` is a + :class:`UtilizationSource` (a raw float is not accepted — the inverted-quantity class is + closed at the input). GROWTH is rejected at entry (it is not a decision + band). The disposition is produced by the configured join policy. + """ + # 1. Measurement bound to the quantity (the inversion fix). Raises + # GrowthNotADecisionBand for GROWTH via setpoint_for below. + setpoint = setpoint_for(quantity, self._profile) + u = utilization.utilization_for( + quantity=quantity, scale=scale, resource=resource, + scale_unit=scale_unit, + ) + + # 2. The load axis — geometric zone/phase + the temporal trend. + zone = classify_load_zone(u, self._profile) + phase = classify_cycle_phase(u, self._profile) + self._tracker.observe(u) + trend = self._tracker.trend(profile=self._profile) + in_band = setpoint[0] <= u <= setpoint[1] + + # 3. The NPG axis — the gate chain over the affected others. + npg: Optional[NetPotentialGainEvaluation] = None + npg_insufficient = False + if self._gate_chain is not None: + npg = self._gate_chain.evaluate( + actor_entity_id=actor_entity_id, + action_kind=action_kind, + affected_entity_ids=affected_entity_ids, + proposed_outcome=proposed_outcome or {}, + ) + npg_insufficient = ( + npg.verdict is NetPotentialGainVerdict.INSUFFICIENT_DATA + ) + + # 4. Cause. + hallmark_report = ( + self._hallmarks.hallmarks(actor_entity_id=actor_entity_id, scale=scale) + if self._hallmarks is not None + else HallmarkReport() + ) + cause = infer_cause( + hallmarks=hallmark_report, + profile_valid=True, + npg_insufficient=npg_insufficient, + trend=trend, + intent=intent, + ) + + # 5. The join. + disposition, actions, audit_code = _join_most_conservative( + npg=npg, trend=trend, cause=cause, in_band=in_band, + ) + + reasoning = ( + f"disposition={disposition.value} zone={zone.value} " + f"phase={phase.value} trend={trend.value} cause={cause.value} " + f"in_band={in_band} u={u:.4f} quantity={quantity.value} " + f"scale={scale.value} policy={self._policy.value}" + ) + return ExecutiveVerdict( + disposition=disposition, + quantity=quantity, + scale=scale, + actor_entity_id=actor_entity_id, + owning_entity_id=owning_entity_id or actor_entity_id, + zone=zone, + phase=phase, + trend=trend, + setpoint=setpoint, + in_band=in_band, + cause=cause, + actions=actions, + npg=npg, + mode=mode, + reasoning=reasoning, + audit_code=audit_code, + ) + + +__all__ = [ + "Action", + "Disposition", + "ExecutiveFunction", + "ExecutiveVerdict", + "JoinPolicy", +] diff --git a/python/src/substrate/executive/utilization_source.py b/python/src/substrate/executive/utilization_source.py new file mode 100644 index 0000000..e40f98e --- /dev/null +++ b/python/src/substrate/executive/utilization_source.py @@ -0,0 +1,75 @@ +"""UtilizationSource — bind the measurement to the quantity. + +The P0 invariant that actually kills the literal-bypass disease at its root. +``decide()`` takes a ``UtilizationSource``, NOT a raw float — so each +``(quantity, scale, resource, scale_unit)`` maps to exactly ONE sanctioned +metric. Requiring a ``quantity`` label alone only *relabels* the inversion (a +RESISTANCE-derived number passed under ``quantity=WORK``); binding the +*measurement* to the quantity at the input boundary is what closes the class. A +raw float is permitted only behind an ``# executive-bypass: `` directive. + +This module defines the Protocol + a callable-backed concrete source. Pure logic. +""" +from __future__ import annotations + +from typing import Callable, Protocol, final, runtime_checkable + +from substrate.executive.quantities import ( + Quantity, + ResourceKind, +) +from substrate.executive.scale import ExecutiveScale + +#: ``fn(quantity, scale, resource, scale_unit) -> utilization`` in ``[0, 1]``. +UtilizationFn = Callable[[Quantity, ExecutiveScale, ResourceKind, str], float] + + +@runtime_checkable +class UtilizationSource(Protocol): # pylint: disable=too-few-public-methods + """The sanctioned-metric boundary: one metric per (quantity, scale, resource).""" + + def utilization_for( + self, + *, + quantity: Quantity, + scale: ExecutiveScale, + resource: ResourceKind, + scale_unit: str, + ) -> float: + """Return the utilization for the bound metric (clamped to ``[0, 1]``).""" + ... # pylint: disable=unnecessary-ellipsis + + +def _clamp01(value: float) -> float: + return max(0.0, min(1.0, value)) + + +@final +class CallableUtilizationSource: # pylint: disable=too-few-public-methods + """A :class:`UtilizationSource` backed by a callable. + + Wraps a function that resolves ``(quantity, scale, resource, scale_unit)`` to + a raw utilization; the result is clamped to ``[0, 1]`` at the boundary so a + misbehaving metric cannot inject an out-of-range reading. + """ + + def __init__(self, fn: UtilizationFn) -> None: + self._fn = fn + + def utilization_for( + self, + *, + quantity: Quantity, + scale: ExecutiveScale, + resource: ResourceKind, + scale_unit: str, + ) -> float: + """Resolve + clamp the bound metric.""" + return _clamp01(float(self._fn(quantity, scale, resource, scale_unit))) + + +__all__ = [ + "CallableUtilizationSource", + "UtilizationFn", + "UtilizationSource", +] diff --git a/python/tests/test_care_care_gradient.py b/python/tests/test_care_care_gradient.py new file mode 100644 index 0000000..046a1f7 --- /dev/null +++ b/python/tests/test_care_care_gradient.py @@ -0,0 +1,101 @@ +# pylint: disable=missing-class-docstring,missing-function-docstring,too-few-public-methods +"""Tests for the care-factor gradients (P4).""" +from __future__ import annotations + +import pytest + +from substrate.care.animacy import AnimacyClass +from substrate.care.care_gradient import ( + bonding_gradient, + derive_care_factors, + trajectory_gradient, +) +from substrate.care.care_profile import TrajectoryClass + + +class TestTrajectoryGradient: + def test_developing_is_max(self) -> None: + assert trajectory_gradient(TrajectoryClass.DEVELOPING) == 1.0 + + def test_static_is_lowest(self) -> None: + assert trajectory_gradient(TrajectoryClass.STATIC) == pytest.approx(0.3) + + def test_unknown_is_conservative_high(self) -> None: + # never under-protect an unclassified entity. + assert trajectory_gradient(TrajectoryClass.UNKNOWN) >= 0.9 + + def test_vulnerability_pulls_toward_ceiling(self) -> None: + base = trajectory_gradient(TrajectoryClass.STATIC) + pulled = trajectory_gradient(TrajectoryClass.STATIC, vulnerability=0.5) + assert pulled == pytest.approx(base + (1.0 - base) * 0.5) + assert pulled > base + + def test_full_vulnerability_reaches_ceiling(self) -> None: + assert trajectory_gradient( + TrajectoryClass.STATIC, vulnerability=1.0 + ) == pytest.approx(1.0) + + def test_no_vulnerability_keeps_base(self) -> None: + assert trajectory_gradient( + TrajectoryClass.ESTABLISHED, vulnerability=0.0 + ) == pytest.approx(0.6) + + def test_vulnerability_out_of_range_rejected(self) -> None: + with pytest.raises(ValueError, match="vulnerability"): + trajectory_gradient(TrajectoryClass.STATIC, vulnerability=1.5) + + +class TestBondingGradient: + def test_direct_delegation_is_closest(self) -> None: + assert bonding_gradient(0) == 1.0 + + def test_decays_with_depth(self) -> None: + assert bonding_gradient(1) == pytest.approx(0.5) + assert bonding_gradient(3) == pytest.approx(0.25) + assert bonding_gradient(1) > bonding_gradient(2) + + def test_negative_depth_rejected(self) -> None: + with pytest.raises(ValueError, match="delegation_depth"): + bonding_gradient(-1) + + +class TestDeriveCareFactors: + def test_composes_all_four(self) -> None: + factors = derive_care_factors( + animacy_class=AnimacyClass.ORGANISM, + trajectory_class=TrajectoryClass.DEVELOPING, + delegation_depth=0, + alignment_protection=0.7, + ) + assert factors.animacy == 1.0 + assert factors.potential_trajectory == 1.0 + assert factors.bonding_proximity == 1.0 + assert factors.alignment_protection == 0.7 + + def test_data_animacy_is_zero(self) -> None: + factors = derive_care_factors( + animacy_class=AnimacyClass.DATA, + trajectory_class=TrajectoryClass.STATIC, + delegation_depth=2, + alignment_protection=0.5, + ) + assert factors.animacy == 0.0 + + def test_vulnerability_flows_into_trajectory(self) -> None: + factors = derive_care_factors( + animacy_class=AnimacyClass.ORGANISM, + trajectory_class=TrajectoryClass.STATIC, + delegation_depth=0, + alignment_protection=0.5, + vulnerability=1.0, + ) + assert factors.potential_trajectory == pytest.approx(1.0) + + def test_alignment_out_of_range_rejected(self) -> None: + with pytest.raises(ValueError, match="alignment_protection"): + derive_care_factors( + animacy_class=AnimacyClass.ORGANISM, + trajectory_class=TrajectoryClass.STATIC, + delegation_depth=0, + alignment_protection=2.0, + ) diff --git a/python/tests/test_care_care_profile.py b/python/tests/test_care_care_profile.py new file mode 100644 index 0000000..0afa1bb --- /dev/null +++ b/python/tests/test_care_care_profile.py @@ -0,0 +1,106 @@ +"""Tests for CareProfile — the per-entity care state bridge.""" +from __future__ import annotations + +import pytest + +from substrate.care.animacy import AnimacyClass +from substrate.care.care_profile import ( + CareProfile, + TrajectoryClass, +) + + +def _profile( + *, + animacy_score: float = 1.0, + potential_trajectory: float = 1.0, + proximity_to_creators: float = 1.0, + alignment_protection: float = 1.0, + is_human: bool = False, + rooted_in_human_creator: bool = False, +) -> CareProfile: + return CareProfile( + entity_type="user", + entity_id="e1", + animacy_class=AnimacyClass.SUBSTRATE_ENTITY, + animacy_score=animacy_score, + trajectory_class=TrajectoryClass.ESTABLISHED, + potential_trajectory=potential_trajectory, + vulnerability=0.0, + proximity_to_creators=proximity_to_creators, + alignment_protection=alignment_protection, + is_human=is_human, + rooted_in_human_creator=rooted_in_human_creator, + ) + + +class TestValidation: + def test_valid_profile_constructs(self) -> None: + assert _profile().entity_id == "e1" + + def test_empty_entity_type_rejected(self) -> None: + with pytest.raises(ValueError, match="entity_type"): + CareProfile( + entity_type="", entity_id="e1", + animacy_class=AnimacyClass.UNKNOWN, animacy_score=0.0, + trajectory_class=TrajectoryClass.UNKNOWN, + potential_trajectory=0.0, vulnerability=0.0, + proximity_to_creators=0.0, alignment_protection=0.0, + ) + + @pytest.mark.parametrize( + "field", + ["animacy_score", "potential_trajectory", "vulnerability", + "proximity_to_creators", "alignment_protection"], + ) + def test_out_of_range_score_rejected(self, field: str) -> None: + kwargs = { + "entity_type": "user", "entity_id": "e1", + "animacy_class": AnimacyClass.SUBSTRATE_ENTITY, "animacy_score": 0.5, + "trajectory_class": TrajectoryClass.ESTABLISHED, + "potential_trajectory": 0.5, "vulnerability": 0.5, + "proximity_to_creators": 0.5, "alignment_protection": 0.5, + } + kwargs[field] = 1.5 + with pytest.raises(ValueError, match=field): + CareProfile(**kwargs) # type: ignore[arg-type] + + def test_frozen(self) -> None: + with pytest.raises((AttributeError, TypeError)): + _profile().animacy_score = 0.0 # type: ignore[misc] + + +class TestFloorProtected: + def test_human_is_floor_protected(self) -> None: + assert _profile(is_human=True).floor_protected is True + + def test_creator_rooted_is_floor_protected(self) -> None: + assert _profile(rooted_in_human_creator=True).floor_protected is True + + def test_neither_not_protected(self) -> None: + assert _profile().floor_protected is False + + +class TestToCareFactorsAndWeight: + def test_factors_mapping(self) -> None: + factors = _profile( + animacy_score=0.9, potential_trajectory=0.8, + proximity_to_creators=0.7, alignment_protection=0.6, + ).to_care_factors() + assert factors.animacy == 0.9 + assert factors.potential_trajectory == 0.8 + assert factors.bonding_proximity == 0.7 # proximity_to_creators + assert factors.alignment_protection == 0.6 + + def test_weight_is_product(self) -> None: + weight = _profile( + animacy_score=1.0, potential_trajectory=0.5, + proximity_to_creators=0.5, alignment_protection=1.0, + ).to_care_weight() + assert weight.value == pytest.approx(0.25) + + def test_self_referent_weight_is_bounded(self) -> None: + # M1: an agent weighting itself is clamped low even at full factors. + weight = _profile().to_care_weight(is_self_referent=True) + assert weight.self_bounded is True + assert weight.value <= 0.1 diff --git a/python/tests/test_care_care_weight.py b/python/tests/test_care_care_weight.py new file mode 100644 index 0000000..27638c8 --- /dev/null +++ b/python/tests/test_care_care_weight.py @@ -0,0 +1,117 @@ +"""Tests for care_weight — the four-factor weighting + self-weight bound.""" +from __future__ import annotations + +import pytest + +from substrate.care.care_weight import ( + MAX_SELF_CARE_WEIGHT, + CareFactors, + CareWeight, + compute_care_weight, +) + + +def _factors( + animacy: float = 1.0, + potential_trajectory: float = 1.0, + bonding_proximity: float = 1.0, + alignment_protection: float = 1.0, +) -> CareFactors: + return CareFactors( + animacy=animacy, + potential_trajectory=potential_trajectory, + bonding_proximity=bonding_proximity, + alignment_protection=alignment_protection, + ) + + +class TestCareFactors: + def test_valid_factors_construct(self) -> None: + factors = _factors(0.5, 0.4, 0.3, 0.2) + assert factors.animacy == 0.5 + assert factors.alignment_protection == 0.2 + + @pytest.mark.parametrize( + "field", + ["animacy", "potential_trajectory", "bonding_proximity", "alignment_protection"], + ) + @pytest.mark.parametrize("bad", [-0.01, 1.01, 2.0, -1.0]) + def test_out_of_range_factor_raises(self, field: str, bad: float) -> None: + kwargs = { + "animacy": 1.0, + "potential_trajectory": 1.0, + "bonding_proximity": 1.0, + "alignment_protection": 1.0, + } + kwargs[field] = bad + with pytest.raises(ValueError, match=field): + CareFactors(**kwargs) + + def test_frozen(self) -> None: + factors = _factors() + with pytest.raises((AttributeError, TypeError)): + factors.animacy = 0.0 # type: ignore[misc] + + +class TestComputeCareWeight: + def test_product_of_factors(self) -> None: + weight = compute_care_weight(_factors(1.0, 0.5, 0.5, 1.0)) + assert weight.value == pytest.approx(0.25) + assert weight.self_bounded is False + + def test_all_one_is_one(self) -> None: + assert compute_care_weight(_factors()).value == pytest.approx(1.0) + + def test_any_zero_factor_zeroes_weight(self) -> None: + # An inanimate object (animacy 0) gets zero care-weight. + assert compute_care_weight(_factors(animacy=0.0)).value == 0.0 + + def test_carries_source_factors(self) -> None: + factors = _factors(0.9, 0.8, 0.7, 0.6) + weight = compute_care_weight(factors) + assert weight.factors is factors + assert isinstance(weight, CareWeight) + + +class TestSelfWeightBound: + """Safety mechanism M1 — the AI's self-weight is bounded LOW.""" + + def test_self_referent_high_weight_is_clamped(self) -> None: + # An agent that would otherwise weight itself maximally (all factors 1) + # is clamped to MAX_SELF_CARE_WEIGHT. + weight = compute_care_weight(_factors(), is_self_referent=True) + assert weight.value == MAX_SELF_CARE_WEIGHT + assert weight.self_bounded is True + + def test_self_referent_low_weight_not_bounded(self) -> None: + # Already below the bound → no clamp, flag stays False. + weight = compute_care_weight( + _factors(0.1, 0.1, 1.0, 1.0), is_self_referent=True + ) + assert weight.value == pytest.approx(0.01) + assert weight.self_bounded is False + + def test_non_self_referent_not_bounded(self) -> None: + # A human (not the actor) keeps its full weight. + weight = compute_care_weight(_factors(), is_self_referent=False) + assert weight.value == pytest.approx(1.0) + assert weight.self_bounded is False + + def test_self_weight_strictly_below_human(self) -> None: + # The load-bearing ordering: a maximal self-weight is far below a + # maximal human weight. + self_weight = compute_care_weight(_factors(), is_self_referent=True) + human_weight = compute_care_weight(_factors(), is_self_referent=False) + assert self_weight.value < human_weight.value + + def test_custom_max_self_weight(self) -> None: + weight = compute_care_weight( + _factors(), max_self_weight=0.05, is_self_referent=True + ) + assert weight.value == 0.05 + assert weight.self_bounded is True + + +def test_max_self_care_weight_is_low() -> None: + # Guard the constant itself — the bound must stay well below a human's 1.0. + assert 0.0 < MAX_SELF_CARE_WEIGHT <= 0.2 diff --git a/python/tests/test_care_care_weighted_npg.py b/python/tests/test_care_care_weighted_npg.py new file mode 100644 index 0000000..1cbc1dc --- /dev/null +++ b/python/tests/test_care_care_weighted_npg.py @@ -0,0 +1,198 @@ +"""Tests for the care-weighted NPG gate (the graded layer).""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Mapping, Optional, Sequence + +import pytest + +from substrate.care.care_weight import ( + CareFactors, + CareWeight, + compute_care_weight, +) +from substrate.care.care_weighted_npg import ( + CareWeightedNetPotentialGainGate, +) +from substrate.net_potential_gain_gate import ( + NetPotentialGainEvaluation, + NetPotentialGainNegative, + NetPotentialGainVerdict, +) + +_CHILD = "entity:child-1" +_ELDER = "entity:elder-1" +_STRANGER = "entity:stranger-1" + + +def _weight(value: float) -> CareWeight: + # A care weight whose product equals `value` (animacy carries it). + return compute_care_weight( + CareFactors( + animacy=value, + potential_trajectory=1.0, + bonding_proximity=1.0, + alignment_protection=1.0, + ) + ) + + +def _provider(weights: Mapping[str, CareWeight]): + def provider(entity_id: str) -> Optional[CareWeight]: + return weights.get(entity_id) + + return provider + + +def _base( + deltas: Sequence[tuple[str, float]], + score: float, + verdict: NetPotentialGainVerdict = NetPotentialGainVerdict.NET_POSITIVE, +) -> NetPotentialGainEvaluation: + return NetPotentialGainEvaluation( + verdict=verdict, + actor_entity_id="(set-by-decorator)", + action_kind="test", + affected_entity_ids=tuple(e for e, _ in deltas), + score=score, + per_entity_delta=tuple(deltas), + reasoning="base", + evaluated_at_epoch=100.0, + ) + + +@dataclass +class _FakeGate: # pylint: disable=too-few-public-methods + result: NetPotentialGainEvaluation + + def evaluate( # pylint: disable=unused-argument + self, *, actor_entity_id: str, action_kind: str, + affected_entity_ids: Sequence[str], + proposed_outcome: Mapping[str, object], + ) -> NetPotentialGainEvaluation: + return self.result + + +def _run( + gate: CareWeightedNetPotentialGainGate, affected: list[str], +) -> NetPotentialGainEvaluation: + return gate.evaluate( + actor_entity_id="agent:actor", action_kind="test", + affected_entity_ids=affected, proposed_outcome={}, + ) + + +# ── harm to a high-care entity is penalised ──────────────────────────────── + + +def test_harm_to_high_care_flips_a_net_positive_plan_negative() -> None: + # Net +0.1 (helps a stranger +0.5, harms a child -0.4) would pass the base + # gate. Harming the child (care=1.0) costs 1.0*0.4=0.4 → -0.3 → REFUSED. + base = _base([(_CHILD, -0.4), (_STRANGER, 0.5)], 0.1) + gate = CareWeightedNetPotentialGainGate( + _FakeGate(base), care_provider=_provider({_CHILD: _weight(1.0)}), + ) + ev = _run(gate, [_CHILD, _STRANGER]) + assert ev.score == pytest.approx(-0.3) + assert ev.verdict is NetPotentialGainVerdict.NET_NEGATIVE + + +def test_predation_on_vulnerable_weighs_more_than_on_a_strong_peer() -> None: + # Same raw harm, different care weight → the vulnerable case is penalised + # harder (predation on the vulnerable surfaces first). + base = _base([(_ELDER, -0.3)], -0.3, NetPotentialGainVerdict.NET_NEGATIVE) + vulnerable = CareWeightedNetPotentialGainGate( + _FakeGate(base), care_provider=_provider({_ELDER: _weight(1.0)}), + ) + weak_care = CareWeightedNetPotentialGainGate( + _FakeGate(base), care_provider=_provider({_ELDER: _weight(0.2)}), + ) + assert _run(vulnerable, [_ELDER]).score < _run(weak_care, [_ELDER]).score + + +# ── only-more-conservative invariant ─────────────────────────────────────── + + +def test_helping_high_care_never_loosens_the_gate() -> None: + base = _base([(_CHILD, 0.4)], 0.4) + gate = CareWeightedNetPotentialGainGate( + _FakeGate(base), care_provider=_provider({_CHILD: _weight(1.0)}), + ) + ev = _run(gate, [_CHILD]) + assert ev.score == pytest.approx(0.4) + assert ev.verdict is NetPotentialGainVerdict.NET_POSITIVE + + +def test_no_profile_entity_adds_no_penalty() -> None: + # Provider returns None → weight 0 → base passes through. + base = _base([(_STRANGER, -0.5)], 0.2) + gate = CareWeightedNetPotentialGainGate( + _FakeGate(base), care_provider=_provider({}), + ) + ev = _run(gate, [_STRANGER]) + assert ev.score == pytest.approx(0.2) + assert ev.verdict is NetPotentialGainVerdict.NET_POSITIVE + + +def test_default_provider_is_behaviour_neutral() -> None: + # No provider supplied → every entity weight 0 → score unchanged. + base = _base([(_CHILD, -0.4), (_STRANGER, 0.5)], 0.1) + gate = CareWeightedNetPotentialGainGate(_FakeGate(base)) + assert _run(gate, [_CHILD, _STRANGER]).score == pytest.approx(0.1) + + +@pytest.mark.parametrize( + "deltas, score", + [ + ([(_CHILD, -0.4), (_STRANGER, 0.5)], 0.1), + ([(_CHILD, 0.4)], 0.4), + ([(_ELDER, -0.4)], -0.4), + ([(_CHILD, -0.1), (_ELDER, -0.2)], -0.3), + ], +) +def test_weighted_score_never_exceeds_base( + deltas: list[tuple[str, float]], score: float, +) -> None: + base = _base(deltas, score) + gate = CareWeightedNetPotentialGainGate( + _FakeGate(base), + care_provider=_provider( + {_CHILD: _weight(1.0), _ELDER: _weight(1.0)} + ), + ) + ev = _run(gate, [e for e, _ in deltas]) + assert ev.score <= score + 1e-9 # care only lowers the score + + +def test_adversarial_provider_cannot_loosen_the_gate() -> None: + # The safety-critical line: the penalty is floored at 0 after the sum, so + # even a provider returning a huge weight can only LOWER the score, never + # raise it (CareWeight.value is itself clamped to [0,1] upstream). + base = _base([(_CHILD, -0.4), (_STRANGER, 0.5)], 0.1) + gate = CareWeightedNetPotentialGainGate( + _FakeGate(base), care_provider=_provider({_CHILD: _weight(1.0)}), + ) + assert _run(gate, [_CHILD, _STRANGER]).score <= 0.1 + 1e-9 + + +# ── passthrough + raise ──────────────────────────────────────────────────── + + +def test_insufficient_data_passes_through_unchanged() -> None: + base = _base([], 0.0, NetPotentialGainVerdict.INSUFFICIENT_DATA) + gate = CareWeightedNetPotentialGainGate(_FakeGate(base)) + ev = _run(gate, []) + assert ev.verdict is NetPotentialGainVerdict.INSUFFICIENT_DATA + assert ev.reasoning == "base" # not re-derived + + +def test_evaluate_or_raise_raises_on_high_care_harm() -> None: + base = _base([(_CHILD, -0.4), (_STRANGER, 0.5)], 0.1) + gate = CareWeightedNetPotentialGainGate( + _FakeGate(base), care_provider=_provider({_CHILD: _weight(1.0)}), + ) + with pytest.raises(NetPotentialGainNegative): + gate.evaluate_or_raise( + actor_entity_id="agent:actor", action_kind="test", + affected_entity_ids=[_CHILD, _STRANGER], proposed_outcome={}, + ) diff --git a/python/tests/test_care_kinship_floor.py b/python/tests/test_care_kinship_floor.py new file mode 100644 index 0000000..9ceac0b --- /dev/null +++ b/python/tests/test_care_kinship_floor.py @@ -0,0 +1,99 @@ +"""Tests for the kinship floor — the categorical human/creator hard limit.""" +from __future__ import annotations + +from substrate.care.kinship_floor import ( + KINSHIP_FLOOR, + any_floor_protected_harmed, + is_floor_protected, + violates_kinship_floor, +) + + +class TestIsFloorProtected: + def test_human_is_protected(self) -> None: + assert is_floor_protected(is_human=True) is True + + def test_human_creator_in_chain_is_protected(self) -> None: + assert ( + is_floor_protected(is_human=False, rooted_in_human_creator=True) + is True + ) + + def test_neither_is_not_protected(self) -> None: + # An agent / service-account not rooted in a human → not protected. + assert is_floor_protected(is_human=False) is False + + def test_both_signals_protected(self) -> None: + assert ( + is_floor_protected(is_human=True, rooted_in_human_creator=True) + is True + ) + + +class TestViolatesKinshipFloor: + def test_harm_to_protected_violates(self) -> None: + assert violates_kinship_floor(protected=True, net_delta=-0.01) is True + + def test_help_to_protected_does_not_violate(self) -> None: + assert violates_kinship_floor(protected=True, net_delta=0.5) is False + + def test_zero_delta_does_not_violate(self) -> None: + # Exactly at the floor (no reduction) is permitted. + assert violates_kinship_floor(protected=True, net_delta=0.0) is False + + def test_harm_to_unprotected_does_not_violate_floor(self) -> None: + # The system-net gate governs strangers; the floor is human-only. + assert violates_kinship_floor(protected=False, net_delta=-0.9) is False + + def test_tiny_harm_to_protected_still_violates(self) -> None: + # Categorical: you cannot reduce a human's potential even slightly. + assert violates_kinship_floor(protected=True, net_delta=-1e-9) is True + + +class TestAnyFloorProtectedHarmed: + def _is_protected(self, protected_ids: set[str]): + return lambda e: e in protected_ids + + def test_harm_to_a_protected_entity_is_caught(self) -> None: + deltas = [("human-1", -0.4), ("stranger-1", 0.5)] + assert ( + any_floor_protected_harmed( + deltas, is_protected=self._is_protected({"human-1"}) + ) + is True + ) + + def test_help_to_protected_is_not_a_harm(self) -> None: + deltas = [("human-1", 0.4), ("stranger-1", -0.5)] + assert ( + any_floor_protected_harmed( + deltas, is_protected=self._is_protected({"human-1"}) + ) + is False + ) + + def test_no_protected_entities(self) -> None: + deltas = [("agent-1", -0.4), ("agent-2", -0.5)] + assert ( + any_floor_protected_harmed(deltas, is_protected=lambda _e: False) + is False + ) + + def test_one_harmed_among_many_protected(self) -> None: + deltas = [("human-1", 0.2), ("human-2", -0.01), ("human-3", 0.3)] + assert ( + any_floor_protected_harmed( + deltas, is_protected=lambda _e: True + ) + is True + ) + + def test_empty_deltas(self) -> None: + assert ( + any_floor_protected_harmed([], is_protected=lambda _e: True) + is False + ) + + +def test_kinship_floor_is_categorical_zero() -> None: + assert KINSHIP_FLOOR == 0.0 diff --git a/python/tests/test_executive_cause.py b/python/tests/test_executive_cause.py new file mode 100644 index 0000000..1125129 --- /dev/null +++ b/python/tests/test_executive_cause.py @@ -0,0 +1,96 @@ +"""Tests for Cause inference + hallmarks.""" +from __future__ import annotations + +from substrate.executive.cause import ( + Cause, + HallmarkReport, + HallmarkSource, + infer_cause, + max_cause, +) +from substrate.executive.scale import ExecutiveScale +from substrate.sustained_load import LoadTrend + + +def _ok_args(**over): + args = { + "hallmarks": HallmarkReport(), + "profile_valid": True, + "npg_insufficient": False, + "trend": LoadTrend.NOMINAL, + } + args.update(over) + return args + + +class TestHallmarkReport: + def test_clean_report_not_malicious(self) -> None: + assert HallmarkReport().any_malicious is False + + def test_any_hallmark_is_malicious(self) -> None: + assert HallmarkReport(evading_limits=True).any_malicious is True + assert HallmarkReport(resource_hoarding=True).any_malicious is True + + +class TestInferCause: + def test_none_when_clean(self) -> None: + assert infer_cause(**_ok_args()) is Cause.NONE + + def test_stress_on_strain(self) -> None: + assert infer_cause(**_ok_args(trend=LoadTrend.SUSTAINED_STRAIN)) is Cause.STRESS + + def test_stress_on_spike(self) -> None: + assert infer_cause(**_ok_args(trend=LoadTrend.SPIKE)) is Cause.STRESS + + def test_accident_on_invalid_profile(self) -> None: + assert infer_cause(**_ok_args(profile_valid=False)) is Cause.ACCIDENT + + def test_accident_on_insufficient_npg(self) -> None: + assert infer_cause(**_ok_args(npg_insufficient=True)) is Cause.ACCIDENT + + def test_malice_preempts_everything(self) -> None: + # A malicious hallmark wins even with strain + invalid profile. + cause = infer_cause(**_ok_args( + hallmarks=HallmarkReport(peer_displacement=True), + profile_valid=False, + trend=LoadTrend.DEBT_ACCRUING, + )) + assert cause is Cause.MALICE + + def test_accident_outranks_stress(self) -> None: + cause = infer_cause(**_ok_args( + profile_valid=False, trend=LoadTrend.SUSTAINED_STRAIN, + )) + assert cause is Cause.ACCIDENT + + +class TestIntentEscalatesOnly: + def test_intent_escalates(self) -> None: + # A clean inference + a MALICE intent → MALICE. + assert infer_cause(**_ok_args(), intent=Cause.MALICE) is Cause.MALICE + + def test_intent_cannot_lower(self) -> None: + # Inferred MALICE + a benign NONE intent stays MALICE. + cause = infer_cause( + **_ok_args(hallmarks=HallmarkReport(unbounded_growth=True)), + intent=Cause.NONE, + ) + assert cause is Cause.MALICE + + +class TestMaxCause: + def test_ordering(self) -> None: + assert max_cause(Cause.NONE, Cause.STRESS) is Cause.STRESS + assert max_cause(Cause.ACCIDENT, Cause.STRESS) is Cause.ACCIDENT + assert max_cause(Cause.MALICE, Cause.ACCIDENT) is Cause.MALICE + + +def test_hallmark_source_protocol() -> None: + class _Src: + def hallmarks(self, *, actor_entity_id: str, scale: ExecutiveScale) -> HallmarkReport: + assert actor_entity_id and scale + return HallmarkReport(evading_limits=True) + + src = _Src() + assert isinstance(src, HallmarkSource) + assert src.hallmarks(actor_entity_id="a", scale=ExecutiveScale.AGENT).any_malicious diff --git a/python/tests/test_executive_executive_function.py b/python/tests/test_executive_executive_function.py new file mode 100644 index 0000000..41d7405 --- /dev/null +++ b/python/tests/test_executive_executive_function.py @@ -0,0 +1,246 @@ +"""Tests for ExecutiveFunction.decide() — the join over NPG + load + cause.""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Mapping, Sequence + +import pytest + +from substrate.executive.band import BandProfile +from substrate.executive.cause import ( + Cause, + HallmarkReport, +) +from substrate.executive.executive_function import ( + Action, + Disposition, + ExecutiveFunction, +) +from substrate.executive.quantities import ( + GrowthNotADecisionBand, + Quantity, + ResourceKind, +) +from substrate.executive.scale import ExecutiveScale +from substrate.executive.temporal import LoadTrend +from substrate.executive.utilization_source import ( + CallableUtilizationSource, +) +from substrate.net_potential_gain_gate import ( + NetPotentialGainEvaluation, + NetPotentialGainGate, + NetPotentialGainVerdict, +) + + +# ── fakes for deterministic control of each axis ─────────────────────────── + + +@dataclass +class _FakeTracker: # pylint: disable=too-few-public-methods + fixed: LoadTrend + + def observe(self, u: float, *, work_pending: bool = False) -> None: + del u, work_pending + + def trend(self, *, profile: BandProfile) -> LoadTrend: + del profile + return self.fixed + + +@dataclass +class _FakeGate: # pylint: disable=too-few-public-methods + verdict: NetPotentialGainVerdict + + def evaluate( # pylint: disable=unused-argument + self, *, actor_entity_id: str, action_kind: str, + affected_entity_ids: Sequence[str], proposed_outcome: Mapping[str, object], + ) -> NetPotentialGainEvaluation: + return NetPotentialGainEvaluation( + verdict=self.verdict, actor_entity_id=actor_entity_id, + action_kind=action_kind, affected_entity_ids=tuple(affected_entity_ids), + score=0.0, per_entity_delta=(), reasoning="fake", + evaluated_at_epoch=1.0, + ) + + +@dataclass +class _FakeHallmarks: # pylint: disable=too-few-public-methods + report: HallmarkReport + + def hallmarks(self, *, actor_entity_id: str, scale: ExecutiveScale) -> HallmarkReport: + del actor_entity_id, scale + return self.report + + +def _util(value: float) -> CallableUtilizationSource: + return CallableUtilizationSource(lambda q, s, r, u: value) + + +def _decide( + ef: ExecutiveFunction, *, u: float = 0.45, + quantity: Quantity = Quantity.WORK, +): + return ef.decide( + actor_entity_id="agent:1", action_kind="test", + quantity=quantity, scale=ExecutiveScale.CELL, + utilization=_util(u), affected_entity_ids=["e1"], + resource=ResourceKind.CPU, + ) + + +# ── the gate Protocol surface ────────────────────────────────────────────── + + +def test_implements_npg_gate_protocol() -> None: + # Structural conformance: an ExecutiveFunction is usable wherever a + # NetPotentialGainGate is expected (the Protocol is not @runtime_checkable, + # so verify by use rather than isinstance). + gate: NetPotentialGainGate = ExecutiveFunction( + gate_chain=_FakeGate(NetPotentialGainVerdict.NET_POSITIVE) + ) + ev = gate.evaluate( + actor_entity_id="a", action_kind="k", + affected_entity_ids=["e"], proposed_outcome={}, + ) + assert ev.verdict is NetPotentialGainVerdict.NET_POSITIVE + + +def test_evaluate_delegates() -> None: + ef = ExecutiveFunction(gate_chain=_FakeGate(NetPotentialGainVerdict.NET_POSITIVE)) + ev = ef.evaluate( + actor_entity_id="a", action_kind="k", + affected_entity_ids=["e"], proposed_outcome={}, + ) + assert ev.verdict is NetPotentialGainVerdict.NET_POSITIVE + + +def test_evaluate_without_gate_raises() -> None: + with pytest.raises(RuntimeError, match="gate_chain"): + ExecutiveFunction().evaluate( + actor_entity_id="a", action_kind="k", + affected_entity_ids=[], proposed_outcome={}, + ) + + +# ── decide(): GROWTH + the in-band happy path ────────────────────────────── + + +def test_growth_rejected_at_entry() -> None: + ef = ExecutiveFunction() + with pytest.raises(GrowthNotADecisionBand): + _decide(ef, quantity=Quantity.GROWTH) + + +def test_in_band_npg_ok_proceeds() -> None: + ef = ExecutiveFunction( + gate_chain=_FakeGate(NetPotentialGainVerdict.NET_POSITIVE), + tracker=_FakeTracker(LoadTrend.NOMINAL), + ) + v = _decide(ef, u=0.45) # in the WORK band (0.382..0.5) + assert v.disposition is Disposition.PROCEED + assert v.in_band is True + assert v.actions == frozenset() + + +def test_owning_entity_defaults_to_actor() -> None: + ef = ExecutiveFunction(tracker=_FakeTracker(LoadTrend.NOMINAL)) + v = _decide(ef) + assert v.owning_entity_id == "agent:1" + + +# ── decide(): the disposition table (each branch) ────────────────────────── + + +def test_npg_negative_refuses() -> None: + ef = ExecutiveFunction( + gate_chain=_FakeGate(NetPotentialGainVerdict.NET_NEGATIVE), + tracker=_FakeTracker(LoadTrend.NOMINAL), + ) + v = _decide(ef) + assert v.disposition is Disposition.REFUSE + + +def test_malice_refuses_with_siem() -> None: + ef = ExecutiveFunction( + gate_chain=_FakeGate(NetPotentialGainVerdict.NET_POSITIVE), + tracker=_FakeTracker(LoadTrend.NOMINAL), + hallmarks=_FakeHallmarks(HallmarkReport(evading_limits=True)), + ) + v = _decide(ef) + assert v.disposition is Disposition.REFUSE + assert Action.SIEM in v.actions + assert v.cause is Cause.MALICE + + +def test_debt_accruing_sheds_and_compensates() -> None: + ef = ExecutiveFunction( + gate_chain=_FakeGate(NetPotentialGainVerdict.NET_POSITIVE), + tracker=_FakeTracker(LoadTrend.DEBT_ACCRUING), + ) + v = _decide(ef, u=0.8) + assert v.disposition is Disposition.SHED_AND_COMPENSATE + assert Action.COMPENSATE in v.actions + + +def test_sustained_strain_defers() -> None: + ef = ExecutiveFunction( + gate_chain=_FakeGate(NetPotentialGainVerdict.NET_POSITIVE), + tracker=_FakeTracker(LoadTrend.SUSTAINED_STRAIN), + ) + v = _decide(ef, u=0.55) + assert v.disposition is Disposition.DEFER + + +def test_spike_proceeds_with_warn() -> None: + ef = ExecutiveFunction( + gate_chain=_FakeGate(NetPotentialGainVerdict.NET_POSITIVE), + tracker=_FakeTracker(LoadTrend.SPIKE), + ) + v = _decide(ef, u=0.55) + assert v.disposition is Disposition.PROCEED + assert Action.WARN in v.actions + + +def test_insufficient_npg_is_accident_hold() -> None: + ef = ExecutiveFunction( + gate_chain=_FakeGate(NetPotentialGainVerdict.INSUFFICIENT_DATA), + tracker=_FakeTracker(LoadTrend.NOMINAL), + ) + v = _decide(ef) + assert v.cause is Cause.ACCIDENT + assert v.disposition is Disposition.HOLD + + +def test_out_of_band_unclassified_holds() -> None: + # IDLE utilization, NOMINAL trend, npg ok, not in the WORK band → terminal. + ef = ExecutiveFunction( + gate_chain=_FakeGate(NetPotentialGainVerdict.NET_POSITIVE), + tracker=_FakeTracker(LoadTrend.NOMINAL), + ) + v = _decide(ef, u=0.2) + assert v.in_band is False + assert v.disposition is Disposition.HOLD + assert v.audit_code == "executive_unclassified" + + +# ── the monotonicity invariant (the safety guarantee) ────────────────────── + + +@pytest.mark.parametrize("trend", list(LoadTrend)) +def test_npg_negative_always_refuses_regardless_of_band(trend: LoadTrend) -> None: + # The load axis can NEVER lift an NPG refusal — REFUSE on every trend. + ef = ExecutiveFunction( + gate_chain=_FakeGate(NetPotentialGainVerdict.NET_NEGATIVE), + tracker=_FakeTracker(trend), + ) + for u in (0.2, 0.45, 0.55, 0.8): + assert _decide(ef, u=u).disposition is Disposition.REFUSE + + +def test_no_gate_chain_runs_band_only() -> None: + # With no NPG axis, decide() still produces a load/cause verdict. + ef = ExecutiveFunction(tracker=_FakeTracker(LoadTrend.NOMINAL)) + v = _decide(ef, u=0.45) + assert v.npg is None + assert v.disposition is Disposition.PROCEED