P300: live board and work parity comparisons

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",
+        )

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

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."""

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)}'