diff --git a/docs/p241-native-measurements.md b/docs/p241-native-measurements.md new file mode 100644 index 0000000000..dbffd40d09 --- /dev/null +++ b/docs/p241-native-measurements.md @@ -0,0 +1,59 @@ +# P241: shared commander load and growth + +Owner: pi-cn-phase2. PR #1965. Rebased onto requested main `15ba6edce`. + +## Ownership + +- `commander/capacity.py:Load` owns native and CLI load, the weighting formula, + and selection of receiving child commanders with retained orders. The CLI + uses its supplied commander context; it does not construct a competing one. +- `turn_digests.py:OrderRead.for_commander` owns the shared order/holder selection + and incarnation check. Both full edit reads and count-only reads consume it. + `OrderEditCountRead.count` serves load and the unjudged standing rule. The + competing `CommsUnjudgedTool.readers/count` methods are deleted. +- The heartbeat reader/writer, its CLI write and its config-directory creation + are deleted. Registry parentage/status and held orders supply that fact. +- `commander/standing.py:MergeGrowth` reads one captured first-parent Git history + with numstat, not a fetch followed by two subprocesses per revision. + `repo.py:Repo.git` owns bounded checked execution and remote discovery. +- Main's `CommanderContext`, argument parser, order lifecycle and recorded-check + progress implementations are preserved. `ToolCaller.acting_name` selects the + explicitly bound caller or the context's configured commander. The CLI uses + the existing concrete caller states and FieldCodec normalization, not a nullable + name or a second context. Checked order reads obtain their cutoff from the order. + +Before-rebase AST evidence uses the existing audit Package/Repository parser +across Core, tools and tests: `p241-2200-ast-before.txt` in this unit's scratch. +Receiver bindings are syntactic; dynamic resolution and state freshness are not +proved by the census. The previous retained source evidence is also preserved. + +## Verification and limitations + +The 20:52 checkpoint was broken: a CLI FieldCodec reference lacked its import. +Its retained tests reported 11 passed, one failure and 36 errors. Both ratchets +exited 2 because the command omitted the required `--head`; those were not +successful checks. The current rebase removes that obsolete CLI identity change. + +Current checks use the shared interpreter, background `timeout 300` commands, +and the unchanged checker against the requested main. Results and application +receipts are retained under `/home/ts/.cache/agent-scratch/pi-cn-phase2/` with +prefix `p241-2200-`. No installation or provider input is performed here. + +Final source checkpoint `d00ab51d7`: **49 tests passed** under background +`timeout 300`; source ratchet **1 solely SourceLines +8**, with no introduced +findings; tools ratchet **0**. Both ran separately against `15ba6edce`, with +`--head` supplied. Before/after AST acquisition reported no parse omissions. + +Actual configured saved-state native handler read **32 open / 30 overdue / +1,860 unjudged / total 278.0**. The real CLI then read **32 / 30 / 1,861**; +the count increased by one between reads, so this is not a frozen parity sample. The +focused test checks exact native/CLI equality on one retained fixture. This is +not an installed native-transport or provider-process verification. + +The 22:10 readiness deadline was missed: the final fork argument representation +needed correction. No failed or running check was represented as successful. + +This delivers the shared load/growth checkpoint. Broader timing/alert parity and +P244's comparisons are not claimed complete. Earlier native/CLI samples are +historical evidence only, not proof of the rebased artifact. No blocked goal is +reset, and no uncertain operation is replayed. diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 888180377f..5daaf8adaa 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -28,42 +28,43 @@ ## Parity matrix -| # | Capability | Script source today | Native owner | Phase | + +| # | Capability | Current commander | Native owner | Native phases | |---|---|---|---|---| -| 1 | Create a problem with owner, order and deadline | `order.py problem`; `problems.json` | `CommsTodoCreateTool` (exists) | 1 | -| 2 | Assign, reassign, drop an order | `order.py send`, `--reassign`, `drop`; `deadlines.json` | `CommsOrderTool`, `CommsTodoTool` assignment and release (exist) | 1 | -| 3 | Complete only on the delivered commit | `order.py done` | `CommsOrderDoneTool` (exists), returning a delivery state: merged to `main`, merged to another branch, not merged | 1, 3 | -| 4 | Close with installed evidence | `order.py close`; `closed.jsonl` | `CommsProblemCloseTool` (exists) | 1 | -| 5 | Board, overdue, silent, spinning | `order.py board`, `overdue`; `fleet-board.md` | `CommsTodoListTool` and the fleet tool, derived (exist) | 1 | -| 6 | Branch hygiene | `branches-*.txt` by hand | `CommsWorkBranchesTool` (exists) | done | -| 7 | Strikes and the strike limit | `order.py strike`; `strikes.jsonl` | `Strike` rows on the order; `StrikeLimit` declared on the commanded author; one `comms_fire` | 3 | -| 8 | Verdicts on a unit's tool calls | `judge.py ok`, `back`; `verdicts.jsonl` | `Verdict` on the order; a send-back revises the order Goal and records the strike in one write | 3 | -| 9 | Plain words in verdicts and strikes | judge and order checks | `PlainText` value type | 3 | -| 10 | Refuse soft or vague orders | `order.py` | `OrderText` value type on every order action | 3 | -| 11 | What a unit did during its order | `tool_watch.py`; `edits.jsonl`; `judge.py show` | `comms_order_work`, read from both transcript families by `tool_call_id` | 4 | -| 12 | What the commander has not judged yet | `judge.py unjudged`; `vibe_watch.py` | `comms_unjudged`: tool calls and unit messages after the last verdict, derived | 4 | -| 13 | Public story and roasts | `fleet-narrative.md`; `judge.py roast` | `comms_order_story`, `comms_fleet_story`; each send-back verdict posts its plain reason to the order's channel | 5 | -| 14 | Heartbeat: next unit every minute | `rotate.py` (60 s loop) | a scheduled commander wake owned by Core | 7 | -| 15 | Standing rules broken right now | `rotate.py broken_rules` | a `StandingRule` family, each member deriving its violations on read | 7 | -| 16 | Rule review | `rules.py`; `rules.jsonl`; a maximum age | the rules as instruction segments in the commander's turn context, with provenance | 7 | -| 17 | Load and forking | `capacity.py`: 6 units per commander, 10-minute beat age; `forks/` | `CommanderLoad` derived from open orders and the unjudged queue; forks are registered threads | 7 | -| 18 | Growth alarm | `growth_watch.py`: merges growing `src` by more than 40 net lines | a growth check derived on read from git per configured repository | 7 | -| 19 | Storage hygiene | `storage.py`: worktree limit per unit, cleanup due under 30 GB free, nothing allocates under 5 GB | `UnitStorage` derived on read from worktree paths and free disk | 7 | -| 20 | Commander reaches units only through orders | `send_guard.py` (PreToolUse hook) | stays a hook, at the harness boundary, checking against Core's order tools | 7 | -| 21 | Paths, repositories, commander identity | `CommanderConfig` from `commander.json` | Core's project and registry configuration | 8 | -| 33 | Several commanders on one machine and bus, one per project | none (one `commander.json`; a second install overwrites the first) | each commander is a registered thread that owns one project. Orders, strikes, load, board, alarms and forks are keyed by that commander, and the commander for a call comes from its `--from` thread. The send guard refuses a raw send from any registered commander. Two commanders for one project are refused when the second registers | 1, 8 | -| 22 | Seeing the fleet: board, strikes, queue, story, alarms | `fleet-board.md`, `order.py board`, the terminal output of `rotate.py` and `judge.py` | a commander page in Toad, rendering Core's answers through `ui_model` | 6 | -| 23 | An order states the user-visible run that closes it | scratch `rts/order.py send` (`Works when:`) | `OrderText` requires it; `CommsProblemCloseTool.works` is checked against it | 3 | -| 24 | Reassigning past two owners needs the commander's trace | scratch `rts/order.py send --reassign --traced` | the transfer operation reads owners from the Todo's assignment history and refuses without a traced source location | 3 | -| 25 | Unit wording gets a correction, not a strike | scratch `vibe_watch.py` | `PlainText` applies to the commander's own verdicts and strikes; unit wording never becomes a strike | 3 | -| 26 | Alerts carry only decisions | none (today every line interrupts) | the commander's wake receives decision events only, each once | 7 | -| 27 | Momentum prediction per unit, compared with the outcome | none | a prediction record on the order, shown on the commander page beside what happened | 6 | -| 28 | Measured load and self-report | `capacity.py` unit count | `CommanderLoad`: look debt, judge debt, inflow, calibration, context, plus a one-word self-report calibrated against them | 7 | -| 29 | No handoff of finished, checked work | none (scratch `order.py send` moved P507 after its fix was pushed and checked) | the transfer operation refuses an order whose current owner has a delivered commit with its check passing; that owner merges it, and the check's recorded run is reused, never rerun by another unit | 3 | -| 30 | A proof is bound to the build it ran on | none (screenshots and logs are bare file paths) | the run record written by the build check (`tools/build_runtime.py`) carries runtime ID and Core/Toad revisions; `CommsProblemCloseTool` refuses when that runtime does not contain the order's delivered commit | 3 | -| 31 | An order states its intent as a deletion target | none (orders name a line, so units fix the line: P520 swapped `is_set()` for `except ValueError` around every caller) | the order declares the decision family being deleted, its owner, and a family-wide count from the existing NRA/refactor-audit scanners; completion runs that count on the delivered commit and refuses unless it is zero with no new site of the same kind | 3 | -| 32 | Where units' time goes, classified | none (`activity.log` has call times but no durations or turn boundaries) | Core's turn records carry each call's start and end and each turn's start reason; `UnitTime` derives per unit and fleet the minutes in model, tool runs (build, check, test, shell), waiting on another owner, waiting on the commander, and idle; the commander marks each large wait necessary or self-inflicted | 6, 7 | -| 33 | Order progress beside its deadline; the unit never sees a due time | `order.py board`, `overdue`, `due`, `send`; `rotate.py` (before: a due time and "Delivered" only, so merged work drew the strike threat) | `OrderProgress` in Core (`todos.py`), derived on read from the unit's checkout and the repository's PRs, never stored: no commit, committed not pushed, pushed with no PR, open PR, merged; `CommsTodoListTool` returns it per held order. Merged work draws no strike; no first commit within 10 minutes, or no new commit, push or PR change for 15 minutes, is a STALL with the progress line. `CommsOrderTool` sends the unit no due time: the deadline stays the commander's fire line | 1, 7 | +| 1 | Create a problem with owner, order and deadline | `order.py problem`; `problems.json` | `CommsTodoCreateTool` | `one_order_owner` | +| 2 | Assign, reassign, drop an order | `order.py send`, `--reassign`, `drop`; `deadlines.json` | `CommsOrderTool`, `CommsTodoTool` assignment and release | `one_order_owner` | +| 3 | Complete only on the delivered commit | `order.py done` | `CommsOrderDoneTool`: merged to main, another branch, or not merged | `one_order_owner`, `record_plane` | +| 4 | Close with installed evidence | `order.py close`; `closed.jsonl` | `CommsProblemCloseTool` | `one_order_owner` | +| 5 | Board, overdue, silent, spinning | `order.py board`, `overdue`; `fleet-board.md` | `CommsTodoListTool` and the fleet tool, derived | `one_order_owner` | +| 6 | Branch hygiene | `branches-*.txt` by hand | `CommsWorkBranchesTool` | `one_order_owner` | +| 7 | Strikes and the strike limit | `order.py strike`; `strikes.jsonl` | `Strike` rows, author-declared `StrikeLimit`, one `comms_fire` | `record_plane` | +| 8 | Verdicts on a unit's tool calls | `judge.py ok`, `back`; `verdicts.jsonl` | Order `Verdict`; send-back revises the Goal and records its strike atomically | `record_plane` | +| 9 | Plain words in verdicts and strikes | judge and order checks | `PlainText` | `record_plane` | +| 10 | Refuse soft or vague orders | `order.py` | `OrderText` on every order action | `record_plane` | +| 11 | What a unit did during its order | `tool_watch.py`; `edits.jsonl`; `judge.py show` | `comms_order_work`: both transcript families by `tool_call_id` | `review_queue` | +| 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` | +| 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 | `growth_watch.py`: merges growing src by more than 40 net lines | `commander.config.Repo.main_changes` → `commander_measurements.MergeGrowth.read` | `control_plane` | +| 19 | Storage hygiene | `storage.py`: checkout limit, cleanup under 30 GB, allocation floor 5 GB | `UnitStorage`, derived from worktree paths and free disk | `control_plane` | +| 20 | Commander reaches units only through orders | `send_guard.py` (PreToolUse hook) | Harness hook checking Core order tools | `control_plane` | +| 21 | Paths, repositories, commander identity | `CommanderConfig` from `commander.json` | Core project and registry configuration | `cutover` | +| 22 | Seeing the fleet: board, strikes, queue, story, alarms | `fleet-board.md`, `order.py board`, `rotate.py`, `judge.py` | Toad commander page rendering Core through `ui_model` | `commander_page` | +| 23 | An order states the user-visible run that closes it | scratch `rts/order.py send` (`Works when:`) | `OrderText` requires the run; `CommsProblemCloseTool.works` checks it | `record_plane` | +| 24 | Reassigning past two owners needs the commander's trace | scratch `order.py send --reassign --traced` | Transfer reads Todo owner history and requires traced source location | `record_plane` | +| 25 | Unit wording gets a correction, not a strike | scratch `vibe_watch.py` | `PlainText` for commander verdicts/strikes; unit wording gets correction | `record_plane` | +| 26 | Alerts carry only decisions | none; every line interrupts | `messages.AlertKind.require_publishable`, `tools.CommsDecisionTool` | `control_plane` | +| 27 | Momentum prediction per unit, compared with the outcome | none | Prediction record and later outcome on commander page; `CommitMomentum` is a partial candidate | `commander_page` | +| 28 | Measured load and self-report | `capacity.py` unit count | `commander.capacity.Load.of` reports orders, overdue, edits and forks; self-report comparison unverified | `control_plane` | +| 29 | No handoff of finished, checked work | none; scratch transfer moved P507 after its fix was pushed and checked | Transfer refuses delivered checked work; that owner merges; check is reused | `record_plane` | +| 30 | A proof is bound to the build it ran on | none; screenshots and logs are bare paths | Build run records runtime and Core/Toad revisions; close requires delivered commit | `record_plane` | +| 31 | An order states its intent as a deletion target | none; P520 renamed a decision rather than deleting it | Order owns deletion family/site/count; delivery refuses nonzero or renamed decisions | `record_plane` | +| 32 | Where units' time goes, classified | none; activity.log lacks durations and turn boundaries | `UnitTime`: model, build/check/test/shell, dependency/commander waits and idle; classify large waits | `commander_page`, `control_plane` | +| 33 | Order progress beside its deadline; the unit never sees a due time | `order.py board`, `overdue`, `due`, `send`; `rotate.py` | Derived `OrderProgress`; merged work draws no strike; first commit 10 min, later progress 15 min STALL; units see no due time | `one_order_owner`, `control_plane` | +| 34 | Several commanders on one machine and bus, one per project | none; a second commander.json install overwrites the first | Registry commander/project ownership; caller from --from; scoped orders, strikes, load, board, alarms and forks; raw-send guard; duplicate project commander refused | `one_order_owner`, `cutover` | **Dropped deliberately:** old verdicts and `edits.jsonl`. They judged copies of tool calls, and any old call can be judged again from the transcripts. diff --git a/src/agent_comms/commander/capacity.py b/src/agent_comms/commander/capacity.py index f0f1d62f47..b3dda4f026 100644 --- a/src/agent_comms/commander/capacity.py +++ b/src/agent_comms/commander/capacity.py @@ -1,43 +1,37 @@ -"""Report commander load and live forks; no command refuses on capacity.""" +"""One commander load report for native tools and the commander's CLI.""" from __future__ import annotations -import os, time +import time from dataclasses import dataclass -from pathlib import Path from ..comms import Comms - -BEAT_AGE = 10 * 60 - - -def live_forks(forks: Path) -> list[str]: - os.makedirs(forks, exist_ok=True) - return [n[:-5] for n in os.listdir(forks) if n.endswith('.beat') and time.time() - os.path.getmtime(f'{forks}/{n}') < BEAT_AGE] - - -def beat(forks: Path, name: str) -> None: - os.makedirs(forks, exist_ok=True) - with open(f'{forks}/{name}.beat', 'a'): - os.utime(f'{forks}/{name}.beat') +from ..field_codec import projected +from ..tool_result import ToolResult +from ..turn_digests import OrderEditCountRead @dataclass(frozen=True) -class Load: - """The commander load on one Comms: its open and overdue orders, its unjudged edits, live forks.""" +class Load(ToolResult): + """Order and edit debt, with receiving children that themselves command orders.""" owed: int overdue: int unjudged: int forks: tuple[str, ...] @classmethod - def of(cls, comms: Comms, commander: str, forks: Path) -> Load: - from ..tools import CommsUnjudgedTool - actor, now = comms.registry.require(commander).incarnation, time.time() - owed = [t for t in comms.goals.todos.held_for_commander(actor) if t.pending] + 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) + 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), - CommsUnjudgedTool.count(comms, commander), tuple(live_forks(forks))) + OrderEditCountRead.count(comms, owner.name), forks) - @property + @projected(view="wire") def total(self) -> float: return self.owed + 2 * self.overdue + self.unjudged / 10 @@ -47,4 +41,4 @@ def __str__(self) -> str: def main(a, ctx): - print(Load.of(ctx.comms, ctx.commander, ctx.state / 'forks')) + print(Load.of(ctx.comms, ctx.commander)) diff --git a/src/agent_comms/commander/config.py b/src/agent_comms/commander/config.py index 6d89c50408..c837f375d7 100644 --- a/src/agent_comms/commander/config.py +++ b/src/agent_comms/commander/config.py @@ -96,7 +96,7 @@ def write(self, home: Path) -> Path: record.write_text(self.model_dump_json(indent=1) + '\n') path.unlink(missing_ok=True) path.symlink_to(record) - for directory in (self.state, self.state / 'forks', self.state / 'orders', self.worktrees, + for directory in (self.state, self.state / 'orders', self.worktrees, self.scratch, self.board.parent): directory.mkdir(parents=True, exist_ok=True) return path diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index d233fcdb78..33e6ee33de 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -76,7 +76,7 @@ VoidStrikeTodo, WithdrawTodo, ) -from ..tool_caller import ThreadCaller +from ..tool_caller import NoThreadCaller, ThreadCaller from ..tools import ( CommsChangeProblemTool, CommsFireTool, @@ -89,7 +89,6 @@ CommsTodoListTool, TodoList, ) -from . import capacity from .config import CommanderConfig, CommanderRole from .send_guard import CommanderSession @@ -765,10 +764,12 @@ def fire(self, unit: str, why: str) -> None: @dataclass(frozen=True, kw_only=True) class CommanderVerb(CommanderRole, DeclaredFamily, DeclaredOptions, affix="Verb"): """One commander verb: its options are its fields, its parser comes from them, and run() does the work.""" - fork: str | None = option("--fork", default=None, help="a forked commander's name; keeps its beat fresh") + fork: ThreadCaller | NoThreadCaller = option( + "--fork", type=ThreadCaller, normalize=FieldCodec.encode, + default=NoThreadCaller(), help="the acting commander's name") def who(self, ctx: Commander) -> str: - return self.fork or ctx.commander + return self.fork.acting_name(ctx.commander) def apply(self, ctx: Commander) -> None: self.gate(ctx) @@ -776,8 +777,6 @@ def apply(self, ctx: Commander) -> None: def gate(self, ctx: Commander) -> None: self.require_commander(ctx.comms.registry.require(self.who(ctx))) - if self.fork: - capacity.beat(ctx.state / 'forks', self.fork) @abstractmethod def run(self, ctx: Commander) -> None: ... diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index 5fa54d176e..fb4430b38c 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -4,12 +4,10 @@ import asyncio import logging -import subprocess import time from collections.abc import Callable, Iterable, Iterator from contextlib import AbstractContextManager from dataclasses import dataclass -from pathlib import Path from typing import ClassVar from ..comms import Comms @@ -25,11 +23,8 @@ ) from ..thread_identity import StallWatcherIncarnation, ThreadIncarnation from ..threads import Thread -from ..todos import ( - FlatTrend, OrderProgress, PullRequests, Todo, -) -from ..tools import CommsUnjudgedTool -from ..turn_digests import TurnDigestRead, TurnsSince +from ..todos import FlatTrend, OrderProgress, PullRequests, Todo +from ..turn_digests import OrderEditCountRead, TurnDigestRead, TurnsSince from .config import CommanderConfig @@ -202,7 +197,7 @@ class Unjudged(ReasonedStandingRule, UndatedStandingRule): @classmethod def reason(cls, comms: Comms, commander: str) -> str | None: - total = CommsUnjudgedTool.count(comms, commander) + total = OrderEditCountRead.count(comms, commander) return str(total) if total > cls.limit else None @@ -220,37 +215,33 @@ class MergeGrowth(DatedStandingRule): src: tuple[int, int] tests: tuple[int, int] - @staticmethod - def git(checkout: Path, *args: str) -> str: - return subprocess.run( - ["git", "-C", str(checkout), *args], capture_output=True, text=True - ).stdout - - @classmethod - def numstat(cls, checkout: Path, sha: str, path: str) -> tuple[int, int]: - diff = cls.git(checkout, "diff", "--numstat", f"{sha}^1", sha, "--", path) - counted = [ - (int(a), int(d)) - for a, d, _ in (row.split("\t", 2) for row in diff.splitlines()) - if a.isdigit() - ] - return sum(a for a, _ in counted), sum(d for _, d in counted) - @classmethod def merges(cls, repo: Repo) -> Iterator[MergeGrowth]: - """Each first-parent commit on the repository's main inside the window, newest first.""" - cls.git(repo.checkout, "fetch", "-q", "origin", "main") - log = cls.git( - repo.checkout, - "log", - "--first-parent", - f"--since={cls.window}", - "--format=%H %ct %s", - "origin/main", + """Read first-parent growth from one captured, already-fetched Git history. + + Fetch belongs to the checkout's update owner, not a standing-rule read. + Git emits each commit's first-parent diff in this same process. + """ + log = repo.git( + repo.checkout, "log", "--first-parent", f"--since={cls.window}", + "--format=%x1e%H%x00%ct%x00%s", "--numstat", "--no-renames", + "--diff-merges=first-parent", "origin/main", "--", repo.source, "tests", ) - for sha, at, subject in (line.split(" ", 2) for line in log.splitlines()): - src, tests = (cls.numstat(repo.checkout, sha, path) for path in (repo.source, "tests")) - yield cls(repo.github, sha, subject, float(at), src, tests) + for record in log.split("\x1e")[1:]: + header, _, diff = record.partition("\n") + sha, at, subject = header.split("\x00", 2) + src, tests = [0, 0], [0, 0] + for row in diff.splitlines(): + if not row: + continue + added, deleted, path = row.split("\t", 2) + if not (added.isdigit() and deleted.isdigit()): + continue + for root, counts in ((repo.source, src), ("tests", tests)): + if path == root or path.startswith(root.rstrip("/") + "/"): + counts[0] += int(added) + counts[1] += int(deleted) + yield cls(repo.github, sha, subject, float(at), (src[0], src[1]), (tests[0], tests[1])) @classmethod def violations(cls, comms: Comms, commander: str) -> tuple[MergeGrowth, ...]: diff --git a/src/agent_comms/commander_parity/__init__.py b/src/agent_comms/commander_parity/__init__.py new file mode 100644 index 0000000000..22073f86a4 --- /dev/null +++ b/src/agent_comms/commander_parity/__init__.py @@ -0,0 +1,92 @@ +"""Parity declarations own matrix generation and the cutover decision.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Iterable +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, ClassVar + +from ..declared_family import DeclaredFamily + +if TYPE_CHECKING: + from ..plan_phases import PlanPhase + from ..todos import Todo + + +class ParityComparison(ABC): + @property + @abstractmethod + def passed(self) -> bool: ... + + def describe(self) -> str: + return repr(self) + + +@dataclass(frozen=True) +class UnverifiedParity(ParityComparison): + """Missing behavior has no evidence from which to claim a pass.""" + + reason: str = "The native/current workflow comparison has not run." + passed: ClassVar[bool] = False + + def describe(self) -> str: + return self.reason + + +class ParityRow( + DeclaredFamily, affix="ParityRow", discovery_package="agent_comms.commander_parity" +): + number: ClassVar[int] + capability: ClassVar[str] + current: ClassVar[str] + native: ClassVar[str] + phases: ClassVar[tuple[type[PlanPhase], ...]] + + @classmethod + def compare(cls, todos: Iterable[Todo]) -> ParityComparison: + return UnverifiedParity() + + @classmethod + def verify(cls) -> ParityComparison: + """Run this declared workflow now; a run is not durable cutover authority.""" + return UnverifiedParity() + + @classmethod + def rows(cls) -> tuple[type[ParityRow], ...]: + rows = tuple(sorted(cls.members_with(cls), key=lambda row: row.number)) + numbers = tuple(row.number for row in rows) + if not numbers or min(numbers) < 1 or len(set(numbers)) != len(numbers): + raise ValueError("Parity declarations require unique positive row numbers") + return rows + + @classmethod + def done(cls, todos: Iterable[Todo]) -> bool: + history = tuple(todos) + return all(row.compare(history).passed for row in cls.rows()) + + @classmethod + def matrix(cls) -> str: + """Render declared facts only; no mutable completion column lives here.""" + lines = [ + "", + "| # | Capability | Current commander | Native owner | Native phases |", + "|---|---|---|---|---|", + ] + for row in cls.rows(): + owners = ", ".join(f"`{phase.declared_name}`" for phase in row.phases) + lines.append( + f"| {row.number} | {row.capability} | {row.current} | {row.native} | {owners} |" + ) + return "\n".join(lines) + "\n" + + @classmethod + def render_plan(cls, path: Path) -> str: + """Replace a generated section, never decode prose as executable identity.""" + text = path.read_text() + before, marker, remainder = text.partition("## Parity matrix\n") + _, end, after = remainder.partition("\n**Dropped deliberately:**") + if not marker or not end: + raise ValueError("Plan has no bounded parity matrix section") + return before + marker + "\n" + cls.matrix() + end + after diff --git a/src/agent_comms/commander_parity/__main__.py b/src/agent_comms/commander_parity/__main__.py new file mode 100644 index 0000000000..ee378b8178 --- /dev/null +++ b/src/agent_comms/commander_parity/__main__.py @@ -0,0 +1,42 @@ +"""Render the commander matrix from its owning declarations.""" + +import argparse +import json +from pathlib import Path + +from . import ParityRow + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--plan", type=Path, help="Render the full plan instead of only the table") + parser.add_argument( + "--verify", action="store_true", help="Run live comparisons; do not grant cutover" + ) + args = parser.parse_args() + if args.verify: + comparisons = tuple((row, row.verify()) for row in ParityRow.rows()) + print( + json.dumps( + { + "verified": sum(result.passed for _, result in comparisons), + "total": len(comparisons), + "rows": [ + dict( + number=row.number, + capability=row.capability, + passed=result.passed, + result=result.describe(), + ) + for row, result in comparisons + ], + }, + indent=2, + ) + ) + raise SystemExit(0 if all(result.passed for _, result in comparisons) else 1) + print(ParityRow.render_plan(args.plan) if args.plan else ParityRow.matrix(), end="") + + +if __name__ == "__main__": + main() diff --git a/src/agent_comms/commander_parity/load.py b/src/agent_comms/commander_parity/load.py new file mode 100644 index 0000000000..66ec536cef --- /dev/null +++ b/src/agent_comms/commander_parity/load.py @@ -0,0 +1,105 @@ +"""Observe the native load measurement used by the installed capacity command.""" + +from __future__ import annotations + +import subprocess +from dataclasses import dataclass +from pathlib import Path + +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))) +""" + + +def state_files(directory: Path) -> tuple[tuple[str, FileRevision], ...]: + return tuple( + (str(path.relative_to(directory)), FileRevision.from_stat(path.stat())) + for path in sorted(directory.rglob("*")) + if path.is_file() + ) + + +@dataclass(frozen=True) +class MeasuredOutput: + current: str + native: str + source: str + root: str + commander: str + + +@dataclass(frozen=True) +class LoadComparison(ParityComparison): + observation: MeasuredOutput + state_before: tuple[tuple[str, FileRevision], ...] + state_after: tuple[tuple[str, FileRevision], ...] + + @property + def passed(self) -> bool: + return ( + self.observation.current.strip() == self.observation.native + and self.state_before == self.state_after + ) + + def describe(self) -> str: + return ( + f"current={self.observation.current.strip()!r}; " + f"native={self.observation.native!r}; owner={self.observation.source}; " + f"commander={self.observation.commander}; root={self.observation.root}; " + f"private_state_unchanged={self.state_before == self.state_after}" + ) + + @classmethod + def read(cls) -> ParityComparison: + config = CommanderConfig.read(CommanderConfig.path()) + 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, + ) + if current.returncode: + return UnverifiedParity(current.stderr) + observed = FieldCodec.decode(MeasuredOutput, FieldCodec.parse(current.stdout)) + source = Path(observed.source) + if ( + observed.root != str(config.root) + or observed.commander != config.commander + or not any(source.is_relative_to(repo.checkout) for repo in config.repos) + ): + return UnverifiedParity( + "The installed command used a different configuration or source" + ) + return cls(observed, before, state_files(config.state)) diff --git a/src/agent_comms/commander_parity/rows.py b/src/agent_comms/commander_parity/rows.py new file mode 100644 index 0000000000..0c64e8c69a --- /dev/null +++ b/src/agent_comms/commander_parity/rows.py @@ -0,0 +1,306 @@ +"""Each native/current workflow declares its identity once.""" + +from ..plan_phases import ( + CommanderPagePhase, + ControlPlanePhase, + CutoverPhase, + OneOrderOwnerPhase, + RecordPlanePhase, + ReviewQueuePhase, + StoryPhase, +) +from . import ParityComparison, ParityRow +from .load import LoadComparison + + +class CreateProblemParityRow(ParityRow): + number = 1 + capability = "Create a problem with owner, order and deadline" + current = "`order.py problem`; `problems.json`" + native = "`CommsTodoCreateTool`" + phases = (OneOrderOwnerPhase,) + + +class AssignOrderParityRow(ParityRow): + number = 2 + capability = "Assign, reassign, drop an order" + current = "`order.py send`, `--reassign`, `drop`; `deadlines.json`" + native = "`CommsOrderTool`, `CommsTodoTool` assignment and release" + phases = (OneOrderOwnerPhase,) + + +class DeliveredCommitParityRow(ParityRow): + number = 3 + capability = "Complete only on the delivered commit" + current = "`order.py done`" + native = "`CommsOrderDoneTool`: merged to main, another branch, or not merged" + phases = (OneOrderOwnerPhase, RecordPlanePhase) + + +class InstalledEvidenceParityRow(ParityRow): + number = 4 + capability = "Close with installed evidence" + current = "`order.py close`; `closed.jsonl`" + native = "`CommsProblemCloseTool`" + phases = (OneOrderOwnerPhase,) + + +class BoardParityRow(ParityRow): + number = 5 + capability = "Board, overdue, silent, spinning" + current = "`order.py board`, `overdue`; `fleet-board.md`" + native = "`CommsTodoListTool` and the fleet tool, derived" + phases = (OneOrderOwnerPhase,) + + +class BranchHygieneParityRow(ParityRow): + number = 6 + capability = "Branch hygiene" + current = "`branches-*.txt` by hand" + native = "`CommsWorkBranchesTool`" + phases = (OneOrderOwnerPhase,) + + +class StrikeLimitParityRow(ParityRow): + number = 7 + capability = "Strikes and the strike limit" + current = "`order.py strike`; `strikes.jsonl`" + native = "`Strike` rows, author-declared `StrikeLimit`, one `comms_fire`" + phases = (RecordPlanePhase,) + + +class VerdictParityRow(ParityRow): + number = 8 + capability = "Verdicts on a unit's tool calls" + current = "`judge.py ok`, `back`; `verdicts.jsonl`" + native = "Order `Verdict`; send-back revises the Goal and records its strike atomically" + phases = (RecordPlanePhase,) + + +class PlainWordsParityRow(ParityRow): + number = 9 + capability = "Plain words in verdicts and strikes" + current = "judge and order checks" + native = "`PlainText`" + phases = (RecordPlanePhase,) + + +class OrderWordingParityRow(ParityRow): + number = 10 + capability = "Refuse soft or vague orders" + current = "`order.py`" + native = "`OrderText` on every order action" + phases = (RecordPlanePhase,) + + +class OrderWorkParityRow(ParityRow): + number = 11 + capability = "What a unit did during its order" + current = "`tool_watch.py`; `edits.jsonl`; `judge.py show`" + native = "`comms_order_work`: both transcript families by `tool_call_id`" + phases = (ReviewQueuePhase,) + + +class UnjudgedParityRow(ParityRow): + number = 12 + capability = "What the commander has not judged yet" + current = "`judge.py unjudged`; `vibe_watch.py`" + native = "`comms_unjudged`: calls and unit messages after the last verdict" + phases = (ReviewQueuePhase,) + + +class PublicStoryParityRow(ParityRow): + number = 13 + capability = "Public story and roasts" + current = "`fleet-narrative.md`; `judge.py roast`" + native = "`comms_order_story`, `comms_fleet_story`; send-back reason posted once" + phases = (StoryPhase,) + + +class ScheduledWakeParityRow(ParityRow): + number = 14 + capability = "Heartbeat: next unit every minute" + current = "`rotate.py` (60 s loop)" + native = "Core-owned scheduled commander wake" + phases = (ControlPlanePhase,) + + +class StandingRulesParityRow(ParityRow): + number = 15 + capability = "Standing rules broken right now" + current = "`rotate.py broken_rules`" + native = "`StandingRule` members derive violations on read" + phases = (ControlPlanePhase,) + + +class RuleReviewParityRow(ParityRow): + number = 16 + capability = "Rule review" + current = "`rules.py`; `rules.jsonl`; maximum age" + native = "Rule instruction segments in the commander's turn context, with provenance" + phases = (ControlPlanePhase,) + + +class LoadForkingParityRow(ParityRow): + number = 17 + capability = "Load and forking" + current = "`agent-comms-commander capacity`: load and live fork reports; ten-minute beat age" + native = ( + "Reporting-only `commander.capacity.Load.of`; no capacity refusal or automatic fork policy" + ) + phases = (ControlPlanePhase,) + + @classmethod + def verify(cls) -> ParityComparison: + return LoadComparison.read() + + +class GrowthAlarmParityRow(ParityRow): + number = 18 + capability = "Growth alarm" + current = "`growth_watch.py`: merges growing src by more than 40 net lines" + native = "`commander.config.Repo.main_changes` → `commander_measurements.MergeGrowth.read`" + phases = (ControlPlanePhase,) + + +class StorageHygieneParityRow(ParityRow): + number = 19 + capability = "Storage hygiene" + current = "`storage.py`: checkout limit, cleanup under 30 GB, allocation floor 5 GB" + native = "`UnitStorage`, derived from worktree paths and free disk" + phases = (ControlPlanePhase,) + + +class OrderOnlyContactParityRow(ParityRow): + number = 20 + capability = "Commander reaches units only through orders" + current = "`send_guard.py` (PreToolUse hook)" + native = "Harness hook checking Core order tools" + phases = (ControlPlanePhase,) + + +class ConfigurationParityRow(ParityRow): + number = 21 + capability = "Paths, repositories, commander identity" + current = "`CommanderConfig` from `commander.json`" + native = "Core project and registry configuration" + phases = (CutoverPhase,) + + +class CommanderPageParityRow(ParityRow): + number = 22 + capability = "Seeing the fleet: board, strikes, queue, story, alarms" + current = "`fleet-board.md`, `order.py board`, `rotate.py`, `judge.py`" + native = "Toad commander page rendering Core through `ui_model`" + phases = (CommanderPagePhase,) + + +class ApplicationRunParityRow(ParityRow): + number = 23 + capability = "An order states the user-visible run that closes it" + current = "scratch `rts/order.py send` (`Works when:`)" + native = "`OrderText` requires the run; `CommsProblemCloseTool.works` checks it" + phases = (RecordPlanePhase,) + + +class TracedTransferParityRow(ParityRow): + number = 24 + capability = "Reassigning past two owners needs the commander's trace" + current = "scratch `order.py send --reassign --traced`" + native = "Transfer reads Todo owner history and requires traced source location" + phases = (RecordPlanePhase,) + + +class UnitWordingParityRow(ParityRow): + number = 25 + capability = "Unit wording gets a correction, not a strike" + current = "scratch `vibe_watch.py`" + native = "`PlainText` for commander verdicts/strikes; unit wording gets correction" + phases = (RecordPlanePhase,) + + +class DecisionAlertsParityRow(ParityRow): + number = 26 + capability = "Alerts carry only decisions" + current = "none; every line interrupts" + native = "`messages.AlertKind.require_publishable`, `tools.CommsDecisionTool`" + phases = (ControlPlanePhase,) + + +class MomentumParityRow(ParityRow): + number = 27 + capability = "Momentum prediction per unit, compared with the outcome" + current = "none" + native = ( + "Prediction record and later outcome on commander page; " + "`CommitMomentum` is a partial candidate" + ) + phases = (CommanderPagePhase,) + + +class MeasuredLoadParityRow(ParityRow): + number = 28 + capability = "Measured load and self-report" + current = "`capacity.py` unit count" + native = ( + "`commander.capacity.Load.of` reports orders, overdue, edits and forks; " + "self-report comparison unverified" + ) + phases = (ControlPlanePhase,) + + +class FinishedWorkTransferParityRow(ParityRow): + number = 29 + capability = "No handoff of finished, checked work" + current = "none; scratch transfer moved P507 after its fix was pushed and checked" + native = "Transfer refuses delivered checked work; that owner merges; check is reused" + phases = (RecordPlanePhase,) + + +class BuildBoundProofParityRow(ParityRow): + number = 30 + capability = "A proof is bound to the build it ran on" + current = "none; screenshots and logs are bare paths" + native = "Build run records runtime and Core/Toad revisions; close requires delivered commit" + phases = (RecordPlanePhase,) + + +class DeletionTargetParityRow(ParityRow): + number = 31 + capability = "An order states its intent as a deletion target" + current = "none; P520 renamed a decision rather than deleting it" + native = "Order owns deletion family/site/count; delivery refuses nonzero or renamed decisions" + phases = (RecordPlanePhase,) + + +class UnitTimeParityRow(ParityRow): + number = 32 + capability = "Where units' time goes, classified" + current = "none; activity.log lacks durations and turn boundaries" + native = ( + "`UnitTime`: model, build/check/test/shell, dependency/commander waits and idle; " + "classify large waits" + ) + phases = (CommanderPagePhase, ControlPlanePhase) + + +class OrderProgressParityRow(ParityRow): + number = 33 + capability = "Order progress beside its deadline; the unit never sees a due time" + current = "`order.py board`, `overdue`, `due`, `send`; `rotate.py`" + native = ( + "Derived `OrderProgress`; merged work draws no strike; first commit 10 min, " + "later progress 15 min STALL; units see no due time" + ) + phases = (OneOrderOwnerPhase, ControlPlanePhase) + + +class MultipleCommandersParityRow(ParityRow): + number = 34 + capability = "Several commanders on one machine and bus, one per project" + current = "none; a second commander.json install overwrites the first" + native = ( + "Registry commander/project ownership; caller from --from; scoped orders, strikes, " + "load, board, alarms and forks; raw-send guard; duplicate project commander refused" + ) + phases = (OneOrderOwnerPhase, CutoverPhase) diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index c42d40d3c8..04865c669b 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -12,6 +12,7 @@ from collections.abc import Iterable from typing import TYPE_CHECKING, ClassVar +from .commander_parity import ParityRow from .declared_family import DeclaredFamily from .phase_tasks import PhaseTask, UnspecifiedPhaseTask @@ -89,11 +90,17 @@ class ControlPlanePhase(PlanPhase): after = (RecordPlanePhase,) -class CutoverPhase(PlanPhase): - # The parity gate in the Markdown plan has no executable declaration yet. - # These dependencies do not establish parity or authorize deleting scripts. +class ParityPhase(PlanPhase): after = (RecordPlanePhase, ReviewQueuePhase, ControlPlanePhase) + @classmethod + def done(cls, todos: Iterable[Todo]) -> bool: + return ParityRow.done(todos) + + +class CutoverPhase(PlanPhase): + after = (RecordPlanePhase, ReviewQueuePhase, ControlPlanePhase, ParityPhase) + class StoryPhase(PlanPhase): """A view of the native records, not a prerequisite for their cutover.""" diff --git a/src/agent_comms/repo.py b/src/agent_comms/repo.py index b79abe7083..b6cfe20dbb 100644 --- a/src/agent_comms/repo.py +++ b/src/agent_comms/repo.py @@ -28,14 +28,19 @@ def contains(self, sha: str) -> bool: result.check_returncode() return result.returncode == 0 - @classmethod - def of(cls, checkout: Path) -> Repo: - """Its GitHub name comes from the checkout's own origin remote.""" - url = subprocess.run( - ["git", "-C", str(checkout), "remote", "get-url", "origin"], + @staticmethod + def git(checkout: Path, *args: str) -> str: + return subprocess.run( + ["git", "-C", str(checkout), *args], capture_output=True, text=True, check=True, - ).stdout.strip() + timeout=30, + ).stdout + + @classmethod + def of(cls, checkout: Path) -> Repo: + """Its GitHub name comes from the checkout's own origin remote.""" + url = cls.git(checkout, "remote", "get-url", "origin").strip() github = url.removesuffix(".git").replace(":", "/").split("/")[-2:] return cls(checkout=checkout.resolve(), github="/".join(github)) diff --git a/src/agent_comms/tool_caller.py b/src/agent_comms/tool_caller.py index 00723d54df..6c968d2900 100644 --- a/src/agent_comms/tool_caller.py +++ b/src/agent_comms/tool_caller.py @@ -38,6 +38,9 @@ def listing(self, comms: Comms, threads: tuple[ThreadRecord, ...]) -> ThreadList @abstractmethod def catalog(self, comms: Comms, tools: tuple[dict[str, object], ...]) -> ToolCatalogResult: ... + @abstractmethod + def acting_name(self, default: str) -> str: ... + @abstractmethod def subject(self, named: str) -> str: ... @@ -67,6 +70,9 @@ def catalog(self, comms: Comms, tools: tuple[dict[str, object], ...]) -> ToolCat def refuse_thread(self) -> NoReturn: raise CallingThreadUnavailable("This tool requires a thread caller.") + def acting_name(self, default: str) -> str: + return default + def subject(self, named: str) -> str: if not named: return self.refuse_thread() @@ -96,6 +102,9 @@ def listing(self, comms: Comms, threads: tuple[ThreadRecord, ...]) -> ThreadList def catalog(self, comms: Comms, tools: tuple[dict[str, object], ...]) -> ToolCatalogResult: return ThreadToolCatalogResult(tools, comms.registry.require(self.name).service_tier) + def acting_name(self, default: str) -> str: + return self.name + def subject(self, named: str) -> str: return named or self.name diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index f438bc251a..cf39fb398f 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -12,6 +12,7 @@ from pydantic import JsonValue +from .commander.capacity import Load from .progress_claims import ProgressClaim from .pi_vocabulary import ServiceTier, ThinkingLevel, ThinkingLevelField from .cli_commands import ( @@ -79,7 +80,7 @@ TaskScopeSelection, OriginalTaskChange, ) from .order_work import OrderWork -from .turn_digests import LastTurns, OrderEditCountRead, OrderEditsRead, ThreadTurns, TurnsSince, TurnDigestRead, UnjudgedEdits +from .turn_digests import LastTurns, OrderEditsRead, ThreadTurns, TurnsSince, TurnDigestRead, UnjudgedEdits from .relationships import RelationshipEdit from .work_branches import WorkBranch, WorkBranches from .tool_output import MAX_INLINE_OUTPUT_BYTES @@ -763,25 +764,18 @@ class CommsUnjudgedTool(ThreadReadTool): label = "Unjudged edits" description = "Read original edit calls since each open order, excluding that order's recorded verdicts." - @classmethod - def readers(cls, comms: Comms, commander: str, read_type=OrderEditsRead): - actor = comms.registry.require(commander).incarnation - for todo in comms.goals.todos.held_for_commander(actor): - assignment = todo.held() - holder = comms.registry.require(assignment.owner) - if holder.incarnation != assignment.owner_identity: - raise ToolRefusedError( - f"Order {todo.id} holder incarnation no longer exists" - ) - yield read_type(comms.root, holder, TurnsSince(todo.order.at.recorded_at("An order")), todo) + def apply(self, comms: Comms, caller: ToolCaller) -> UnjudgedEdits: + return UnjudgedEdits(tuple(edit for read in OrderEditsRead.for_commander( + comms, caller.subject(self.thread)) for edit in read.read())) - @classmethod - def count(cls, comms: Comms, commander: str) -> int: - return sum(read.read() for read in cls.readers(comms, commander, OrderEditCountRead)) - def apply(self, comms: Comms, caller: ToolCaller) -> UnjudgedEdits: - return UnjudgedEdits(tuple(edit for read in self.readers(comms, caller.subject(self.thread)) - for edit in read.read())) +@dataclass(frozen=True, kw_only=True) +class CommsCommanderLoadTool(ThreadReadTool): + label = "Measured commander load" + description = "Read canonical order/edit debt and receiving child commanders; never self-reported." + + def apply(self, comms: Comms, caller: ToolCaller) -> Load: + return Load.of(comms, caller.subject(self.thread)) @dataclass(frozen=True, kw_only=True) diff --git a/src/agent_comms/turn_digests.py b/src/agent_comms/turn_digests.py index 69ba2f14f7..f9af553518 100644 --- a/src/agent_comms/turn_digests.py +++ b/src/agent_comms/turn_digests.py @@ -3,19 +3,21 @@ from __future__ import annotations from abc import ABC, abstractmethod -from collections.abc import Iterable +from collections.abc import Iterable, Iterator from dataclasses import dataclass from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Self +from .errors import ToolRefusedError +from .external_transcript import ExternalTranscript from .thread_execution import ConversationPreparation from .thread_identity import ThreadIncarnation -from .external_transcript import ExternalTranscript -from .transcript_events import TurnDigest -from .tool_result import ToolResult from .tool_declaration import EditCall +from .tool_result import ToolResult +from .transcript_events import TurnDigest if TYPE_CHECKING: + from .comms import Comms from .threads import Thread from .todos import Todo @@ -117,9 +119,19 @@ def from_turns(self, turns: tuple[TurnDigest, ...]) -> ThreadTurns: class OrderRead[Result](TranscriptRead[Result]): todo: Todo + @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): + assignment = todo.held() + holder = comms.registry.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) + def edits_from_turns(self, turns: tuple[TurnDigest, ...]) -> tuple[EditCall, ...]: judged = {verdict.tool_call_id for verdict in self.todo.verdicts} - since = self.todo.order.at.recorded_at("An order") + since = self.todo.order.checked().at.recorded_at("An order") edits = {edit.id: edit for turn in reversed(turns) for event in turn.events for edit in event.edit_calls(self.thread.name, self.todo.id, since) @@ -140,6 +152,10 @@ def from_external(self, source) -> tuple[EditCall, ...]: class OrderEditCountRead(OrderRead[int]): """Load needs a count, never external event bodies or edit hunks.""" + @classmethod + def count(cls, comms: Comms, commander: str) -> int: + return sum(read.read() for read in cls.for_commander(comms, commander)) + def from_turns(self, turns: tuple[TurnDigest, ...]) -> int: return len(self.edits_from_turns(turns)) diff --git a/tests/test_commander_capacity.py b/tests/test_commander_capacity.py index fe0ded1098..c56aac065a 100644 --- a/tests/test_commander_capacity.py +++ b/tests/test_commander_capacity.py @@ -2,13 +2,31 @@ import pytest +from agent_comms.commander import capacity +from agent_comms.commander.capacity import Load +from agent_comms.field_codec import FieldCodec +from agent_comms.tools import CommsCommanderLoadTool +from test_commander_problem_orders import create + @pytest.mark.parametrize("unjudged", [0, 1, 395, 1_000_000]) -def test_capacity_reports_load_and_forks(commander_script, unjudged): - script, _ = commander_script - load = script.capacity.Load(24, 14, unjudged, ("judge",)) +def test_capacity_reports_load_and_forks(unjudged): + load = Load(24, 14, unjudged, ("judge",)) + assert FieldCodec.encode(load)["total"] == load.total assert load.total == 24 + 2 * 14 + unjudged / 10 assert str(load) == ( f"load {load.total:.0f} (24 open orders, 14 overdue, {unjudged} unjudged edits), " "1 live forks ['judge']" ) + + +def test_native_and_cli_share_load_without_a_heartbeat_store(commander, commander_script, capsys): + comms, caller = commander + context, _ = commander_script + create(commander, 'P1', deadline=1) + loaded = CommsCommanderLoadTool(thread='lead').apply(comms, caller) + assert loaded == Load.of(context.comms, context.commander) + assert (loaded.owed, loaded.overdue, loaded.unjudged, loaded.forks) == (1, 1, 0, ()) + capacity.main([], context) + assert capsys.readouterr().out == f'{loaded}\n' + assert not (context.state / 'forks').exists() diff --git a/tests/test_commander_parity.py b/tests/test_commander_parity.py new file mode 100644 index 0000000000..a034cd029e --- /dev/null +++ b/tests/test_commander_parity.py @@ -0,0 +1,61 @@ +from pathlib import Path + +from agent_comms.commander_parity import ParityComparison, ParityRow, UnverifiedParity +from agent_comms.plan_phases import ( + CommanderPagePhase, + ControlPlanePhase, + CutoverPhase, + ParityPhase, + PlanPhase, + RecordPlanePhase, + ReviewQueuePhase, + StoryPhase, +) + +PLAN = Path(__file__).resolve().parents[1] / "plans/commander-native-plan.md" + + +class MatchingComparison(ParityComparison): + passed = True + + +def test_generated_matrix_is_a_derived_output(): + assert ParityRow.render_plan(PLAN) == PLAN.read_text() + rows = ParityRow.rows() + assert len(rows) == len({row.number for row in rows}) == 34 + assert all(row.capability and row.current and row.native and row.phases for row in rows) + assert all(not row.compare(()).passed for row in rows) + + +def test_cutover_gate_is_computed_from_every_comparison(monkeypatch): + assert CutoverPhase.after == ( + RecordPlanePhase, + ReviewQueuePhase, + ControlPlanePhase, + ParityPhase, + ) + assert not ParityPhase.done(()) + for row in ParityRow.rows(): + monkeypatch.setattr(row, "compare", classmethod(lambda cls, history: MatchingComparison())) + assert ParityPhase.done(()) + last = ParityRow.rows()[-1] + monkeypatch.setattr(last, "compare", classmethod(lambda cls, history: UnverifiedParity())) + assert not ParityPhase.done(()) + assert not any( + dependency in (StoryPhase, CommanderPagePhase) + for phase in PlanPhase.members_with(PlanPhase) + for dependency in phase.after + ) + + +def test_generated_prose_is_not_input_to_the_gate(tmp_path): + plan = tmp_path / "plan.md" + plan.write_text( + "Before\n## Parity matrix\n\nHand edited, invalid prose.\n" + "**Dropped deliberately:** historical text\nAfter\n" + ) + rendered = ParityRow.render_plan(plan) + assert rendered.startswith("Before\n## Parity matrix\n\n") + assert rendered.endswith("**Dropped deliberately:** historical text\nAfter\n") + assert "Hand edited" not in rendered + assert not ParityPhase.done(()) diff --git a/tests/test_commander_parity_load.py b/tests/test_commander_parity_load.py new file mode 100644 index 0000000000..6e26d798fa --- /dev/null +++ b/tests/test_commander_parity_load.py @@ -0,0 +1,24 @@ +from dataclasses import replace + +from agent_comms.commander.capacity import Load +from agent_comms.commander_parity.load import LoadComparison, MeasuredOutput, state_files + + +def test_load_comparison_requires_every_reported_field_and_no_private_write(tmp_path): + load = Load(37, 8, 1692, ("merger",)) + observed = MeasuredOutput(str(load) + "\n", str(load), "capacity.py", "/bus", "lead") + comparison = LoadComparison(observed, (), ()) + assert comparison.passed + for different in ( + replace(load, owed=38), + replace(load, overdue=9), + replace(load, unjudged=1693), + replace(load, forks=()), + ): + assert not replace(comparison, observation=replace(observed, native=str(different))).passed + path = tmp_path / "old-state" + path.write_text("before") + before = state_files(tmp_path) + path.write_text("after") + after = state_files(tmp_path) + assert not LoadComparison(observed, before, after).passed diff --git a/tests/test_commander_verbs.py b/tests/test_commander_verbs.py index 80c536086a..0814654571 100644 --- a/tests/test_commander_verbs.py +++ b/tests/test_commander_verbs.py @@ -5,12 +5,14 @@ import pytest +from agent_comms.tool_caller import NoThreadCaller, ThreadCaller + DOCUMENTED = [ ("order", "send worker 23:59 order.txt --clean --reassign 'missed two deadlines' --fork f1", {"unit": "worker", "due": "23:59", "body_file": "order.txt", "clean": True, - "reassign": "missed two deadlines", "fork": "f1"}), + "reassign": "missed two deadlines", "fork": ThreadCaller("f1")}), ("order", "strike worker --why 'P1 late' --caught 'lead at install'", - {"unit": "worker", "why": "P1 late", "caught": "lead at install", "fork": None}), + {"unit": "worker", "why": "P1 late", "caught": "lead at install", "fork": NoThreadCaller()}), ("order", "strike worker --why 'P1 late'", {"unit": "worker", "why": "P1 late", "caught": "review"}), ("order", "done worker m1 --sha abc123 --seen proof.txt", {"unit": "worker", "message": "m1", "sha": "abc123", "seen": "proof.txt"}), @@ -41,7 +43,7 @@ ("judge", "alarm", {}), ("judge", "ok 0123456789:ab --msg m1 --serves 'the words' --line one --line two --why why --fork f1", {"ids": ("0123456789:ab",), "msg": "m1", "serves": "the words", "lines": ("one", "two"), "why": "why", - "fork": "f1"}), + "fork": ThreadCaller("f1")}), ("judge", "ok 0123456789:ab --msg m1 --serves 'the words' --line one --why why --no-owner proof", {"serves": "the words", "no_owner": "proof"}), ("judge", "back 0123456789:ab --msg m1 --line one --why why --roast r --fix f --slop IDEN-5" diff --git a/tests/test_install_gate.py b/tests/test_install_gate.py index 519ca528ae..b9d5d33e71 100644 --- a/tests/test_install_gate.py +++ b/tests/test_install_gate.py @@ -13,6 +13,7 @@ import tempfile import unittest from contextlib import closing +from dataclasses import dataclass from pathlib import Path from unittest.mock import patch @@ -20,12 +21,14 @@ from install_gate import ( DEFAULT_SCRATCH, REPO, + Desktop, ExistingStatusChecks, FieldCodec, GateArguments, GateError, GateReport, InstallSurface, + Laptop, Machine, MalformedRecord, MissingProtection, @@ -46,6 +49,17 @@ ) +@dataclass(frozen=True) +class IsolatedMachine(Machine, declared_name=None): + name: str + bus: str + transport: tuple[str, ...] = () + + @property + def root(self) -> str: + return self.bus + + class InstallGateTests(unittest.TestCase): def setUp(self): DEFAULT_SCRATCH.mkdir(mode=0o700, parents=True, exist_ok=True) @@ -94,7 +108,7 @@ def test_read_only_wal_snapshot_and_closed_store(self): if path.is_file() } destination = self.root / "snapshot" - Machine("local-test", str(bus)).capture(destination, self.root / "snapshot.log") + IsolatedMachine("local-test", str(bus)).capture(destination, self.root / "snapshot.log") after = { path.name: hashlib.sha256(path.read_bytes()).hexdigest() for path in bus.iterdir() @@ -109,6 +123,34 @@ def test_read_only_wal_snapshot_and_closed_store(self): with closing(sqlite3.connect(destination / "bus/closed.sqlite3")) as copy: self.assertEqual(copy.execute("SELECT * FROM records").fetchall(), []) + def test_declared_hosts_use_the_local_leg_without_ssh(self): + for local, remote in ((Desktop(), Laptop()), (Laptop(), Desktop())): + with ( + self.subTest(host=local.name), + patch("install_gate.socket.gethostname", return_value=local.hostname), + ): + self.assertEqual( + local.command("printf", "%s", "two words"), ("printf", "%s", "two words") + ) + self.assertEqual(remote.transport[-1], remote.address) + self.assertEqual( + remote.command("printf", "%s", "two words")[-1], "printf %s 'two words'" + ) + + def test_laptop_root_comes_from_its_selected_route(self): + route = ( + '{"root":"/actual/laptop/bus","version":2,' + '"wire_root_id":"00000000000000000000000000000000"}' + ) + with ( + patch("install_gate.socket.gethostname", return_value="desktop"), + patch("install_gate.run", return_value=route) as read, + ): + self.assertEqual(Laptop().root, "/actual/laptop/bus") + self.assertEqual(read.call_args.args[0], "ssh") + self.assertIn(Laptop.address, read.call_args.args) + self.assertIn("active-route.json", read.call_args.args[-1]) + def test_sandbox_rejects_writes(self): target = self.root / "must-not-exist" result = subprocess.run( @@ -135,14 +177,16 @@ def test_sandbox_rejects_writes(self): def test_unreachable_desktop_and_symlinks_are_failures(self): with self.assertRaises(GateError): - Machine("desktop", "/unused", ("false",)).capture( + IsolatedMachine("desktop", "/unused", ("false",)).capture( self.root / "unreachable", self.root / "unreachable.log" ) bus = self.root / "source" bus.mkdir() (bus / "unsafe").symlink_to(self.root) with self.assertRaises(GateError): - Machine("local-test", str(bus)).capture(self.root / "linked", self.root / "linked.log") + IsolatedMachine("local-test", str(bus)).capture( + self.root / "linked", self.root / "linked.log" + ) def test_rehearsal_requires_completed_result_not_just_exit_zero(self): target = self.root / "target" diff --git a/tests/test_plan_phases.py b/tests/test_plan_phases.py index 669ccccd83..82dfc41aeb 100644 --- a/tests/test_plan_phases.py +++ b/tests/test_plan_phases.py @@ -68,7 +68,8 @@ def test_views_wait_for_native_work_but_never_block_cutover(commander, monkeypat accepted(todo) if todo.order.serves in required else todo for todo in todos ) - assert CutoverPhase.ready(native_done) + # Native source completion alone cannot bypass the independent parity gate. + assert not CutoverPhase.ready(native_done) assert StoryPhase.ready(native_done) assert CommanderPagePhase.ready(native_done) assert not SkillPhase.ready(native_done) diff --git a/tools/install_gate.py b/tools/install_gate.py index 1a2156d802..15678e3602 100644 --- a/tools/install_gate.py +++ b/tools/install_gate.py @@ -18,6 +18,7 @@ import shlex import shutil import signal +import socket import subprocess import sys import tarfile @@ -317,11 +318,24 @@ def require_current(self) -> None: raise GateError(f"Snapshot route version {self.version} is not supported") -@dataclass(frozen=True) -class Machine: - name: str - root: str - transport: tuple[str, ...] = () +class Machine(DeclaredFamily, affix="Machine"): + name: ClassVar[str] + hostname: ClassVar[str] + address: ClassVar[str] + + @property + @abstractmethod + def root(self) -> str: ... + + @property + def transport(self) -> tuple[str, ...]: + if self.hostname == socket.gethostname(): + return () + return ("ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", self.address) + + def command(self, *command: str) -> tuple[str, ...]: + transport = self.transport + return (*transport, shlex.join(command)) if transport else command def capture(self, destination: Path, log: Path) -> SnapshotCopy: destination.mkdir(mode=0o700) @@ -340,8 +354,7 @@ def capture(self, destination: Path, log: Path) -> SnapshotCopy: "-", self.root, ) - if self.transport: - command = (*self.transport, shlex.join(command)) + command = self.command(*command) # The producer has no writable mounts, including SQLite SHM and temp. with ExitStack() as resources: errors = resources.enter_context(log.open("wb")) @@ -382,6 +395,34 @@ def capture(self, destination: Path, log: Path) -> SnapshotCopy: return context +class Laptop(Machine): + name = "laptop" + hostname = "ts" + address = "ts@192.168.2.18" + + @property + def root(self) -> str: + document = FieldCodec.decode( + RouteDocument, + FieldCodec.parse( + run( + *self.command( + "sh", "-c", 'cat "$HOME/.local/state/agent-comms/active-route.json"' + ) + ) + ), + ) + document.require_current() + return str(document.selection.root) + + +class Desktop(Machine): + name = "desktop" + hostname = "desktop" + address = "ts@192.168.2.11" + root = "/var/tmp/agent-comms-live-20260927-wzjtqhza" + + def storage_floor() -> float: return float(StorageRule.load(Path.home() / ".agent-comms").floor_gib) * 2**30 @@ -998,23 +1039,7 @@ class GateArguments(argparse.Namespace): def run_machines( args: GateArguments, target: Target, report: Path, results: dict[str, MachineResult] ) -> dict[str, MachineResult]: - machines = ( - Machine("laptop", str(Path.home() / ".agent-comms")), - Machine( - "desktop", - "/var/tmp/agent-comms-live-20260927-wzjtqhza", - ( - "ssh", - "-i", - str(Path.home() / ".ssh/openhcs_laptop_ed25519"), - "-o", - "BatchMode=yes", - "-o", - "ConnectTimeout=10", - "ts@192.168.2.11", - ), - ), - ) + machines = tuple(host() for host in Machine.members_with(Machine)) results.update((machine.name, UnfinishedMachineResult()) for machine in machines) with tempfile.TemporaryDirectory(dir=Path.home() / "wt", prefix="install-gate-") as directory: work = Path(directory)