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/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/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 9e3e6dc714..452e0a7eb3 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 @@ -761,25 +762,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_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"