diff --git a/src/agent_comms/commander_parity/fleet.py b/src/agent_comms/commander_parity/fleet.py new file mode 100644 index 0000000000..4fd21f6bd6 --- /dev/null +++ b/src/agent_comms/commander_parity/fleet.py @@ -0,0 +1,261 @@ +"""Live board, branch and work comparisons against their current consumers.""" + +from __future__ import annotations + +import io +import re +from collections.abc import Callable +from configparser import ConfigParser +from contextlib import redirect_stdout +from dataclasses import dataclass +from sys import monitoring +from types import CodeType + +from pydantic import JsonValue + +from ..commander.judge import Edit +from ..commander.order import BoardVerb, Commander, hhmm +from ..field_codec import FieldCodec +from ..native_entries import NativeEntry +from ..native_tool_call import CodingCall +from ..tool_caller import ThreadCaller +from ..tool_declaration import EditCall +from ..tools import ( + CommsOrderWorkTool, + CommsThreadsTool, + CommsTodoListTool, + CommsUnjudgedTool, + CommsWorkBranchesTool, + TodoList, + UnjudgedEdits, +) +from ..work_branches import WorkBranch, WorkBranches +from . import ParityComparison, UnverifiedParity + + +@dataclass(frozen=True) +class ReadComparison(ParityComparison): + """Compare projected facts, not a count that can hide different records.""" + + owner: str + current: JsonValue + native: JsonValue + + @property + def passed(self) -> bool: + return self.current == self.native + + def describe(self) -> str: + return f"owner={self.owner}; current_native_equal={self.passed}" + + +def observe[T]( + code: CodeType, + current: Callable[[], object], + native: Callable[[tuple[T, ...]], object], + owner: str, + *, + expected: int = 1, +) -> ParityComparison: + """Observe the real native return used by the current path without replacing either. + + A second read of a changing fleet is not a comparison of the first read. + Only this function's return event is enabled; other profilers are untouched. + """ + results: list[T] = [] + tool = monitoring.PROFILER_ID + if monitoring.get_tool(tool): + return UnverifiedParity(f"{owner}: the Python profiler slot is already owned") + + def returned(returned_code: CodeType, offset: int, value: T) -> None: + results.append(value) + + monitoring.use_tool_id(tool, __name__) + try: + monitoring.register_callback(tool, monitoring.events.PY_RETURN, returned) + monitoring.set_local_events(tool, code, monitoring.events.PY_RETURN) + rendered = current() + finally: + monitoring.set_local_events(tool, code, 0) + monitoring.register_callback(tool, monitoring.events.PY_RETURN, None) + monitoring.free_tool_id(tool) + if not results or len(results) != expected: + return UnverifiedParity( + f"{owner}: expected {expected} live owner returns, observed {len(results)}" + ) + return ReadComparison( + owner, FieldCodec.encode(rendered), FieldCodec.encode(native(tuple(results))) + ) + + +@dataclass(frozen=True) +class BoardFacts: + problem: str + owner_file: str + holder: str + order: str + due: str + progress: str + delivered: bool + strikes: int + + @classmethod + def rendered(cls, board: str) -> tuple[BoardFacts, ...]: + rows = [] + for line in board.splitlines(): + if re.match(r"^\| P[1-9][0-9]*:", line): + values = tuple(part.strip() for part in line.split("|")[1:-1]) + if len(values) != 9: + raise ValueError("The current board has an ambiguous row") + problem, owner_file, holder, order, due, progress, delivered, strikes, _ = values + rows.append( + cls( + problem, + owner_file, + "" if holder.startswith("unassigned:") else holder, + order, + due, + progress, + ConfigParser.BOOLEAN_STATES[delivered.partition(",")[0]], + int(strikes), + ) + ) + return tuple(rows) + + @classmethod + def native_rows(cls, listed: TodoList) -> tuple[BoardFacts, ...]: + return tuple( + cls( + f"{todo.id}: {todo.text}", + todo.order.checked().owner.to_wire(), + todo.holder or "", + todo.order.checked().id, + hhmm(todo.deadline), + listed.progress[todo.id].line if todo.id in listed.progress else "-", + not todo.pending, + todo.strike_count, + ) + for todo in sorted(listed.todos, key=lambda item: int(item.id[1:])) + ) + + +class BoardComparison: + @classmethod + def read(cls) -> ParityComparison: + context = Commander.read(Commander.path()) + fleet = CommsThreadsTool().apply(context.comms, ThreadCaller(context.commander)) + names = {record.thread.name for record in fleet.threads} + + def current() -> tuple[BoardFacts, ...]: + output = io.StringIO() + with redirect_stdout(output): + BoardVerb().run(context) + return BoardFacts.rendered(output.getvalue()) + + def native(observed: tuple[TodoList, ...]) -> tuple[BoardFacts, ...]: + listed = observed[0] + if any(todo.holder and todo.holder not in names for todo in listed.todos): + raise ValueError("A board holder is absent from the live native fleet") + return BoardFacts.native_rows(listed) + + return observe( + CommsTodoListTool.apply.__code__, + current, + native, + "CommsTodoListTool + CommsThreadsTool / Commander BoardVerb", + ) + + +class BranchComparison: + @classmethod + def read(cls) -> ParityComparison: + context = Commander.read(Commander.path()) + if not context.repos: + return UnverifiedParity("CommsWorkBranchesTool has no configured repository to invoke") + + def current() -> tuple[tuple[WorkBranch, ...], ...]: + return tuple( + CommsWorkBranchesTool(checkout=str(repo.checkout), unlinked_only=False) + .apply(context.comms, ThreadCaller(context.commander)) + .branches + for repo in context.repos + ) + + def native( + readings: tuple[tuple[WorkBranch, ...], ...], + ) -> tuple[tuple[WorkBranch, ...], ...]: + return readings + + return observe( + WorkBranches.read.__code__, + current, + native, + "CommsWorkBranchesTool / actual consumed Git and forge branch records", + expected=len(context.repos), + ) + + +class OrderWorkComparison: + @classmethod + def read(cls) -> ParityComparison: + context = Commander.read(Commander.path()) + held = context.held_orders() + if not held: + return UnverifiedParity("CommsOrderWorkTool has no live held problem to invoke") + current: list[JsonValue] = [] + native: list[JsonValue] = [] + for todo in held: + reply = CommsOrderWorkTool(problem=todo.id).apply( + context.comms, + ThreadCaller(context.commander), + ) + since = todo.order.checked().at.recorded_at("An order") + calls: dict[str, str] = {} + session = context.comms.registry.require(todo.held().owner).session_file + if not session: + return UnverifiedParity("Current tool_watch has no native session for this order") + with open(session, "rb") as source: + for line in source: + if not line.endswith(b"\n"): + return UnverifiedParity( + "Current tool_watch source ends in an incomplete record" + ) + entry = NativeEntry.read(line) + if entry.timestamp.at.timestamp() >= since: + for message in entry.messages: + if message.assistant: + for pi_call in message.tool_calls: + call = CodingCall.decode( + pi_call.id, pi_call.name, pi_call.arguments + ) + calls[pi_call.id] = call.digest() + current.append(FieldCodec.encode((todo.id, tuple(reversed(calls.items()))))) + native.append( + FieldCodec.encode( + ( + todo.id, + tuple((call.tool_call_id, call.digest) for call in reply.calls), + ) + ) + ) + return ReadComparison( + "CommsOrderWorkTool / actual tool_watch NativeEntry and CodingCall digests", + current, + native, + ) + + +class UnjudgedComparison: + @classmethod + def read(cls) -> ParityComparison: + context = Commander.read(Commander.path()) + + def native(observed: tuple[UnjudgedEdits, ...]) -> tuple[EditCall, ...]: + return observed[0].edits + + return observe( + CommsUnjudgedTool.apply.__code__, + lambda: Edit.unjudged(context.comms, context.commander), + native, + "CommsUnjudgedTool / judge.Edit.unjudged", + ) diff --git a/src/agent_comms/commander_parity/rows.py b/src/agent_comms/commander_parity/rows.py index ab22a2feb2..7ac526abf0 100644 --- a/src/agent_comms/commander_parity/rows.py +++ b/src/agent_comms/commander_parity/rows.py @@ -11,6 +11,7 @@ StoryPhase, ) from . import ParityComparison, ParityRow, UnverifiedParity +from .fleet import BoardComparison, BranchComparison, OrderWorkComparison, UnjudgedComparison from .load import LoadComparison from .measurements import RuleComparison, StorageComparison, measured_load @@ -54,6 +55,10 @@ class BoardParityRow(ParityRow): native = "`CommsTodoListTool` and the fleet tool, derived" phases = (OneOrderOwnerPhase,) + @classmethod + def verify(cls) -> ParityComparison: + return BoardComparison.read() + class BranchHygieneParityRow(ParityRow): number = 6 @@ -62,6 +67,10 @@ class BranchHygieneParityRow(ParityRow): native = "`CommsWorkBranchesTool`" phases = (OneOrderOwnerPhase,) + @classmethod + def verify(cls) -> ParityComparison: + return BranchComparison.read() + class StrikeLimitParityRow(ParityRow): number = 7 @@ -102,6 +111,10 @@ class OrderWorkParityRow(ParityRow): native = "`comms_order_work`: both transcript families by `tool_call_id`" phases = (ReviewQueuePhase,) + @classmethod + def verify(cls) -> ParityComparison: + return OrderWorkComparison.read() + class UnjudgedParityRow(ParityRow): number = 12 @@ -110,6 +123,10 @@ class UnjudgedParityRow(ParityRow): native = "`comms_unjudged`: calls and unit messages after the last verdict" phases = (ReviewQueuePhase,) + @classmethod + def verify(cls) -> ParityComparison: + return UnjudgedComparison.read() + class PublicStoryParityRow(ParityRow): number = 13 @@ -118,6 +135,13 @@ class PublicStoryParityRow(ParityRow): native = "`comms_order_story`, `comms_fleet_story`; send-back reason posted once" phases = (StoryPhase,) + @classmethod + def verify(cls) -> ParityComparison: + return UnverifiedParity( + "Missing native owners CommsOrderStoryTool and CommsFleetStoryTool; " + "the narrative file alone does not prove native story or exactly-once publication." + ) + class ScheduledWakeParityRow(ParityRow): number = 14 @@ -247,6 +271,13 @@ class MomentumParityRow(ParityRow): ) phases = (CommanderPagePhase,) + @classmethod + def verify(cls) -> ParityComparison: + return UnverifiedParity( + "Missing native CommitMomentum and prediction/outcome owner; " + "there is no current prediction workflow to compare." + ) + class MeasuredLoadParityRow(ParityRow): number = 28 diff --git a/src/agent_comms/native_tools.py b/src/agent_comms/native_tools.py index d4dc9e90c4..16fc1609f2 100644 --- a/src/agent_comms/native_tools.py +++ b/src/agent_comms/native_tools.py @@ -4,6 +4,7 @@ import json from abc import abstractmethod +from collections.abc import Mapping from dataclasses import dataclass, field from pathlib import Path from typing import ClassVar @@ -71,6 +72,10 @@ def create_call( return CoordinationCall( call_id=call_id, declared_name=cls.declared_name, arguments=arguments) + def recorded_digest(self, arguments: Mapping[str, object], *, call_id: str) -> str: + """Use the same native call owner as tool_watch, with original fields and identity.""" + return self.create_call(call_id, dict(arguments)).digest() + @classmethod def native_extension(cls, package: Path) -> Path: return package / "agent-comms-extensions/global-agent-comms/index.mjs" @@ -86,6 +91,9 @@ class CoordinationCall(NativeRequestTools, NativeExecutableTool, declared_name=N def reads(self) -> str: return json.dumps(self.arguments) + def recorded_digest(self, arguments: Mapping[str, object], *, call_id: str) -> str: + return self.digest() + class ArgumentTool: """One declared execution record supplies validation and native serialization.""" diff --git a/src/agent_comms/tool_declaration.py b/src/agent_comms/tool_declaration.py index ea109bd1f3..9e369f2652 100644 --- a/src/agent_comms/tool_declaration.py +++ b/src/agent_comms/tool_declaration.py @@ -79,17 +79,16 @@ def digest(self) -> str: """What the call does in one line, enough to judge it.""" return f'{self.declared_name} {one_line(self.reads(), 260)}' + def recorded_digest(self, arguments: Mapping[str, object], *, call_id: str) -> str: + """The recorded call uses this declaration's existing presentation.""" + return self.digest() + @classmethod def caption(cls, name: str, tool: ToolDeclaration | None) -> str: """A call reads as its label, then what its record names; with no record, its label.""" reads = one_line(text := tool.reads(), len(text)) if tool else '' return cls.for_name(name).label + (' ' + reads if reads else '') - @classmethod - def call_digest(cls, name: str, tool: ToolDeclaration | None, arguments: object) -> str: - """A call in one line: its record's digest; with no record, its name and raw arguments.""" - return tool.digest() if tool else f'{name} {one_line(json.dumps(arguments), 260)}' - @classmethod def result_diff(cls, result, ok): """Opaque metadata cannot invent edit evidence.""" @@ -105,3 +104,6 @@ class UndeclaredTool(ToolDeclaration, declared_name=None): @property def label(self) -> str: return self.name + + def recorded_digest(self, arguments: Mapping[str, object], *, call_id: str) -> str: + return f'{self.name} {one_line(json.dumps(arguments), 260)}' diff --git a/src/agent_comms/transcript_events.py b/src/agent_comms/transcript_events.py index 175f29c615..44d468991c 100644 --- a/src/agent_comms/transcript_events.py +++ b/src/agent_comms/transcript_events.py @@ -368,8 +368,8 @@ def title(self) -> str: def record_call(self, calls: dict[str, OrderCall], since: float) -> None: """A call recorded before ``since`` is not work since then.""" if not self.before(since): - digest = tool_declaration.ToolDeclaration.call_digest( - self.tool_name, self.tool, self.raw_input) + declaration = self.tool or tool_declaration.UndeclaredTool(self.tool_name) + digest = declaration.recorded_digest(self.raw_input, call_id=self.tool_call_id) calls[self.tool_call_id] = OrderCall( self.tool_call_id, self.tool_name, digest, RunningCallState) diff --git a/tests/test_commander_tool_watch.py b/tests/test_commander_tool_watch.py index 55371d8d1d..689f0104c4 100644 --- a/tests/test_commander_tool_watch.py +++ b/tests/test_commander_tool_watch.py @@ -10,6 +10,8 @@ from agent_comms.native_tool_call import CodingCall from agent_comms.native_tool_call import SelectedToolDenied from agent_comms.todos import StorageStrikeTodo +from agent_comms.tool_declaration import ToolDeclaration +from agent_comms.transcript_events import ToolStartTranscript from test_commander_problem_orders import create @@ -52,6 +54,30 @@ def test_coordination_call_digests_its_arguments(): assert call.digest() == 'comms_threads {}' +def test_recorded_coordination_digest_keeps_original_request_arguments(): + arguments = {'from': 'worker', 'to': 'lead', 'body': 'the original body'} + event = ToolStartTranscript( + position=1, timestamp=1.0, tool_call_id='original-call', tool_name='comms_send', + raw_input=arguments, + tool=ToolDeclaration.record('comms_send', arguments, call_id='original-call'), + ) + calls = {} + event.record_call(calls, 0.0) + assert calls['original-call'].digest == CodingCall.decode( + 'original-call', 'comms_send', arguments).digest() + assert 'the original body' in calls['original-call'].digest + + +def test_undeclared_record_keeps_original_name_and_arguments(): + event = ToolStartTranscript( + position=1, timestamp=1.0, tool_call_id='external-call', tool_name='external_unknown', + raw_input={'original': 'argument'}, + ) + calls = {} + event.record_call(calls, 0.0) + assert calls['external-call'].digest == 'external_unknown {"original": "argument"}' + + def test_a_malformed_line_raises_instead_of_skipping_history(tool_watch, commander, commander_script): comms, _ = commander whole = json.dumps({"type": "message", "id": "m1", "timestamp": "2026-10-18T22:00:00.000Z",