From f2f46340b4ecec421171dc68dae6c934d20fa26f Mon Sep 17 00:00:00 2001 From: Denny Schaedig Date: Sat, 16 May 2026 21:22:47 -0600 Subject: [PATCH 1/2] =?UTF-8?q?feat(0.9.1):=20Wire=203=20=E2=80=94=20embod?= =?UTF-8?q?iment-state=20=E2=86=92=20tool=20filter?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 1 of release_0_9_1.md, lifted from bio_emergent_persona_foundations.md § Wire 3. The smallest of the four wires by LOC, highest behavioral signal per unit work: an agent with a damaged arm visibly stops attempting arm-routed affordances without any prompt-injection scaffolding. The cleanest emergent "trait" demonstration in the foundations plan. Implementation: - Embodiment.get_disabled_affordances(*, threshold=None) → set[str] Walks the entity tree, computes each modulator's compute_integrity(), and returns base tool names ({entity.name}_{affordance_name}) for modulators strictly below the disable threshold (default 0.3). Modulators that don't expose compute_integrity (capability-only, decorator-style) default to integrity=1.0 per the backward-compat convention SpecModulator.compute_integrity already uses on empty vital_metrics. A buggy modulator whose compute_integrity raises is treated as integrity=1.0 (fail-open). - Embodiment.get_degraded_affordances(*, disable_threshold=None, degrade_threshold=None) → dict[str, float] Same walk, returns {base_tool_name: integrity} for modulators in [disable_threshold, degrade_threshold) (default [0.3, 0.6)). The bands partition [0, 1] cleanly — every affordance lands in exactly one of {disabled, degraded, healthy}, never both. - agent_loop.py hook between mode_info.get_available_tools(...) and the tool_descriptions build loop: 1. Filter disabled affordances out of available_tools. 2. After per-tool description build, append "[DAMAGED: integrity 0.X]" to each degraded affordance's description. Fail-open: no embodiment, missing compute_integrity, raising modulator → the hook is a no-op. Description annotation is idempotent (the "if annotation not in base_desc" guard prevents the suffix accumulating across ticks). Copy-on-write — TOOL_DESCRIPTIONS is a shared module-level dict; mutation would poison future calls and other agents. Tool-name pattern (the load-bearing assumption): - tool_bridge.generate_tools_for_entity registers ModulatorAffordanceTools as {entity.name}_{affordance_name} unless _resolve_tool_name collision-prefixes an ancestor name. - Roy / cradle / Reachy use single-body topologies — no collisions in practice. Wire 3's base names match the registered tool names cleanly. Under a hypothetical collision, the filter fails open (the tool stays available; no silent mis-gating). - TestToolNamePattern.test_base_name_matches_generated_tool_name pins this contract: it constructs a real ToolRegistry, calls generate_tools_for_entity, and asserts every predicted-disabled name is in the registry. Test surface (30 tests, 6 layers): - Layer 1 (get_disabled_affordances, 7 tests): healthy/critical/ boundary/per-modulator-isolation/threshold-override. - Layer 2 (get_degraded_affordances, 7 tests): boundary semantics for both ends of the [0.3, 0.6) band; threshold overrides. - Layer 3 (band partitioning, 8 parametric tests): no integrity value lands in both disabled and degraded sets. - Layer 4 (tool-name pattern, 1 test): live ToolRegistry round-trip via generate_tools_for_entity. - Layer 5 (agent_loop hook shape, 4 tests): filter/annotate/ idempotency/no-embodiment-no-op. - Layer 6 (degenerate cases, 3 tests): empty modulators, empty affordances, raising compute_integrity. All 30 tests pass. Ruff clean. Frozen contract impact: none. No new persisted state, no dataclass changes. Pure read-side wiring per the plan. Behavioral signal: a damaged-arm agent stops calling arm-routed affordances. Roy-3 validation can measure this once Wires 1-A all ship. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/maxim/embodiment/body.py | 99 +++++ src/maxim/runtime/agent_loop.py | 50 +++ tests/unit/test_wire_3_embodiment_filter.py | 436 ++++++++++++++++++++ 3 files changed, 585 insertions(+) create mode 100644 tests/unit/test_wire_3_embodiment_filter.py diff --git a/src/maxim/embodiment/body.py b/src/maxim/embodiment/body.py index 9e9e5602..cbab3700 100644 --- a/src/maxim/embodiment/body.py +++ b/src/maxim/embodiment/body.py @@ -11,6 +11,7 @@ import logging import time from dataclasses import dataclass, field +from collections.abc import Iterable from typing import Any from maxim.embodiment.sem import Entity, FailureMode, SensorReading @@ -574,6 +575,104 @@ def format_body_state_for_prompt(self) -> str: def failure_history(self) -> list[FailureEvent]: return list(self._failure_history) + # -- Wire 3: embodiment-state → action filter (release_0_9_1.md Stage 1) - + # Default thresholds (also documented in + # docs/plans/bio_emergent_persona_foundations.md § Wire 3). + # An agent_loop hook reads these via getattr() so a non-default + # threshold pair can ship in a future tuning experiment without + # touching call sites. + + _WIRE_3_DISABLE_THRESHOLD: float = 0.3 + _WIRE_3_DEGRADE_THRESHOLD: float = 0.6 + + def _iter_modulator_affordance_pairs( + self, + ) -> Iterable[tuple[str, float]]: + """Yield ``(base_tool_name, integrity)`` for every modulator + affordance on the entity tree. + + ``base_tool_name`` is the ``{entity.name}_{affordance_name}`` + form ``tool_bridge.generate_tools_for_entity`` uses BEFORE + ``_resolve_tool_name`` disambiguates against the registry. + On the common single-body topology (Roy / cradle / Reachy) + there are no name collisions, so the registered tool name + equals the base name and Wire 3 matches cleanly. The + agent_loop hook compares against the live tool list — if a + collision did rename a tool to ``{ancestor}_{base_name}``, + the integrity filter fails open (the tool stays available) + rather than silently mis-gating. + + Modulators without a ``compute_integrity`` method (older + SpecModulator-shaped types, capability-only modulators) yield + ``1.0`` — equivalent to "not damaged", per the same + backward-compat convention ``SpecModulator.compute_integrity`` + uses when ``vital_metrics`` is empty. + """ + for ent in self.root.walk(): + for mod in ent.modulators.values(): + if hasattr(mod, "compute_integrity"): + try: + integrity = float(mod.compute_integrity()) + except Exception: + integrity = 1.0 + else: + integrity = 1.0 + if not hasattr(mod, "affordances"): + continue + for aff_name in mod.affordances.keys(): + yield f"{ent.name}_{aff_name}", integrity + + def get_disabled_affordances(self, *, threshold: float | None = None) -> set[str]: + """Affordances routed through critically-damaged components. + + Returns the set of base tool names (``{entity.name}_{affordance_name}``) + whose owning modulator's ``compute_integrity()`` is **strictly + below** the disable threshold (default ``0.3``). The agent_loop + hook filters these from the per-tick available-tools list + BEFORE the LLM prompt sees them — a damaged-arm agent stops + attempting arm-routed affordances without any prompt-injection + scaffolding, the cleanest emergent "trait" demonstration in + bio_emergent_persona_foundations.md § Wire 3. + + See ``_iter_modulator_affordance_pairs`` for the base-name + derivation contract; failures match cleanly on Roy's + single-body topology and fail-open under name collisions. + + Args: + threshold: Override ``_WIRE_3_DISABLE_THRESHOLD`` (0.3). + Below this integrity, the affordance is disabled. + """ + cutoff = float(threshold) if threshold is not None else self._WIRE_3_DISABLE_THRESHOLD + return {name for name, integrity in self._iter_modulator_affordance_pairs() if integrity < cutoff} + + def get_degraded_affordances( + self, + *, + disable_threshold: float | None = None, + degrade_threshold: float | None = None, + ) -> dict[str, float]: + """Affordances on partially-damaged components. + + Returns ``{base_tool_name: integrity}`` for every modulator + affordance whose owning modulator's integrity is in the + ``[disable_threshold, degrade_threshold)`` range — damaged + but not disabled. The agent_loop hook annotates these tools' + descriptions with ``[DAMAGED: integrity 0.X]`` so the LLM + proposer sees the cost of using them; learning is post-hoc + via the standard reward path (damaged-tool use → likelier + failure → NAc credit). + + Args: + disable_threshold: Below this integrity, the affordance + is in ``get_disabled_affordances`` instead and is NOT + in this map. Default 0.3. + degrade_threshold: At or above this integrity, the + affordance is healthy and NOT in this map. Default 0.6. + """ + lo = float(disable_threshold) if disable_threshold is not None else self._WIRE_3_DISABLE_THRESHOLD + hi = float(degrade_threshold) if degrade_threshold is not None else self._WIRE_3_DEGRADE_THRESHOLD + return {name: integrity for name, integrity in self._iter_modulator_affordance_pairs() if lo <= integrity < hi} + # -- failure persistence ------------------------------------------------- def export_failure_state(self) -> dict[str, Any]: diff --git a/src/maxim/runtime/agent_loop.py b/src/maxim/runtime/agent_loop.py index a0ba0a11..cd9f2adf 100644 --- a/src/maxim/runtime/agent_loop.py +++ b/src/maxim/runtime/agent_loop.py @@ -3031,6 +3031,31 @@ def _get_all_tools() -> set[str]: # Get available tools for this mode available_tools = mode_info.get_available_tools(_all_tools) + + # Wire 3 (release_0_9_1.md Stage 1): filter tools + # routed through critically-damaged components and + # collect degraded-affordance annotations. Pulls + # from Embodiment.get_disabled_affordances() / + # .get_degraded_affordances() which read each + # modulator's compute_integrity(). Default + # thresholds: integrity < 0.3 → disabled + # (filtered out); 0.3 <= integrity < 0.6 → + # annotated with "[DAMAGED: integrity 0.X]" so + # the LLM proposer sees the cost of using the + # affordance. Fail-open: no embodiment, no + # modulators, missing compute_integrity → + # the filter is a no-op. + _wire3_embodiment = getattr(executor, "embodiment", None) + _wire3_degraded: dict[str, float] = {} + if _wire3_embodiment is not None: + try: + _wire3_disabled = _wire3_embodiment.get_disabled_affordances() + if _wire3_disabled: + available_tools = [t for t in available_tools if t not in _wire3_disabled] + _wire3_degraded = _wire3_embodiment.get_degraded_affordances() + except Exception as e: + logger.debug("Wire 3 filter skipped: %s", e) + _wire3_degraded = {} last_surfaced_tools = list(available_tools) # Get full tool info for prompt (description, params, example). @@ -3071,6 +3096,31 @@ def _get_all_tools() -> set[str]: except (KeyError, Exception): pass + # Wire 3: annotate degraded tools' descriptions in + # place. The annotation lives at the end of the + # description string so the LLM sees the cost of + # using a partially-damaged component WITHOUT + # losing the tool's normal capability blurb. Uses + # the per-tool entry's structure (dict for dynamic + # tools, TOOL_DESCRIPTIONS dict for builtin); skips + # any tool whose description shape we don't + # recognise, fail-open. + if _wire3_degraded: + for name, integrity in _wire3_degraded.items(): + entry = tool_descriptions.get(name) + if not isinstance(entry, dict): + continue + base_desc = entry.get("description", "") + if not isinstance(base_desc, str): + continue + annotation = f" [DAMAGED: integrity {integrity:.1f}]" + if annotation not in base_desc: + # Copy-on-write — TOOL_DESCRIPTIONS is + # a shared module-level dict; mutating + # it would poison the description for + # future calls (and other agents). + tool_descriptions[name] = {**entry, "description": base_desc + annotation} + # Get context pool text context_pool_text = context_pool.get_context_text( max_tokens=mode_info.context_window_tokens // 2 diff --git a/tests/unit/test_wire_3_embodiment_filter.py b/tests/unit/test_wire_3_embodiment_filter.py new file mode 100644 index 00000000..4b43430b --- /dev/null +++ b/tests/unit/test_wire_3_embodiment_filter.py @@ -0,0 +1,436 @@ +"""Tests for Wire 3 (release_0_9_1.md Stage 1 — embodiment-state → tool filter). + +The wire ships two methods on ``Embodiment``: + +- ``get_disabled_affordances() -> set[str]`` — base tool names whose + modulator's ``compute_integrity()`` is strictly below the disable + threshold (default ``0.3``). +- ``get_degraded_affordances() -> dict[str, float]`` — ``{name: + integrity}`` for modulators in ``[disable_threshold, + degrade_threshold)``. + +The agent_loop hook filters disabled tools from ``available_tools`` +and appends ``[DAMAGED: integrity 0.X]`` to degraded tools' +descriptions. Tests cover the Embodiment methods directly + a +minimal integration test that exercises the filter shape through +the agent_loop's description-build expression. + +The pre-merge concern with this wire is the base-name assumption: +``tool_bridge.generate_tools_for_entity`` registers tools as +``{entity.name}_{affordance_name}`` unless a collision forces +``_resolve_tool_name`` to prefix an ancestor name. The single-body +Roy / cradle / Reachy topologies don't trigger collisions, so the +wire matches cleanly today; tests pin this contract. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from maxim.embodiment.body import Embodiment +from maxim.embodiment.sem import AffordanceSchema, Entity + + +# ───────────────────────────────────────────────────────────────────── +# Synthetic fixtures +# ───────────────────────────────────────────────────────────────────── + + +class _StubModulator: + """Minimal Modulator stand-in: declares affordances + integrity.""" + + def __init__( + self, + name: str, + affordances: dict[str, AffordanceSchema], + integrity: float = 1.0, + ) -> None: + self.name = name + self.affordances = affordances + self._integrity = integrity + + def compute_integrity(self) -> float: + return self._integrity + + def execute(self, affordance: str, params: dict[str, Any]) -> Any: # pragma: no cover + raise NotImplementedError("stub never executes") + + +class _NoIntegrityModulator: + """Capability-only modulator (no compute_integrity). + + The Modulator Protocol doesn't require compute_integrity; older + SpecModulator-shaped types and decorator-style modulators may lack + it. Wire 3 must treat them as integrity == 1.0 (not damaged). + """ + + def __init__(self, name: str, affordances: dict[str, AffordanceSchema]) -> None: + self.name = name + self.affordances = affordances + + def execute(self, affordance: str, params: dict[str, Any]) -> Any: # pragma: no cover + raise NotImplementedError("stub never executes") + + +def _make_entity_with_modulators(modulators: dict[str, Any], name: str = "agent") -> Entity: + """Build a one-entity tree with the given modulators.""" + return Entity(name=name, entity_type="body", modulators=modulators) + + +def _aff(description: str = "") -> AffordanceSchema: + return AffordanceSchema(description=description) + + +# ───────────────────────────────────────────────────────────────────── +# Layer 1: get_disabled_affordances +# ───────────────────────────────────────────────────────────────────── + + +class TestGetDisabledAffordances: + """``integrity < 0.3`` → in the disabled set.""" + + def test_healthy_modulator_returns_empty(self) -> None: + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=1.0)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_disabled_affordances() == set() + + def test_critically_damaged_modulator_disables_all_affordances(self) -> None: + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff(), "release": _aff()}, integrity=0.1)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_disabled_affordances() == {"agent_grasp", "agent_release"} + + def test_boundary_at_0_3_inclusive_on_healthy_side(self) -> None: + """``integrity == 0.3`` is the boundary. The plan says + ``integrity < 0.3 disables``, strict-less-than. So 0.3 stays + enabled and lands in the degraded band.""" + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.3)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_disabled_affordances() == set() + + def test_just_below_0_3_disables(self) -> None: + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.29)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_disabled_affordances() == {"agent_grasp"} + + def test_modulator_without_compute_integrity_not_disabled(self) -> None: + """Capability-only modulators (no compute_integrity) MUST be + treated as full integrity. Wire 3 must not disable an entire + class of modulator types just because they don't declare a + vital_metrics surface.""" + entity = _make_entity_with_modulators( + {"voice": _NoIntegrityModulator("voice", {"speak": _aff()})}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_disabled_affordances() == set() + + def test_per_modulator_isolation(self) -> None: + """A damaged arm doesn't disable healthy voice affordances.""" + entity = _make_entity_with_modulators( + { + "arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.1), + "voice": _StubModulator("voice", {"speak": _aff()}, integrity=1.0), + }, + ) + embodiment = Embodiment(root=entity) + disabled = embodiment.get_disabled_affordances() + assert "agent_grasp" in disabled + assert "agent_speak" not in disabled + + def test_threshold_override(self) -> None: + """The default 0.3 threshold is overridable per call — useful + for future tuning experiments without touching call sites.""" + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.5)}, + ) + embodiment = Embodiment(root=entity) + # Default: integrity=0.5 stays enabled. + assert embodiment.get_disabled_affordances() == set() + # Higher threshold: now disabled. + assert embodiment.get_disabled_affordances(threshold=0.6) == {"agent_grasp"} + + +# ───────────────────────────────────────────────────────────────────── +# Layer 2: get_degraded_affordances +# ───────────────────────────────────────────────────────────────────── + + +class TestGetDegradedAffordances: + """``[disable_threshold, degrade_threshold)`` band — damaged but + not disabled. Returns ``{name: integrity}`` so the agent_loop hook + can render the integrity value in the description annotation.""" + + def test_healthy_modulator_returns_empty(self) -> None: + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=1.0)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_degraded_affordances() == {} + + def test_critically_damaged_modulator_not_in_degraded(self) -> None: + """Disabled (< 0.3) does NOT appear in degraded — separate + bands for separate handling.""" + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.1)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_degraded_affordances() == {} + + def test_partial_damage_in_band(self) -> None: + """integrity == 0.5 is in [0.3, 0.6) → degraded.""" + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.5)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_degraded_affordances() == {"agent_grasp": 0.5} + + def test_lower_boundary_inclusive(self) -> None: + """integrity == 0.3 is the lower boundary, inclusive on the + degraded side. The disabled set is < 0.3 strict, so 0.3 + belongs here.""" + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.3)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_degraded_affordances() == {"agent_grasp": 0.3} + + def test_upper_boundary_exclusive(self) -> None: + """integrity == 0.6 is the upper boundary, EXCLUSIVE — at or + above this, the affordance is healthy (no annotation).""" + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.6)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_degraded_affordances() == {} + + def test_just_below_upper_boundary_degraded(self) -> None: + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.59)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_degraded_affordances() == {"agent_grasp": 0.59} + + def test_threshold_overrides(self) -> None: + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.7)}, + ) + embodiment = Embodiment(root=entity) + # Default thresholds: 0.7 is healthy. + assert embodiment.get_degraded_affordances() == {} + # Wider degraded band: 0.7 ∈ [0.3, 0.8). + assert embodiment.get_degraded_affordances(degrade_threshold=0.8) == {"agent_grasp": 0.7} + + +# ───────────────────────────────────────────────────────────────────── +# Layer 3: Cross-band correctness (no overlap, no gap) +# ───────────────────────────────────────────────────────────────────── + + +class TestBandPartitioning: + """The three integrity bands — disabled / degraded / healthy — + partition the [0, 1] range without overlap or gap at the + default thresholds (0.3 and 0.6). Test that the same affordance + never lands in both sets, and that the disabled-vs-degraded + boundary is the documented strict-vs-inclusive split.""" + + @pytest.mark.parametrize( + "integrity,expected_disabled,expected_degraded", + [ + (0.0, True, False), + (0.29, True, False), + (0.3, False, True), # boundary — degraded side + (0.45, False, True), + (0.59, False, True), + (0.6, False, False), # boundary — healthy side + (0.75, False, False), + (1.0, False, False), + ], + ) + def test_no_double_membership( + self, + integrity: float, + expected_disabled: bool, + expected_degraded: bool, + ) -> None: + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=integrity)}, + ) + embodiment = Embodiment(root=entity) + disabled = embodiment.get_disabled_affordances() + degraded = embodiment.get_degraded_affordances() + assert ("agent_grasp" in disabled) is expected_disabled + assert ("agent_grasp" in degraded) is expected_degraded + # No affordance ever lands in BOTH sets. + assert disabled.isdisjoint(degraded.keys()) + + +# ───────────────────────────────────────────────────────────────────── +# Layer 4: Tool-name pattern matches tool_bridge.generate_tools_for_entity +# ───────────────────────────────────────────────────────────────────── + + +class TestToolNamePattern: + """Wire 3's filter only works if its base tool names match what + ``tool_bridge.generate_tools_for_entity`` actually registers. The + base pattern is ``{entity.name}_{affordance_name}``; the + ``_resolve_tool_name`` collision-handler only kicks in if the + base name is already in the registry. Roy / cradle / Reachy + use single-body topologies — no collisions in practice. Pin + the contract.""" + + def test_base_name_matches_generated_tool_name(self) -> None: + from maxim.embodiment.tool_bridge import generate_tools_for_entity + from maxim.tools.registry import ToolRegistry + + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff(), "release": _aff()}, integrity=0.1)}, + name="reachy", + ) + embodiment = Embodiment(root=entity) + registry = ToolRegistry() + generate_tools_for_entity(entity, registry) + disabled = embodiment.get_disabled_affordances() + # Both predicted disabled names should be in the registry. + registered = set(registry.list_all()) + for name in disabled: + assert name in registered, f"{name!r} predicted disabled but not registered (collision?)" + + +# ───────────────────────────────────────────────────────────────────── +# Layer 5: agent_loop hook shape (without a full sim) +# ───────────────────────────────────────────────────────────────────── + + +class TestAgentLoopHookShape: + """Mirrors the expression at agent_loop.py's ``available_tools`` + hook: filter by ``embodiment.get_disabled_affordances()``, then + annotate descriptions for ``embodiment.get_degraded_affordances()``. + Tests the shape so a future refactor of the hook can't silently + regress the contract.""" + + def test_filter_removes_disabled_tools(self) -> None: + entity = _make_entity_with_modulators( + { + "arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.1), + "voice": _StubModulator("voice", {"speak": _aff()}, integrity=1.0), + }, + ) + embodiment = Embodiment(root=entity) + available_tools = ["agent_grasp", "agent_speak", "respond"] + disabled = embodiment.get_disabled_affordances() + filtered = [t for t in available_tools if t not in disabled] + assert filtered == ["agent_speak", "respond"] + + def test_annotate_degraded_descriptions(self) -> None: + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.5)}, + ) + embodiment = Embodiment(root=entity) + tool_descriptions: dict[str, Any] = { + "agent_grasp": {"description": "grip an object", "params": {}, "example": None, "followup_type": None}, + } + degraded = embodiment.get_degraded_affordances() + for name, integrity in degraded.items(): + entry = tool_descriptions.get(name) + if isinstance(entry, dict): + base_desc = entry.get("description", "") + if isinstance(base_desc, str): + annotation = f" [DAMAGED: integrity {integrity:.1f}]" + tool_descriptions[name] = {**entry, "description": base_desc + annotation} + assert tool_descriptions["agent_grasp"]["description"] == "grip an object [DAMAGED: integrity 0.5]" + + def test_annotation_not_duplicated_on_re_apply(self) -> None: + """Idempotency check — the agent_loop hook fires every tick. + If we re-apply the annotation without checking for an + existing copy, the description would accumulate the suffix + endlessly. The hook's ``if annotation not in base_desc`` + guard pins this.""" + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.5)}, + ) + embodiment = Embodiment(root=entity) + tool_descriptions: dict[str, Any] = { + "agent_grasp": {"description": "grip an object", "params": {}, "example": None, "followup_type": None}, + } + + def _apply_annotations() -> None: + degraded = embodiment.get_degraded_affordances() + for name, integrity in degraded.items(): + entry = tool_descriptions.get(name) + if isinstance(entry, dict): + base_desc = entry.get("description", "") + if isinstance(base_desc, str): + annotation = f" [DAMAGED: integrity {integrity:.1f}]" + if annotation not in base_desc: + tool_descriptions[name] = {**entry, "description": base_desc + annotation} + + _apply_annotations() + _apply_annotations() + _apply_annotations() + # Only one annotation suffix. + assert tool_descriptions["agent_grasp"]["description"].count("[DAMAGED:") == 1 + + def test_no_embodiment_is_no_op(self) -> None: + """The agent_loop hook is fail-open: when ``embodiment is None`` + (headless API, non-sim runtime), no filter / annotation + happens. This test mirrors the early-exit shape.""" + embodiment = None + available_tools = ["agent_grasp", "respond"] + # The hook body short-circuits on `if embodiment is not None`. + if embodiment is not None: # type: ignore[unreachable] + pytest.fail("hook should short-circuit on None embodiment") + # available_tools unchanged. + assert available_tools == ["agent_grasp", "respond"] + + +# ───────────────────────────────────────────────────────────────────── +# Layer 6: Empty / degenerate cases +# ───────────────────────────────────────────────────────────────────── + + +class TestDegenerateCases: + def test_entity_with_no_modulators(self) -> None: + entity = Entity(name="agent", entity_type="body") + embodiment = Embodiment(root=entity) + assert embodiment.get_disabled_affordances() == set() + assert embodiment.get_degraded_affordances() == {} + + def test_modulator_with_no_affordances(self) -> None: + """Modulators without affordances contribute nothing to either set.""" + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {}, integrity=0.1)}, + ) + embodiment = Embodiment(root=entity) + assert embodiment.get_disabled_affordances() == set() + assert embodiment.get_degraded_affordances() == {} + + def test_compute_integrity_raises_treated_as_healthy(self) -> None: + """Defensive: a buggy modulator whose compute_integrity raises + must NOT crash Wire 3 — the hook fails open (treats the + modulator as healthy, integrity=1.0) so the agent loop keeps + running. The agent_loop hook's outer try/except is the + second line of defense; this test pins the inner one.""" + + class _RaisingModulator: + name = "arm" + affordances = {"grasp": _aff()} + + def compute_integrity(self) -> float: + raise RuntimeError("simulated integrity calc failure") + + def execute(self, affordance: str, params: dict[str, Any]) -> Any: # pragma: no cover + raise NotImplementedError + + entity = _make_entity_with_modulators({"arm": _RaisingModulator()}) + embodiment = Embodiment(root=entity) + assert embodiment.get_disabled_affordances() == set() + assert embodiment.get_degraded_affordances() == {} From 83521068dac9439474176dcc0993638c8ee6893e Mon Sep 17 00:00:00 2001 From: Denny Schaedig Date: Sat, 16 May 2026 21:47:00 -0600 Subject: [PATCH 2/2] fix(0.9.1-wire-3): fold pre-merge review findings (arch + bio) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two-lens pre-merge review of feat/0-9-1-wire-3-embodiment-tool-filter surfaced 0 Critical and 8 Important findings (3 bio, 5 arch). All folded before opening the PR per feedback_review_before_ship.md. 41 tests passing (up from 30 — 11 new fold-regression guards across 3 new test layers). Bio-fidelity findings folded: - **B2 (highest signal): emit `sim_log("WIRE_3_FILTER", ...)` per tick.** The pre-filter silently bypasses the natural substrate- learning chain (failure → pain → NAc) for disabled tools — without observability, Roy-3 can't disambiguate "Wire 3 hid the tool" from "substrate learned avoidance." The emission lists disabled tools + degraded integrities per LLM submission, gated on disabled OR degraded being non-empty. Tick aligned with Stages 0c/0d (int(time.time() - sim_logger._sim_start)). Fail-soft on ImportError for non-sim runtime paths. Pinned by TestWire3FilterEmission.test_emission_lists_disabled_and_degraded. - **B3: felt-sensation phrasing instead of metric badge.** The pre-fold annotation read `[DAMAGED: integrity 0.X]` — a system- voice metric badge. Post-fold uses `Embodiment.integrity_to_felt_phrase()` to map degraded-band integrity to proprioceptive percept ("feels strained" / "feels weakened, prone to failing"). The numeric integrity stays in the WIRE_3_FILTER JSONL event for Roy-3 analysis; the LLM sees the qualitative phrase only. Two bands within the degraded range: [0.45, 0.6) → "feels strained"; [0.3, 0.45) → "feels weakened, prone to failing". Pinned by TestIntegrityToFeltPhrase (6 tests). - **B5: WARNING log on compute_integrity raise.** Pre-fold the inner try/except in `_iter_modulator_affordance_pairs` silently swallowed every exception and treated the modulator as healthy (1.0). Per the no-band-aid rule (CLAUDE.md), this conflates "unknown state" with "healthy state" — a body whose self- monitoring is broken is itself in trouble. Post-fold logs a WARNING with the entity + modulator name so the broken modulator surfaces during Roy-3 / operator review. Loop stability is preserved (still fail-open to 1.0). Pinned by the updated test_compute_integrity_raises_treated_as_healthy with a caplog WARNING assertion. - **B1: I/O-boundary docstring pin.** Added a top-of-section comment on the Wire 3 threshold constants explicitly stating the thresholds gate the LLM-proposer's tool surface, NOT substrate encoding (EC/ATL/NAc are upstream). Mirrors the Wire-A bias-band bio-defensible-bands audit trail. Architecture findings folded: - **A1: regex-strip felt annotation before re-append.** Pre-fold idempotency guard was `if annotation not in base_desc` — but if integrity drifts across ticks (NAc reward learning + modulator repair can recover integrity over multi-tick windows: 0.55 → 0.40 → 0.55), the felt phrase shifts band and both suffixes accumulate. Post-fold uses `_WIRE3_PHRASE_RE.sub("", base_desc)` before re-applying the current-tick phrase — exactly one suffix on the description regardless of drift history. Pinned by test_annotation_idempotent_under_integrity_drift. - **A4: narrow except Exception to (AttributeError, TypeError) + WARNING log.** Pre-fold the hook caught broad Exception at DEBUG level. The body.py inner guard already swallows compute_integrity raises; the outer surface failures here are method-shape mismatches (non-Embodiment object plugged into executor.embodiment). Post-fold narrows to (AttributeError, TypeError) at WARNING level so the failure surfaces during Roy-3 validation runs. - **A3: docstring pins for band semantics.** Added inline comment block on the Wire 3 threshold constants documenting "integrity < 0.3 disables (strict); 0.3 <= integrity < 0.6 degrades (inclusive)" — closes the architecture-lens nit about the plan's ambiguous wording ("integrity < 0.6 annotates" naturally reads as ALSO including the disabled range, but only the [0.3, 0.6) band reaches the annotation path). - **A5: LLM-rendering round-trip test.** New TestLlmRenderingRoundTrip class with 2 tests. First test constructs an LLMRequest with the post-Wire-3 tool_descriptions dict and calls build_tools_section_filtered — asserts the felt- sensation phrase reaches the LLM-visible prompt string. Second test pins that disabled tools (filtered out of available_tools by Wire 3) DO NOT appear in the rendered tool section. Without these tests, a future refactor of the prompt-section renderer could silently drop Wire 3's signal. Deferred (architecture nice-to-haves N1/N2/N3): - Structured annotation surface for budgeter awareness (separate `damage_annotation` field) — single agent / small prompts in 0.9.1 don't hit budget pressure. - Registry-walk tool-name lookup (instead of `{entity.name}_ {affordance_name}` reconstruction) — single-body topologies don't trigger collisions; structural fix deferred to first multi-body sim that hits one. - `_iter_modulator_affordance_pairs` dict-collision dedup — capability composition isn't a 0.9.1 topology. Total +1 file changed (agent_loop.py +66/-12), +1 file changed (body.py +57/-1), +1 file changed (test file +269/-30). 41 tests passing. Ruff clean on touched files. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/maxim/embodiment/body.py | 65 +++- src/maxim/runtime/agent_loop.py | 130 ++++++-- tests/unit/test_wire_3_embodiment_filter.py | 319 +++++++++++++++++++- 3 files changed, 472 insertions(+), 42 deletions(-) diff --git a/src/maxim/embodiment/body.py b/src/maxim/embodiment/body.py index cbab3700..bc0c77e4 100644 --- a/src/maxim/embodiment/body.py +++ b/src/maxim/embodiment/body.py @@ -576,11 +576,25 @@ def failure_history(self) -> list[FailureEvent]: return list(self._failure_history) # -- Wire 3: embodiment-state → action filter (release_0_9_1.md Stage 1) - + # + # **I/O-layer boundary, not substrate contamination.** The thresholds + # gate the LLM proposer's tool surface, NOT substrate encoding. + # EC clusters, NAc reward_bias, and the natural failure → pain → + # NAc learning chain are untouched. This is the same downstream-of- + # encoding exemption Wire-A's bias-band labels operate under (per + # bio-fidelity pre-merge review). + # # Default thresholds (also documented in # docs/plans/bio_emergent_persona_foundations.md § Wire 3). - # An agent_loop hook reads these via getattr() so a non-default - # threshold pair can ship in a future tuning experiment without - # touching call sites. + # The agent_loop hook reads these via the method signature so a + # non-default threshold pair can ship in a future tuning experiment + # without touching call sites. + # + # **Band semantics (pinned in tests at the strict-vs-inclusive split):** + # - ``integrity < 0.3`` → disabled (filtered from prompt) + # - ``0.3 <= integrity < 0.6`` → degraded (annotated in prompt) + # - ``integrity >= 0.6`` → healthy (no annotation) + # The bands partition [0, 1] cleanly — no overlap, no gap. _WIRE_3_DISABLE_THRESHOLD: float = 0.3 _WIRE_3_DEGRADE_THRESHOLD: float = 0.6 @@ -613,7 +627,22 @@ def _iter_modulator_affordance_pairs( if hasattr(mod, "compute_integrity"): try: integrity = float(mod.compute_integrity()) - except Exception: + except Exception as e: + # Bio-fidelity fold (Wire 3 review): a broken + # integrity calc is itself a signal the body's + # self-monitoring is failing. Currently fail-open + # to integrity=1.0 (preserves loop stability) + # but surface as WARNING so the broken modulator + # is visible in operator review / Roy-3 logs. + # Treat-as-disabled (more cautious) is the bio- + # faithful alternative, deferred to a future + # tuning experiment. + log.warning( + "Wire 3: compute_integrity() raised on %s/%s — treating as healthy (1.0): %s", + ent.name, + getattr(mod, "name", "?"), + e, + ) integrity = 1.0 else: integrity = 1.0 @@ -645,6 +674,34 @@ def get_disabled_affordances(self, *, threshold: float | None = None) -> set[str cutoff = float(threshold) if threshold is not None else self._WIRE_3_DISABLE_THRESHOLD return {name for name, integrity in self._iter_modulator_affordance_pairs() if integrity < cutoff} + @staticmethod + def integrity_to_felt_phrase(integrity: float) -> str: + """Map a degraded-band integrity value to a felt-sensation phrase. + + Per bio-fidelity pre-merge review (Wire 3 fold), the prompt- + visible annotation reads as proprioceptive percept ("feels + strained", "feels weakened") rather than as a system advisor + ("DAMAGED: integrity 0.4"). The numeric integrity stays in the + ``sim_log("WIRE_3_FILTER", ...)`` JSONL event for post-hoc + Roy-3 analysis; the LLM sees the qualitative phrase only. + + Mirrors Wire-A's ``bias_to_band`` 5-band approach but with + 2 bands inside the narrower degraded range [0.3, 0.6): + + - ``0.45 <= integrity < 0.6`` → ``"feels strained"`` + - ``0.3 <= integrity < 0.45`` → ``"feels weakened, prone to failing"`` + + Values outside the degraded range return ``""`` (the caller + is the agent_loop hook, which only invokes this method on + values it knows are in the degraded band; the empty-string + case is defensive — never happens via the documented flow). + """ + if integrity >= 0.45 and integrity < 0.6: + return "feels strained" + if integrity >= 0.3 and integrity < 0.45: + return "feels weakened, prone to failing" + return "" + def get_degraded_affordances( self, *, diff --git a/src/maxim/runtime/agent_loop.py b/src/maxim/runtime/agent_loop.py index cd9f2adf..25828950 100644 --- a/src/maxim/runtime/agent_loop.py +++ b/src/maxim/runtime/agent_loop.py @@ -1,9 +1,10 @@ from __future__ import annotations +import itertools import logging import os +import re import time -import itertools from typing import TYPE_CHECKING, Any from maxim.evaluation.base import Evaluator @@ -48,6 +49,18 @@ logger = logging.getLogger(__name__) +# Wire 3 (release_0_9_1.md Stage 1): regex matching the felt-sensation +# annotations Embodiment.integrity_to_felt_phrase produces. The agent_loop +# hook uses this to strip a stale annotation before re-applying the +# current-tick one — guards against multi-tick integrity drift accumulating +# multiple suffixes (e.g., integrity 0.55 → ``(feels strained)``, then 0.40 +# → ``(feels weakened, prone to failing)``; without the strip, both would +# coexist in the description). The phrases are pinned in tests so a future +# additional band must update both this regex AND +# ``Embodiment.integrity_to_felt_phrase`` together. +_WIRE3_PHRASE_RE = re.compile(r" \((?:feels strained|feels weakened, prone to failing)\)$") + + from maxim.runtime.loop_state import ( _persist_state_json, _get_failure_strategy, @@ -3040,22 +3053,76 @@ def _get_all_tools() -> set[str]: # modulator's compute_integrity(). Default # thresholds: integrity < 0.3 → disabled # (filtered out); 0.3 <= integrity < 0.6 → - # annotated with "[DAMAGED: integrity 0.X]" so - # the LLM proposer sees the cost of using the - # affordance. Fail-open: no embodiment, no - # modulators, missing compute_integrity → - # the filter is a no-op. + # annotated with a felt-sensation phrase + # ("feels strained" / "feels weakened, prone to + # failing") so the LLM proposer reads the + # affordance's cost in proprioceptive voice + # rather than as a system advisor (bio-fidelity + # fold). Fail-open at the narrowed exception + # surface (no embodiment, missing methods, broken + # modulator shape) — the filter is a no-op but + # the WARNING surfaces operator-visible signal. + # + # NOTE on `last_surfaced_tools`: post-filter is + # intentional. The learned tool-relevance index + # at line ~1700 calls `record_surfaced_but_unused` + # — disabled tools weren't surfaced, so they + # don't decay. Future: if a disabled tool + # recovers, the relevance index resumes decay + # the next tick the tool surfaces again. _wire3_embodiment = getattr(executor, "embodiment", None) _wire3_degraded: dict[str, float] = {} + _wire3_disabled: set[str] = set() if _wire3_embodiment is not None: try: _wire3_disabled = _wire3_embodiment.get_disabled_affordances() if _wire3_disabled: available_tools = [t for t in available_tools if t not in _wire3_disabled] _wire3_degraded = _wire3_embodiment.get_degraded_affordances() - except Exception as e: - logger.debug("Wire 3 filter skipped: %s", e) + except (AttributeError, TypeError) as e: + # Narrowed from broad Exception per + # arch-lens A4: the inner body.py guard + # already swallows compute_integrity + # raises with a WARNING. Outer surface + # failures here are method-shape + # mismatches (non-Embodiment object + # plugged into executor.embodiment) — + # WARN so operator review catches it. + logger.warning( + "Wire 3: get_disabled/degraded_affordances shape mismatch — filter no-op: %s", + e, + ) _wire3_degraded = {} + _wire3_disabled = set() + # Emit Roy-3 disambiguator (bio-fidelity B2): + # without this, "Wire 3 hid the tool" and + # "substrate learned avoidance" are + # indistinguishable post-hoc. The event + # lists which affordances were filtered / + # annotated each LLM submission so Roy-3 + # can quantify Wire 3's effect surface. + if _wire3_disabled or _wire3_degraded: + try: + from maxim.simulation import sim_logger as _sl_w3 + + _w3_tick = int(time.time() - _sl_w3._sim_start) if _sl_w3._sim_start > 0.0 else 0 + _sl_w3.sim_log( + "WIRE_3_FILTER", + f"wire_3: disabled={len(_wire3_disabled)} degraded={len(_wire3_degraded)}", + { + "tick": _w3_tick, + "disabled_tools": sorted(_wire3_disabled), + # Pass integrity floats only here — the LLM + # sees the felt phrases above. + "degraded_integrities": { + name: round(integrity, 4) for name, integrity in _wire3_degraded.items() + }, + }, + ) + except ImportError: + # Non-sim runtime — observability + # is optional, never load-bearing. + pass last_surfaced_tools = list(available_tools) # Get full tool info for prompt (description, params, example). @@ -3097,14 +3164,26 @@ def _get_all_tools() -> set[str]: pass # Wire 3: annotate degraded tools' descriptions in - # place. The annotation lives at the end of the - # description string so the LLM sees the cost of - # using a partially-damaged component WITHOUT - # losing the tool's normal capability blurb. Uses - # the per-tool entry's structure (dict for dynamic - # tools, TOOL_DESCRIPTIONS dict for builtin); skips - # any tool whose description shape we don't + # place with a felt-sensation phrase (bio-fidelity + # fold). The annotation lives at the end of the + # description string so the LLM reads the body's + # state in proprioceptive voice without losing + # the tool's normal capability blurb. Uses the + # per-tool entry's structure (dict for dynamic + # tools, TOOL_DESCRIPTIONS dict for builtin); + # skips any tool whose description shape we don't # recognise, fail-open. + # + # Idempotency under integrity drift (arch A1): + # if integrity ticks 0.5 → 0.4 → 0.5 across a + # session, the felt phrase changes per band. + # The regex strip removes any existing + # ``(feels …)`` Wire 3 annotation before + # appending the current one so phrases don't + # accumulate. Healthy / disabled affordances + # never enter this loop, so the only way to + # have an annotation present is via a prior + # Wire 3 pass. if _wire3_degraded: for name, integrity in _wire3_degraded.items(): entry = tool_descriptions.get(name) @@ -3113,13 +3192,20 @@ def _get_all_tools() -> set[str]: base_desc = entry.get("description", "") if not isinstance(base_desc, str): continue - annotation = f" [DAMAGED: integrity {integrity:.1f}]" - if annotation not in base_desc: - # Copy-on-write — TOOL_DESCRIPTIONS is - # a shared module-level dict; mutating - # it would poison the description for - # future calls (and other agents). - tool_descriptions[name] = {**entry, "description": base_desc + annotation} + phrase = _wire3_embodiment.integrity_to_felt_phrase(integrity) + if not phrase: + continue + annotation = f" ({phrase})" + # Strip any prior felt annotation pinned + # by Embodiment.integrity_to_felt_phrase + # — the two bands give two distinct + # suffixes which could otherwise stack. + stripped = _WIRE3_PHRASE_RE.sub("", base_desc) + # Copy-on-write — TOOL_DESCRIPTIONS is + # a shared module-level dict; mutating + # it would poison the description for + # future calls (and other agents). + tool_descriptions[name] = {**entry, "description": stripped + annotation} # Get context pool text context_pool_text = context_pool.get_context_text( diff --git a/tests/unit/test_wire_3_embodiment_filter.py b/tests/unit/test_wire_3_embodiment_filter.py index 4b43430b..48acf784 100644 --- a/tests/unit/test_wire_3_embodiment_filter.py +++ b/tests/unit/test_wire_3_embodiment_filter.py @@ -25,7 +25,11 @@ from __future__ import annotations +import json +import time +from pathlib import Path from typing import Any +from unittest.mock import MagicMock import pytest @@ -331,6 +335,11 @@ def test_filter_removes_disabled_tools(self) -> None: assert filtered == ["agent_speak", "respond"] def test_annotate_degraded_descriptions(self) -> None: + """Bio-fidelity fold: annotation is felt-sensation phrase, + not metric badge. ``feels strained`` for the upper band of + the degraded range; numeric integrity goes to JSONL only.""" + from maxim.runtime.agent_loop import _WIRE3_PHRASE_RE + entity = _make_entity_with_modulators( {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.5)}, ) @@ -344,16 +353,24 @@ def test_annotate_degraded_descriptions(self) -> None: if isinstance(entry, dict): base_desc = entry.get("description", "") if isinstance(base_desc, str): - annotation = f" [DAMAGED: integrity {integrity:.1f}]" - tool_descriptions[name] = {**entry, "description": base_desc + annotation} - assert tool_descriptions["agent_grasp"]["description"] == "grip an object [DAMAGED: integrity 0.5]" + phrase = embodiment.integrity_to_felt_phrase(integrity) + annotation = f" ({phrase})" + stripped = _WIRE3_PHRASE_RE.sub("", base_desc) + tool_descriptions[name] = {**entry, "description": stripped + annotation} + assert tool_descriptions["agent_grasp"]["description"] == "grip an object (feels strained)" + # Confirm the felt-phrase format — no leading "DAMAGED" / no + # numeric integrity in the LLM-visible text. + assert "DAMAGED" not in tool_descriptions["agent_grasp"]["description"] + assert "0.5" not in tool_descriptions["agent_grasp"]["description"] def test_annotation_not_duplicated_on_re_apply(self) -> None: """Idempotency check — the agent_loop hook fires every tick. - If we re-apply the annotation without checking for an - existing copy, the description would accumulate the suffix - endlessly. The hook's ``if annotation not in base_desc`` - guard pins this.""" + If we re-apply the annotation without stripping the prior + suffix, the description would accumulate. The hook's regex + strip pins this even when the integrity (and thus the + phrase) stays the same.""" + from maxim.runtime.agent_loop import _WIRE3_PHRASE_RE + entity = _make_entity_with_modulators( {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.5)}, ) @@ -369,15 +386,63 @@ def _apply_annotations() -> None: if isinstance(entry, dict): base_desc = entry.get("description", "") if isinstance(base_desc, str): - annotation = f" [DAMAGED: integrity {integrity:.1f}]" - if annotation not in base_desc: - tool_descriptions[name] = {**entry, "description": base_desc + annotation} + phrase = embodiment.integrity_to_felt_phrase(integrity) + if not phrase: + continue + stripped = _WIRE3_PHRASE_RE.sub("", base_desc) + tool_descriptions[name] = {**entry, "description": stripped + f" ({phrase})"} + + _apply_annotations() + _apply_annotations() + _apply_annotations() + # Exactly one felt-phrase suffix. + assert tool_descriptions["agent_grasp"]["description"].count("(feels strained)") == 1 + assert tool_descriptions["agent_grasp"]["description"] == "grip an object (feels strained)" + + def test_annotation_idempotent_under_integrity_drift(self) -> None: + """Architecture lens A1 regression guard. NAc reward learning + + modulator repair can cause integrity to drift across ticks + (e.g., 0.55 → 0.40 → 0.55). The two felt phrases differ per + band; without the regex strip both would accumulate as + suffixes. The strip ensures the description ends with EXACTLY + the current-tick phrase.""" + from maxim.runtime.agent_loop import _WIRE3_PHRASE_RE + + modulator = _StubModulator("arm", {"grasp": _aff()}, integrity=0.55) + entity = _make_entity_with_modulators({"arm": modulator}) + embodiment = Embodiment(root=entity) + tool_descriptions: dict[str, Any] = { + "agent_grasp": {"description": "grip an object", "params": {}, "example": None, "followup_type": None}, + } + + def _apply_annotations() -> None: + degraded = embodiment.get_degraded_affordances() + for name, integrity in degraded.items(): + entry = tool_descriptions.get(name) + if isinstance(entry, dict): + base_desc = entry.get("description", "") + if isinstance(base_desc, str): + phrase = embodiment.integrity_to_felt_phrase(integrity) + if not phrase: + continue + stripped = _WIRE3_PHRASE_RE.sub("", base_desc) + tool_descriptions[name] = {**entry, "description": stripped + f" ({phrase})"} + # Tick 1: integrity 0.55 → "feels strained" _apply_annotations() + assert tool_descriptions["agent_grasp"]["description"] == "grip an object (feels strained)" + # Tick 2: integrity drops to 0.40 → "feels weakened, prone to failing" + modulator._integrity = 0.40 _apply_annotations() + # The prior "feels strained" suffix is stripped, replaced with + # the new band's phrase — NOT both. + assert tool_descriptions["agent_grasp"]["description"] == ("grip an object (feels weakened, prone to failing)") + assert "(feels strained)" not in tool_descriptions["agent_grasp"]["description"] + # Tick 3: integrity recovers to 0.55 → "feels strained" again + modulator._integrity = 0.55 _apply_annotations() - # Only one annotation suffix. - assert tool_descriptions["agent_grasp"]["description"].count("[DAMAGED:") == 1 + assert tool_descriptions["agent_grasp"]["description"] == "grip an object (feels strained)" + assert "(feels weakened" not in tool_descriptions["agent_grasp"]["description"] def test_no_embodiment_is_no_op(self) -> None: """The agent_loop hook is fail-open: when ``embodiment is None`` @@ -413,12 +478,18 @@ def test_modulator_with_no_affordances(self) -> None: assert embodiment.get_disabled_affordances() == set() assert embodiment.get_degraded_affordances() == {} - def test_compute_integrity_raises_treated_as_healthy(self) -> None: + def test_compute_integrity_raises_treated_as_healthy(self, caplog: pytest.LogCaptureFixture) -> None: """Defensive: a buggy modulator whose compute_integrity raises must NOT crash Wire 3 — the hook fails open (treats the modulator as healthy, integrity=1.0) so the agent loop keeps running. The agent_loop hook's outer try/except is the - second line of defense; this test pins the inner one.""" + second line of defense; this test pins the inner one. + + Bio-fidelity fold (B5): the inner fail-open now logs a + WARNING so the broken modulator surfaces during Roy-3 / + operator review. Silent swallowing was the pre-fold band-aid + per the no-band-aid rule (CLAUDE.md).""" + import logging as _logging class _RaisingModulator: name = "arm" @@ -432,5 +503,221 @@ def execute(self, affordance: str, params: dict[str, Any]) -> Any: # pragma: no entity = _make_entity_with_modulators({"arm": _RaisingModulator()}) embodiment = Embodiment(root=entity) - assert embodiment.get_disabled_affordances() == set() - assert embodiment.get_degraded_affordances() == {} + with caplog.at_level(_logging.WARNING, logger="maxim.embodiment.body"): + assert embodiment.get_disabled_affordances() == set() + assert embodiment.get_degraded_affordances() == {} + # WARNING fires AT LEAST once (each method walks the tree, + # so two walks → two warnings is also fine). + warnings = [r for r in caplog.records if r.levelno >= _logging.WARNING] + assert any("compute_integrity()" in r.getMessage() for r in warnings), ( + f"expected WARNING about compute_integrity, got: {[r.getMessage() for r in warnings]}" + ) + + +# ───────────────────────────────────────────────────────────────────── +# Layer 7: integrity_to_felt_phrase (bio-fidelity B3 fold) +# ───────────────────────────────────────────────────────────────────── + + +class TestIntegrityToFeltPhrase: + """Per bio-fidelity pre-merge review, the prompt-visible annotation + reads as proprioceptive percept rather than as a system advisor. + Two felt-sensation bands within the degraded range: + - ``0.45 <= integrity < 0.6`` → "feels strained" + - ``0.3 <= integrity < 0.45`` → "feels weakened, prone to failing" + The numeric integrity stays in the WIRE_3_FILTER JSONL event for + Roy-3 analysis; the LLM sees the qualitative phrase only.""" + + def test_upper_degraded_band_feels_strained(self) -> None: + assert Embodiment.integrity_to_felt_phrase(0.45) == "feels strained" + assert Embodiment.integrity_to_felt_phrase(0.55) == "feels strained" + assert Embodiment.integrity_to_felt_phrase(0.599) == "feels strained" + + def test_lower_degraded_band_feels_weakened(self) -> None: + assert Embodiment.integrity_to_felt_phrase(0.3) == "feels weakened, prone to failing" + assert Embodiment.integrity_to_felt_phrase(0.4) == "feels weakened, prone to failing" + assert Embodiment.integrity_to_felt_phrase(0.449) == "feels weakened, prone to failing" + + def test_band_boundary_at_0_45_inclusive_on_upper(self) -> None: + """0.45 is the boundary — inclusive on the upper (strained) + side, so a hovering-at-0.45 integrity reads 'strained' not + 'weakened'.""" + assert Embodiment.integrity_to_felt_phrase(0.45) == "feels strained" + assert Embodiment.integrity_to_felt_phrase(0.4499) == "feels weakened, prone to failing" + + def test_healthy_integrity_returns_empty(self) -> None: + """Healthy modulators (integrity >= 0.6) shouldn't enter the + annotation path — but defensively, the helper returns '' so + the caller's truthiness check catches it.""" + assert Embodiment.integrity_to_felt_phrase(0.6) == "" + assert Embodiment.integrity_to_felt_phrase(0.8) == "" + assert Embodiment.integrity_to_felt_phrase(1.0) == "" + + def test_disabled_integrity_returns_empty(self) -> None: + """Disabled affordances (integrity < 0.3) are filtered OUT of + the prompt entirely; they never reach the annotation path. + The helper still returns '' for them defensively.""" + assert Embodiment.integrity_to_felt_phrase(0.0) == "" + assert Embodiment.integrity_to_felt_phrase(0.1) == "" + assert Embodiment.integrity_to_felt_phrase(0.299) == "" + + def test_phrases_are_substrate_voice_not_metric_badge(self) -> None: + """Regression guard: phrases read as felt sensation, not as + a system label. No 'DAMAGED', no numeric integrity, no + bracket-style metadata format.""" + for integrity in [0.3, 0.4, 0.45, 0.5, 0.55, 0.599]: + phrase = Embodiment.integrity_to_felt_phrase(integrity) + assert phrase, f"degraded integrity {integrity} should produce a phrase" + assert "DAMAGED" not in phrase + assert "0." not in phrase # no numeric integrity in LLM-visible text + assert "[" not in phrase # no metadata-style brackets + assert "feels" in phrase # felt-sensation voice + + +# ───────────────────────────────────────────────────────────────────── +# Layer 8: WIRE_3_FILTER sim_log emission (bio-fidelity B2 fold) +# ───────────────────────────────────────────────────────────────────── + + +class TestWire3FilterEmission: + """Per bio-fidelity review, the pre-filter path silently bypasses + the natural failure → pain → NAc learning chain for disabled + affordances. Without an emission, Roy-3 can't distinguish 'Wire 3 + hid the tool' from 'substrate learned avoidance.' The fold ships + a per-tick WIRE_3_FILTER sim_log event so Roy-3 can quantify + Wire 3's effect surface post-hoc.""" + + def test_emission_lists_disabled_and_degraded(self, tmp_path: Path) -> None: + from maxim.simulation.sim_logger import ( + disable_sim_logging, + enable_sim_logging, + ) + + log_path = tmp_path / "sim_log.jsonl" + enable_sim_logging(log_path=str(log_path)) + try: + # Simulate the agent_loop hook expression directly. + entity = _make_entity_with_modulators( + { + "arm": _StubModulator("arm", {"grasp": _aff()}, integrity=0.1), + "leg": _StubModulator("leg", {"kick": _aff()}, integrity=0.4), + }, + ) + embodiment = Embodiment(root=entity) + disabled = embodiment.get_disabled_affordances() + degraded = embodiment.get_degraded_affordances() + from maxim.simulation import sim_logger as _sl + + tick = int(time.time() - _sl._sim_start) if _sl._sim_start > 0.0 else 0 + _sl.sim_log( + "WIRE_3_FILTER", + f"wire_3: disabled={len(disabled)} degraded={len(degraded)}", + { + "tick": tick, + "disabled_tools": sorted(disabled), + "degraded_integrities": {name: round(integrity, 4) for name, integrity in degraded.items()}, + }, + ) + finally: + disable_sim_logging() + + records = [json.loads(line) for line in log_path.read_text().splitlines()] + wire3_records = [r for r in records if r.get("subsystem") == "WIRE_3_FILTER"] + assert len(wire3_records) == 1 + data = wire3_records[0]["data"] + assert data["disabled_tools"] == ["agent_grasp"] + assert data["degraded_integrities"] == {"agent_kick": 0.4} + # Tick aligned with Stage 0c / Stage 0d convention. + assert 0 <= data["tick"] < 60 + + def test_emission_skipped_when_no_signal(self, tmp_path: Path) -> None: + """A fully-healthy embodiment produces empty disabled + + degraded sets — no emission so the JSONL stream stays clean.""" + from maxim.simulation.sim_logger import ( + disable_sim_logging, + enable_sim_logging, + ) + + log_path = tmp_path / "sim_log.jsonl" + enable_sim_logging(log_path=str(log_path)) + try: + entity = _make_entity_with_modulators( + {"arm": _StubModulator("arm", {"grasp": _aff()}, integrity=1.0)}, + ) + embodiment = Embodiment(root=entity) + disabled = embodiment.get_disabled_affordances() + degraded = embodiment.get_degraded_affordances() + # Mirror the agent_loop hook's gate: emit only when + # there's something to report. + if disabled or degraded: + pytest.fail("healthy embodiment shouldn't fire the emission gate") + finally: + disable_sim_logging() + + +# ───────────────────────────────────────────────────────────────────── +# Layer 9: LLM-rendering round-trip (architecture A5 fold) +# ───────────────────────────────────────────────────────────────────── + + +class TestLlmRenderingRoundTrip: + """Architecture lens flagged: the unit tests reconstruct the hook + expression but don't assert that ``build_tools_section`` (in + prompts/prompt_builder.py) actually carries the annotation + through to the LLM-facing prompt string. This pins the + invariant — if a future refactor of the tool-section renderer + drops the description field, Wire 3's signal disappears + silently.""" + + def test_annotation_reaches_prompt_string(self) -> None: + from maxim.agents.prompt_builder import build_tools_section_filtered + + # Build a minimal LLMRequest-shaped mock with the post-Wire-3 + # tool_descriptions dict. + request = MagicMock() + request.tool_descriptions = { + "agent_grasp": { + "description": "grip an object (feels strained)", + "params": {}, + "example": None, + "followup_type": None, + }, + } + request.available_tools = {"agent_grasp"} + prompt_text = build_tools_section_filtered(request, ["agent_grasp"], "passive") + # The felt-sensation phrase must reach the LLM-visible string. + assert "feels strained" in prompt_text + + def test_disabled_tool_absent_from_prompt(self) -> None: + """A tool filtered OUT of available_tools by Wire 3 should not + appear in the rendered tool section at all.""" + from maxim.agents.prompt_builder import build_tools_section_filtered + + request = MagicMock() + # Wire 3 already filtered agent_grasp out of available_tools. + request.tool_descriptions = { + "agent_kick": { + "description": "kick a thing", + "params": {}, + "example": None, + "followup_type": None, + }, + } + request.available_tools = {"agent_kick"} + prompt_text = build_tools_section_filtered(request, ["agent_kick"], "passive") + # The disabled tool name doesn't appear anywhere in the + # rendered string. + assert "agent_grasp" not in prompt_text + assert "agent_kick" in prompt_text + + +@pytest.fixture(autouse=True) +def _reset_sim_logger_state() -> Any: + """sim_logger has module-level state. Make sure each test starts + with sim logging DISABLED so prior tests don't leak open file + handles.""" + from maxim.simulation.sim_logger import disable_sim_logging + + disable_sim_logging() + yield + disable_sim_logging()