P302: verify rules and contact parity through live owners

plans/commander-native-plan.md

@@ -45,7 +45,7 @@
 | 12 | What the commander has not judged yet | `judge.py unjudged`; `vibe_watch.py` | `comms_unjudged`: calls and unit messages after the last verdict | `review_queue` |
 | 13 | Public story and roasts | `fleet-narrative.md`; `judge.py roast` | `comms_order_story`, `comms_fleet_story`; send-back reason posted once | `story` |
 | 14 | Heartbeat: next unit every minute | `rotate.py` (60 s loop) | Core-owned scheduled commander wake | `control_plane` |
-| 15 | Standing rules broken right now | `rotate.py broken_rules` | `StandingRule` members derive violations on read | `control_plane` |
+| 15 | Standing rules broken right now | `CommanderReview.wake` (replaces `rotate.py broken_rules`) | `StandingRule` members derive violations on read | `control_plane` |
 | 16 | Rule review | `rules.py`; `rules.jsonl`; maximum age | Rule instruction segments in the commander's turn context, with provenance | `control_plane` |
 | 17 | Load and forking | `agent-comms-commander capacity`: load and live fork reports; ten-minute beat age | Reporting-only `commander.capacity.Load.of`; no capacity refusal or automatic fork policy | `control_plane` |
 | 18 | Growth alarm | `StandingRule.broken` growth delivery (formerly `growth_watch.py`) | `commander.standing.MergeGrowth.violations`: captured first-parent growth | `control_plane` |

src/agent_comms/commander/capacity.py

@@ -22,14 +22,17 @@ class Load(ToolResult):
     def of(cls, comms: Comms, commander: str) -> Load:
         snapshot, todos, now = comms.registry.snapshot(), comms.goals.todos, time.time()
         owner = snapshot.require(commander)
-        owed = tuple(t for t in todos.held_for_commander(owner.incarnation) if t.pending)
+        held = todos.held_for_commander(owner.incarnation)
+        owed = tuple(t for t in held if t.pending)
         forks = tuple(sorted(
             thread.name for thread in snapshot.threads.values()
             if thread.parent == owner.name and snapshot.statuses[thread.name].receiving
             and todos.held_for_commander(thread.incarnation)
         ))
-        return cls(len(owed), sum(t.overdue(now) for t in owed),
-                   OrderEditCountRead.count(comms, owner.name), forks)
+        unjudged = sum(
+            read.read() for read in OrderEditCountRead.for_orders(comms.root, snapshot, held)
+        )
+        return cls(len(owed), sum(t.overdue(now) for t in owed), unjudged, forks)
 
     @projected(view="wire")
     def total(self) -> float:

src/agent_comms/commander_parity/load.py

@@ -2,42 +2,18 @@
 
 from __future__ import annotations
 
-import subprocess
+import asyncio
 from dataclasses import dataclass
 from pathlib import Path
+from typing import ClassVar
 
+from ..child_process import BoundedRun
 from ..commander.config import CommanderConfig
 from ..field_codec import FieldCodec
+from ..launch_runtime import LaunchRuntime
 from ..private_path import FileRevision
 from . import ParityComparison, UnverifiedParity
-
-# Profile the real return, never replace the measurement, command, bus or reply.
-# Comparing independent reads of a running fleet confuses new edits with drift.
-OBSERVE_CAPACITY = """
-import contextlib, io, json, sys
-import agent_comms.commander.capacity as capacity
-from agent_comms.commander.cli import main
-measured = []
-contexts = []
-def observe(frame, event, value):
-    if event == 'return' and frame.f_code is capacity.Load.of.__func__.__code__:
-        measured.append(value)
-    if event == 'call' and frame.f_code is capacity.main.__code__:
-        contexts.append(frame.f_locals['ctx'])
-output = io.StringIO()
-sys.setprofile(observe)
-try:
-    with contextlib.redirect_stdout(output):
-        main()
-finally:
-    sys.setprofile(None)
-if len(measured) != 1 or len(contexts) != 1:
-    raise ValueError('Capacity must use exactly one native measurement and selected context')
-config = contexts[0]
-print(json.dumps(dict(current=output.getvalue(), native=str(measured[0]),
-                      source=capacity.__file__, root=str(config.root),
-                      commander=config.commander)))
-"""
+from .observations import CapacityObservation, MeasuredOutput
 
 
 def state_files(directory: Path) -> tuple[tuple[str, FileRevision], ...]:
@@ -49,16 +25,9 @@ def state_files(directory: Path) -> tuple[tuple[str, FileRevision], ...]:
 
 
 @dataclass(frozen=True)
-class MeasuredOutput:
-    current: str
-    native: str
-    source: str
-    root: str
-    commander: str
-
-
-@dataclass(frozen=True)
-class LoadComparison(ParityComparison):
+class CommandComparison(ParityComparison):
+    arguments: ClassVar[tuple[str, ...]]
+    timeout: ClassVar[int] = 30
     observation: MeasuredOutput
     state_before: tuple[tuple[str, FileRevision], ...]
     state_after: tuple[tuple[str, FileRevision], ...]
@@ -80,18 +49,27 @@ def describe(self) -> str:
 
     @classmethod
     def read(cls) -> ParityComparison:
-        config = CommanderConfig.read(CommanderConfig.path())
+        return cls.invoke(CommanderConfig.read(CommanderConfig.path()), cls.arguments)
+
+    @classmethod
+    def invoke(cls, config: CommanderConfig, arguments: tuple[str, ...]) -> ParityComparison:
+        """Bind one native invocation and all its observations to the selected configuration."""
         before = state_files(config.state)
         # -I uses installed imports even when this candidate runs with PYTHONPATH=src.
         # HOME and the environment stay unchanged; the actual CLI main runs.
-        current = subprocess.run(
-            (str(config.cli.parent / "python"), "-I", "-c", OBSERVE_CAPACITY, "capacity"),
-            capture_output=True,
-            text=True,
-            timeout=30,
+        runtime = LaunchRuntime.running()
+        current = asyncio.run(
+            BoundedRun.run(
+                (str(runtime.interpreter), "-I", *arguments),
+                timeout=cls.timeout,
+                runtime=runtime,
+            )
         )
-        if current.returncode:
-            return UnverifiedParity(current.stderr)
+        if not current.outcome.successful:
+            return UnverifiedParity(
+                f"{cls.__name__}: configured native owner {current.outcome.summary}; "
+                f"bound={cls.timeout}s; {current.stderr.decode(errors='replace')}"
+            )
         observed = FieldCodec.decode(MeasuredOutput, FieldCodec.parse(current.stdout))
         source = Path(observed.source)
         if (
@@ -103,3 +81,7 @@ def read(cls) -> ParityComparison:
                 "The installed command used a different configuration or source"
             )
         return cls(observed, before, state_files(config.state))
+
+
+class LoadComparison(CommandComparison):
+    arguments = CapacityObservation.arguments()

src/agent_comms/commander_parity/observations.py

@@ -0,0 +1,219 @@
+"""Observe real native command returns and their actual output or stored publication."""
+
+from __future__ import annotations
+
+import argparse
+import io
+import json
+import sys
+from abc import abstractmethod
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager, redirect_stdout
+from dataclasses import dataclass
+from pathlib import Path
+from types import CodeType, MethodType
+from typing import TypeGuard
+
+import agent_comms.standing_rules as native_rules
+from agent_comms.commander import capacity
+from agent_comms.commander.cli import main as commander_main
+from agent_comms.commander.cli import tool_name
+from agent_comms.commander.config import CommanderConfig
+from agent_comms.commander.order import Commander
+from agent_comms.commander.send_guard import CommanderSession, RawSendCommand
+from agent_comms.commander_heartbeat import CommanderReview
+from agent_comms.comms import Comms
+from agent_comms.declared_family import DeclaredFamily
+from agent_comms.field_codec import FieldCodec
+from agent_comms.native_tools import NativeToolDeclaration
+from agent_comms.turn_context import InstructionSegment
+
+
+@dataclass(frozen=True)
+class MeasuredOutput:
+    current: str
+    native: str
+    source: str
+    root: str
+    commander: str
+
+
+class NativeObservation(DeclaredFamily, affix="Observation"):
+    @classmethod
+    def arguments(cls) -> tuple[str, ...]:
+        return (str(Path(__file__).resolve()), cls.declared_name)
+
+    @abstractmethod
+    def read(self, root: Path) -> MeasuredOutput: ...
+
+
+def method_code(method: object) -> CodeType:
+    if not isinstance(method, MethodType):
+        raise TypeError("An observation requires a bound Python method")
+    return method.__func__.__code__
+
+
+@contextmanager
+def returned_values(callbacks: dict[CodeType, Callable[[object], None]]) -> Iterator[None]:
+    """Observe only declared native returns, not every Python call in the command."""
+    monitor = sys.monitoring
+    tool = monitor.PROFILER_ID
+
+    def returned(code: CodeType, offset: int, value: object) -> None:
+        callbacks[code](value)
+
+    monitor.use_tool_id(tool, "commander-parity-observation")
+    try:
+        monitor.register_callback(tool, monitor.events.PY_RETURN, returned)
+        for code in callbacks:
+            monitor.set_local_events(tool, code, monitor.events.PY_RETURN)
+        yield
+    finally:
+        for code in callbacks:
+            monitor.set_local_events(tool, code, 0)
+        monitor.register_callback(tool, monitor.events.PY_RETURN, None)
+        monitor.free_tool_id(tool)
+
+
+class CapacityObservation(NativeObservation):
+    def read(self, root: Path) -> MeasuredOutput:
+        measured: list[capacity.Load] = []
+        contexts: list[Commander] = []
+
+        def measurement(value: object) -> None:
+            if not isinstance(value, capacity.Load):
+                raise TypeError("Load.of returned outside its declared contract")
+            measured.append(value)
+
+        def context(value: object) -> None:
+            if not isinstance(value, Commander):
+                raise TypeError("The commander CLI returned an undeclared context")
+            contexts.append(value)
+
+        output = io.StringIO()
+        # This process owns its command line. The actual CLI selects its root and command.
+        sys.argv[1:] = ["--root", str(root), tool_name(capacity.__name__)]
+        with (
+            returned_values(
+                {
+                    method_code(capacity.Load.of): measurement,
+                    method_code(Commander.read_root): context,
+                }
+            ),
+            redirect_stdout(output),
+        ):
+            commander_main()
+        if len(measured) != 1 or len(contexts) != 1:
+            raise ValueError("Capacity must use one native measurement and selected context")
+        selected = contexts[0]
+        return MeasuredOutput(
+            output.getvalue(),
+            str(measured[0]),
+            str(capacity.__file__),
+            str(selected.comms.root),
+            selected.commander,
+        )
+
+
+def standing_rule_return(value: object) -> TypeGuard[tuple[native_rules.StandingRule, ...]]:
+    """Validate CPython's untyped return, without choosing rule variants."""
+    return isinstance(value, tuple) and all(
+        isinstance(rule, native_rules.StandingRule) for rule in value
+    )
+
+
+class StandingRulesObservation(NativeObservation):
+    def read(self, root: Path) -> MeasuredOutput:
+        if root.resolve() == CommanderConfig.read(CommanderConfig.path()).root.resolve():
+            raise ValueError("Heartbeat publication verification requires an isolated bus copy")
+        config = CommanderConfig.read_root(root)
+        comms = Comms(config.root)
+        measured: list[tuple[str, ...]] = []
+        before = comms.bus.log.latest_sequence()
+
+        def returned(value: object) -> None:
+            if not standing_rule_return(value):
+                raise TypeError("StandingRule.broken returned outside its declared contract")
+            measured.append(tuple(rule.line() for rule in value))
+
+        with returned_values({method_code(native_rules.StandingRule.broken): returned}):
+            CommanderReview().wake(
+                comms, config.commander, comms.registry.require(config.commander)
+            )
+        with comms.bus.log.opened_wire_snapshot() as source:
+            published = tuple(
+                message for message in source.newest_messages() if message.seq > before
+            )
+        if len(measured) != 1 or len(published) != int(bool(measured[0])):
+            raise ValueError("Heartbeat must consume one observation and commit its actual message")
+        if published and (
+            published[0].sender != config.commander or published[0].target != config.commander
+        ):
+            raise ValueError("Heartbeat published outside its configured commander route")
+        lines = tuple(
+            line.removeprefix("RULE ")
+            for message in published
+            for line in message.body.splitlines()
+            if line.startswith("RULE ")
+        )
+        return MeasuredOutput(
+            "\n".join(lines),
+            "\n".join(measured[0]),
+            str(native_rules.__file__),
+            str(comms.root),
+            config.commander,
+        )
+
+
+class RuleReviewObservation(NativeObservation):
+    def read(self, root: Path) -> MeasuredOutput:
+        config = CommanderConfig.read_root(root)
+        current = tuple(
+            rule.line()
+            for rule in native_rules.RulesDue.violations(Comms(config.root), config.commander)
+        )
+        segments = tuple(
+            segment.__name__ for segment in InstructionSegment.members_with(InstructionSegment)
+        )
+        raise SystemExit(
+            "Missing native owner: commander rule-review InstructionSegment with rule-file "
+            f"provenance. Current RulesDue={current!r}; "
+            f"native instruction declarations={segments!r}"
+        )
+
+
+class OrderContactObservation(NativeObservation):
+    def read(self, root: Path) -> MeasuredOutput:
+        config = CommanderConfig.read_root(root)
+        context = Commander.read_root(config.root)
+        units = sorted(context.units())
+        current = tuple(
+            RawSendCommand.refused(
+                ("agent-comms", "send", "--from", config.commander, "--to", unit),
+                CommanderSession("", context),
+            )
+            for unit in units[:1]
+        )
+        native = tuple(
+            tool.declared_name for tool in NativeToolDeclaration.members_with(NativeToolDeclaration)
+        )
+        raise SystemExit(
+            "Missing native owner: order-only-contact native harness hook "
+            "checking Core order tools. "
+            f"Current RawSendCommand={current!r}; native coding declarations={native!r}"
+        )
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("observation", type=NativeObservation.member_named)
+    parser.add_argument(
+        "--root", type=Path, default=CommanderConfig.read(CommanderConfig.path()).root
+    )
+    arguments = parser.parse_args()
+    observed = arguments.observation().read(arguments.root)
+    print(json.dumps(FieldCodec.encode(observed)))
+
+
+if __name__ == "__main__":
+    main()