From b7ffa87fe9c15dd1769e70cc097e332a09d0bc28 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 23:55:22 -0400 Subject: [PATCH 1/5] P302: bound load comparison and observe actual CLI returns without frame locals --- src/agent_comms/commander/capacity.py | 9 +- src/agent_comms/commander_parity/load.py | 70 ++++------ .../commander_parity/observations.py | 123 ++++++++++++++++++ src/agent_comms/turn_digests.py | 21 ++- tests/test_commander_parity_load.py | 3 +- 5 files changed, 173 insertions(+), 53 deletions(-) create mode 100644 src/agent_comms/commander_parity/observations.py diff --git a/src/agent_comms/commander/capacity.py b/src/agent_comms/commander/capacity.py index b3dda4f026..d3a7852f9c 100644 --- a/src/agent_comms/commander/capacity.py +++ b/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: diff --git a/src/agent_comms/commander_parity/load.py b/src/agent_comms/commander_parity/load.py index 66ec536cef..f6a050dd14 100644 --- a/src/agent_comms/commander_parity/load.py +++ b/src/agent_comms/commander_parity/load.py @@ -5,39 +5,13 @@ import subprocess from dataclasses import dataclass from pathlib import Path +from typing import ClassVar from ..commander.config import CommanderConfig from ..field_codec import FieldCodec 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 +23,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,16 +47,25 @@ 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, - ) + try: + current = subprocess.run( + (str(config.cli.parent / "python"), "-I", *arguments), + capture_output=True, + text=True, + timeout=cls.timeout, + ) + except subprocess.TimeoutExpired as error: + return UnverifiedParity( + f"{cls.__name__}: configured native owner timed out after {error.timeout} seconds" + ) if current.returncode: return UnverifiedParity(current.stderr) observed = FieldCodec.decode(MeasuredOutput, FieldCodec.parse(current.stdout)) @@ -103,3 +79,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() diff --git a/src/agent_comms/commander_parity/observations.py b/src/agent_comms/commander_parity/observations.py new file mode 100644 index 0000000000..f92ef2ef88 --- /dev/null +++ b/src/agent_comms/commander_parity/observations.py @@ -0,0 +1,123 @@ +"""Observe real native command returns and their actual output.""" + +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 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.declared_family import DeclaredFamily +from agent_comms.field_codec import FieldCodec + + +@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): 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 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() diff --git a/src/agent_comms/turn_digests.py b/src/agent_comms/turn_digests.py index 335f184d2a..414657faf8 100644 --- a/src/agent_comms/turn_digests.py +++ b/src/agent_comms/turn_digests.py @@ -18,6 +18,7 @@ if TYPE_CHECKING: from .comms import Comms + from .registry_document import RegistrySnapshot from .threads import Thread from .todos import Todo @@ -122,13 +123,25 @@ class OrderRead[Result](TranscriptRead[Result]): @classmethod def for_commander(cls, comms: Comms, commander: str) -> Iterator[Self]: - actor = comms.registry.require(commander).incarnation - for todo in comms.goals.todos.held_for_commander(actor): + snapshot = comms.registry.snapshot() + actor = snapshot.require(commander).incarnation + yield from cls.for_orders( + comms.root, snapshot, comms.goals.todos.held_for_commander(actor) + ) + + @classmethod + def for_orders( + cls, root: Path, snapshot: RegistrySnapshot, todos: Iterable[Todo] + ) -> Iterator[Self]: + """Read these orders against their captured roster, without reopening the registry.""" + for todo in todos: assignment = todo.held() - holder = comms.registry.require(assignment.owner) + holder = snapshot.require(assignment.owner) if holder.incarnation != assignment.owner_identity: raise ToolRefusedError(f"Order {todo.id} holder incarnation no longer exists") - yield cls(comms.root, holder, TurnsSince(todo.order.checked().at.recorded_at("An order")), todo) + yield cls( + root, holder, TurnsSince(todo.order.checked().at.recorded_at("An order")), todo + ) def edits_from_turns(self, turns: tuple[TurnDigest, ...]) -> tuple[EditCall, ...]: judged = {verdict.tool_call_id for verdict in self.todo.verdicts} diff --git a/tests/test_commander_parity_load.py b/tests/test_commander_parity_load.py index 6e26d798fa..98ed6adfa8 100644 --- a/tests/test_commander_parity_load.py +++ b/tests/test_commander_parity_load.py @@ -1,7 +1,8 @@ from dataclasses import replace from agent_comms.commander.capacity import Load -from agent_comms.commander_parity.load import LoadComparison, MeasuredOutput, state_files +from agent_comms.commander_parity.load import LoadComparison, state_files +from agent_comms.commander_parity.observations import MeasuredOutput def test_load_comparison_requires_every_reported_field_and_no_private_write(tmp_path): From a16681672154aa1e9b9db88fa9d9badb1fdeb247 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 23:55:22 -0400 Subject: [PATCH 2/5] P302: read real heartbeat publication on a bounded committed bus copy --- plans/commander-native-plan.md | 2 +- .../commander_parity/observations.py | 98 ++++++++++++++++++- src/agent_comms/commander_parity/rows.py | 15 ++- .../commander_parity/rules_contact.py | 68 +++++++++++++ src/agent_comms/committed_store_copy.py | 55 +++++++++++ tools/cutover/backup.py | 52 +--------- 6 files changed, 239 insertions(+), 51 deletions(-) create mode 100644 src/agent_comms/commander_parity/rules_contact.py create mode 100644 src/agent_comms/committed_store_copy.py diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 05042371b2..9d77a3e9b6 100644 --- a/plans/commander-native-plan.md +++ b/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` | diff --git a/src/agent_comms/commander_parity/observations.py b/src/agent_comms/commander_parity/observations.py index f92ef2ef88..85a9a4fd5e 100644 --- a/src/agent_comms/commander_parity/observations.py +++ b/src/agent_comms/commander_parity/observations.py @@ -1,4 +1,4 @@ -"""Observe real native command returns and their actual output.""" +"""Observe real native command returns and their actual output or stored publication.""" from __future__ import annotations @@ -12,14 +12,21 @@ 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) @@ -108,6 +115,95 @@ def context(value: object) -> None: ) +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) diff --git a/src/agent_comms/commander_parity/rows.py b/src/agent_comms/commander_parity/rows.py index 3772660101..be42a754dd 100644 --- a/src/agent_comms/commander_parity/rows.py +++ b/src/agent_comms/commander_parity/rows.py @@ -21,6 +21,7 @@ ) from .load import LoadComparison from .measurements import RuleComparison, StorageComparison, measured_load +from .rules_contact import OrderContactComparison, RuleReviewComparison, StandingRulesComparison from .story import StoryCase @@ -148,10 +149,14 @@ class ScheduledWakeParityRow(ParityRow): class StandingRulesParityRow(ParityRow): number = 15 capability = "Standing rules broken right now" - current = "`rotate.py broken_rules`" + current = "`CommanderReview.wake` (replaces `rotate.py broken_rules`)" native = "`StandingRule` members derive violations on read" phases = (ControlPlanePhase,) + @classmethod + def verify(cls) -> ParityComparison: + return StandingRulesComparison.read() + class RuleReviewParityRow(ParityRow): number = 16 @@ -160,6 +165,10 @@ class RuleReviewParityRow(ParityRow): native = "Rule instruction segments in the commander's turn context, with provenance" phases = (ControlPlanePhase,) + @classmethod + def verify(cls) -> ParityComparison: + return RuleReviewComparison.read() + class LoadForkingParityRow(ParityRow): number = 17 @@ -206,6 +215,10 @@ class OrderOnlyContactParityRow(ParityRow): native = "Harness hook checking Core order tools" phases = (ControlPlanePhase,) + @classmethod + def verify(cls) -> ParityComparison: + return OrderContactComparison.read() + class ConfigurationParityRow(ParityRow): number = 21 diff --git a/src/agent_comms/commander_parity/rules_contact.py b/src/agent_comms/commander_parity/rules_contact.py new file mode 100644 index 0000000000..176322a7ba --- /dev/null +++ b/src/agent_comms/commander_parity/rules_contact.py @@ -0,0 +1,68 @@ +"""Rules/contact comparisons use the existing installed-command observation lifetime.""" + +from pathlib import Path +from tempfile import TemporaryDirectory + +from ..commander.config import CommanderConfig +from ..committed_store_copy import CommittedStoreCopy +from ..comms import Comms +from ..launch_runtime import CallingThreadUnavailable, ThreadLaunch +from ..maintenance_barrier import MaintenanceBarrier +from ..store_files import _store_lock +from . import ParityComparison, UnverifiedParity +from .load import CommandComparison +from .observations import ( + OrderContactObservation, + RuleReviewObservation, + StandingRulesObservation, +) + + +class StandingRulesComparison(CommandComparison): + arguments = StandingRulesObservation.arguments() + timeout = 120 + + @classmethod + def read(cls) -> ParityComparison: + config = CommanderConfig.read(CommanderConfig.path()) + comms = Comms(config.root) + try: + caller = ThreadLaunch.current().name + except CallingThreadUnavailable: + caller = config.commander + owner = comms.registry.require(caller) + storage = comms.registry.storage_rule + size = storage.tree_gib(config.root) + occupied = storage.tree_gib(owner.storage_roots.scratch) + sum( + storage.tree_gib(path) for path in storage.checkouts_of(owner) + ) + if occupied + size > storage.scratch_cap_gib: + return UnverifiedParity("Committed bus copy exceeds this caller's scratch allowance") + if storage.free_gib() - size < storage.floor_gib: + return UnverifiedParity("Insufficient free space for a committed bus copy") + owner.storage_roots.scratch.mkdir(mode=0o700, parents=True, exist_ok=True) + with TemporaryDirectory(prefix="parity-rules-", dir=owner.storage_roots.scratch) as run: + root = Path(run) / "bus" + held = MaintenanceBarrier(config.root / "registry.json").witness_paths + with _store_lock(comms.bus.path): + CommittedStoreCopy.copy_tree(config.root, root, excluded=held) + selected = config.model_copy(update={"root": root}) + CommanderConfig.root_path(root).write_text(selected.model_dump_json() + "\n") + return cls.invoke(selected, (*cls.arguments, "--root", str(root))) + + def describe(self) -> str: + observed = self.observation + return ( + f"current_rules={len(observed.current.splitlines())}; " + f"native_rules={len(observed.native.splitlines())}; matched={self.passed}; " + f"owner={observed.source}; commander={observed.commander}; root={observed.root}; " + f"private_state_unchanged={self.state_before == self.state_after}" + ) + + +class RuleReviewComparison(CommandComparison): + arguments = RuleReviewObservation.arguments() + + +class OrderContactComparison(CommandComparison): + arguments = OrderContactObservation.arguments() diff --git a/src/agent_comms/committed_store_copy.py b/src/agent_comms/committed_store_copy.py new file mode 100644 index 0000000000..605852b1ef --- /dev/null +++ b/src/agent_comms/committed_store_copy.py @@ -0,0 +1,55 @@ +"""Copy committed stores through their existing SQLite and staging owners.""" + +from __future__ import annotations + +import shutil +import sqlite3 +import stat +from collections.abc import Collection +from contextlib import closing +from pathlib import Path + +from .coordination_database import CoordinationStore +from .locked_store import LockedStore + + +class CommittedStoreCopy: + @staticmethod + def copy_file(source: str | Path, target: str | Path) -> Path: + """Copy one store through its SQLite snapshot, never through live WAL bytes.""" + source, target = Path(source), Path(target) + if CoordinationStore.is_database(source): + with ( + closing(CoordinationStore.read_snapshot(source)) as reader, + closing(sqlite3.connect(target)) as writer, + ): + reader.backup(writer) + shutil.copystat(source, target) + else: + shutil.copy2(source, target) + return target + + @classmethod + def copy_tree(cls, source: Path, target: Path, *, excluded: Collection[Path] = ()) -> None: + """Copy committed files; their owners declare transient names before any stat.""" + + def transient(directory: str, names: list[str]) -> set[str]: + omitted: set[str] = set() + for name in sorted(names): + path = Path(directory) / name + if name in omitted or path in excluded or LockedStore.is_staging(path): + omitted.add(name) + continue + if stat.S_ISSOCK(path.lstat().st_mode): + omitted.add(name) + elif ( + path.is_file() and not path.is_symlink() and CoordinationStore.is_database(path) + ): + omitted.update( + companion.name for companion in CoordinationStore.files(path)[1:] + ) + return omitted + + shutil.copytree( + source, target, symlinks=True, ignore=transient, copy_function=cls.copy_file + ) diff --git a/tools/cutover/backup.py b/tools/cutover/backup.py index dfcb5545fd..c364ed2499 100644 --- a/tools/cutover/backup.py +++ b/tools/cutover/backup.py @@ -5,22 +5,20 @@ import json import os import shutil -import sqlite3 -import stat import subprocess import sys import tarfile import tempfile import time -from collections.abc import Collection -from contextlib import closing, contextmanager +from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import ClassVar, Iterator, Self +from agent_comms.committed_store_copy import CommittedStoreCopy from agent_comms.coordination_database import CoordinationStore from agent_comms.external_transcript import ExternalTranscript -from agent_comms.locked_store import LockedStore, RuntimeFile +from agent_comms.locked_store import RuntimeFile from agent_comms.store_files import _atomic_write_text HOLD_TRANSCRIPTS = """import sys @@ -151,7 +149,7 @@ def restore(self) -> None: print(f'Restored the bus; failed data is kept at {failed}', file=sys.stderr) -class StoreBackup(Backup): +class StoreBackup(CommittedStoreCopy, Backup): """Snapshot declared databases through SQLite; restore only the selected files.""" @property @@ -163,48 +161,6 @@ def digest(path: Path) -> str: return (CoordinationStore.digest(path) if CoordinationStore.is_database(path) else hashlib.sha256(path.read_bytes()).hexdigest()) - @staticmethod - def copy_file(source: str | Path, target: str | Path) -> Path: - """Copy one store through its SQLite snapshot, never through live WAL bytes.""" - source, target = Path(source), Path(target) - if CoordinationStore.is_database(source): - with closing(CoordinationStore.read_snapshot(source)) as reader: - with closing(sqlite3.connect(target)) as writer: - reader.backup(writer) - shutil.copystat(source, target) - else: - shutil.copy2(source, target) - return target - - @classmethod - def copy_tree( - cls, source: Path, target: Path, *, excluded: Collection[Path] = () - ) -> None: - """Copy committed files; their owners declare transient names before any stat.""" - - def transient(directory: str, names: list[str]) -> set[str]: - omitted: set[str] = set() - for name in sorted(names): - path = Path(directory) / name - if name in omitted or path in excluded or LockedStore.is_staging(path): - omitted.add(name) - continue - if stat.S_ISSOCK(path.lstat().st_mode): - omitted.add(name) - elif ( - path.is_file() - and not path.is_symlink() - and CoordinationStore.is_database(path) - ): - omitted.update( - companion.name for companion in CoordinationStore.files(path)[1:] - ) - return omitted - - shutil.copytree( - source, target, symlinks=True, ignore=transient, copy_function=cls.copy_file - ) - def capture(self, files: tuple[Path, ...]) -> dict[str, str]: contents = {} for source in files: From 3572aea03ee50014222bddcbb57d87326d9199be Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 20 Oct 2026 00:13:51 -0400 Subject: [PATCH 3/5] P302: let BoundedRun own native comparison collection and retirement --- src/agent_comms/commander_parity/load.py | 17 +++++------ tests/test_commander_parity_load.py | 39 +++++++++++++++++++++++- 2 files changed, 46 insertions(+), 10 deletions(-) diff --git a/src/agent_comms/commander_parity/load.py b/src/agent_comms/commander_parity/load.py index f6a050dd14..e839acc0f7 100644 --- a/src/agent_comms/commander_parity/load.py +++ b/src/agent_comms/commander_parity/load.py @@ -2,11 +2,12 @@ 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 ..private_path import FileRevision @@ -55,19 +56,17 @@ def invoke(cls, config: CommanderConfig, arguments: tuple[str, ...]) -> ParityCo 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. - try: - current = subprocess.run( + current = asyncio.run( + BoundedRun.run( (str(config.cli.parent / "python"), "-I", *arguments), - capture_output=True, - text=True, timeout=cls.timeout, ) - except subprocess.TimeoutExpired as error: + ) + if not current.outcome.successful: return UnverifiedParity( - f"{cls.__name__}: configured native owner timed out after {error.timeout} seconds" + f"{cls.__name__}: configured native owner {current.outcome.summary}; " + f"bound={cls.timeout}s; {current.stderr.decode(errors='replace')}" ) - if current.returncode: - return UnverifiedParity(current.stderr) observed = FieldCodec.decode(MeasuredOutput, FieldCodec.parse(current.stdout)) source = Path(observed.source) if ( diff --git a/tests/test_commander_parity_load.py b/tests/test_commander_parity_load.py index 98ed6adfa8..786797c6cf 100644 --- a/tests/test_commander_parity_load.py +++ b/tests/test_commander_parity_load.py @@ -1,7 +1,12 @@ +import sys from dataclasses import replace +from pathlib import Path + +import pytest from agent_comms.commander.capacity import Load -from agent_comms.commander_parity.load import LoadComparison, state_files +from agent_comms.commander.config import CommanderConfig +from agent_comms.commander_parity.load import CommandComparison, LoadComparison, state_files from agent_comms.commander_parity.observations import MeasuredOutput @@ -23,3 +28,35 @@ def test_load_comparison_requires_every_reported_field_and_no_private_write(tmp_ path.write_text("after") after = state_files(tmp_path) assert not LoadComparison(observed, before, after).passed + + +@pytest.mark.parametrize( + ("program", "expected"), + ( + ("raise SystemExit(7)", "exited with code 7"), + ("import time; time.sleep(10)", "timed out"), + ), +) +def test_failed_real_children_remain_named_unverified(tmp_path, program, expected): + """Negative process checks produce no fabricated native measurement or parity pass.""" + + class BriefComparison(CommandComparison): + timeout = 3 + + config = CommanderConfig( + commander="reader", + root=tmp_path / "bus", + cli=Path(sys.executable), + state=tmp_path / "state", + board=tmp_path / "board.md", + narrative=tmp_path / "narrative.md", + repos=(), + worktrees=tmp_path / "projects", + scratch=tmp_path / "scratch", + disk=tmp_path, + ) + comparison = BriefComparison.invoke(config, ("-c", program)) + assert not comparison.passed + assert expected in comparison.describe() + assert "bound=3s" in comparison.describe() + assert LoadComparison.timeout == 30 From 94161b203509917db87587fba3e796ba222af3f7 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 20 Oct 2026 00:30:50 -0400 Subject: [PATCH 4/5] P302: derive comparison interpreter from the selected runtime --- src/agent_comms/commander_parity/load.py | 5 ++++- tests/test_commander_parity_load.py | 3 --- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/agent_comms/commander_parity/load.py b/src/agent_comms/commander_parity/load.py index e839acc0f7..7f131371ae 100644 --- a/src/agent_comms/commander_parity/load.py +++ b/src/agent_comms/commander_parity/load.py @@ -10,6 +10,7 @@ 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 from .observations import CapacityObservation, MeasuredOutput @@ -56,10 +57,12 @@ def invoke(cls, config: CommanderConfig, arguments: tuple[str, ...]) -> ParityCo 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. + runtime = LaunchRuntime.running() current = asyncio.run( BoundedRun.run( - (str(config.cli.parent / "python"), "-I", *arguments), + (str(runtime.interpreter), "-I", *arguments), timeout=cls.timeout, + runtime=runtime, ) ) if not current.outcome.successful: diff --git a/tests/test_commander_parity_load.py b/tests/test_commander_parity_load.py index 786797c6cf..2e19bcaaf1 100644 --- a/tests/test_commander_parity_load.py +++ b/tests/test_commander_parity_load.py @@ -1,6 +1,4 @@ -import sys from dataclasses import replace -from pathlib import Path import pytest @@ -46,7 +44,6 @@ class BriefComparison(CommandComparison): config = CommanderConfig( commander="reader", root=tmp_path / "bus", - cli=Path(sys.executable), state=tmp_path / "state", board=tmp_path / "board.md", narrative=tmp_path / "narrative.md", From 390c9a70b11a333cf280ef08cc0546cc9367a775 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 20 Oct 2026 00:58:18 -0400 Subject: [PATCH 5/5] P302: observe the CLI-selected root context return --- .../commander_parity/observations.py | 2 +- tests/test_commander_parity_load.py | 20 ++++++++++++++++++- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/src/agent_comms/commander_parity/observations.py b/src/agent_comms/commander_parity/observations.py index 85a9a4fd5e..98c8afc7bb 100644 --- a/src/agent_comms/commander_parity/observations.py +++ b/src/agent_comms/commander_parity/observations.py @@ -97,7 +97,7 @@ def context(value: object) -> None: returned_values( { method_code(capacity.Load.of): measurement, - method_code(Commander.read): context, + method_code(Commander.read_root): context, } ), redirect_stdout(output), diff --git a/tests/test_commander_parity_load.py b/tests/test_commander_parity_load.py index 2e19bcaaf1..129c5fd33c 100644 --- a/tests/test_commander_parity_load.py +++ b/tests/test_commander_parity_load.py @@ -1,3 +1,5 @@ +import shutil +import sys from dataclasses import replace import pytest @@ -5,7 +7,7 @@ from agent_comms.commander.capacity import Load from agent_comms.commander.config import CommanderConfig from agent_comms.commander_parity.load import CommandComparison, LoadComparison, state_files -from agent_comms.commander_parity.observations import MeasuredOutput +from agent_comms.commander_parity.observations import CapacityObservation, MeasuredOutput def test_load_comparison_requires_every_reported_field_and_no_private_write(tmp_path): @@ -28,6 +30,22 @@ def test_load_comparison_requires_every_reported_field_and_no_private_write(tmp_ assert not LoadComparison(observed, before, after).passed +def test_capacity_observes_actual_cli_selected_root(commander_script, tmp_path, monkeypatch): + original, _ = commander_script + selected = tmp_path / "selected" + shutil.copytree(original.root, selected) + before = state_files(original.root) + monkeypatch.setattr(sys, "argv", ["capacity-observation"]) + + observed = CapacityObservation().read(selected) + + assert observed.root == str(selected.resolve()) + assert observed.commander == original.commander + assert observed.current.strip() == observed.native + assert state_files(original.root) == before + assert sys.monitoring.get_tool(sys.monitoring.PROFILER_ID) is None + + @pytest.mark.parametrize( ("program", "expected"), (