diff --git a/AGENTS.md b/AGENTS.md index f23d2e004e..4907a6696a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -52,10 +52,8 @@ receipt acknowledgments. Tristan should not have to decode compressed jargon. user journey; controlled provider replies are allowed, real UI/transport/state are required. A fork check includes immediate opening and its first actual reply. - Ship substantial useful checkpoints: implementation owners merge and send - Codex the Core/Toad SHAs. Codex alone builds and installs the paired runtime, - switches its pointer and operates its restart queue. Other agents may verify - supplied builds, but must not run `relink.sh`, publication commands or - restart-queue writes, including one-off helpers. + the installer the Core/Toad SHAs. + Installs run through tools/upgrade_bus.py, by whichever unit the commander assigns, on either machine. Deferred CI and final latency targets do not hold those checkpoints. - Retain warm rendered bodies across tab returns within the existing resource limits. Scroll preparation follows velocity and destination intent, then shrinks diff --git a/docs/refactor/p79-native-caller.md b/docs/refactor/p79-native-caller.md index 2091326bf5..f209ac83c6 100644 --- a/docs/refactor/p79-native-caller.md +++ b/docs/refactor/p79-native-caller.md @@ -22,4 +22,4 @@ Before/after AST: current source/tests/tools, actual installed Core and metaclas Local source batch before test trimming: 92 passed, 6 failed. All six fail identically against untouched installed Core 21c25f49d, at wire_log.py:512 with Private bus writer has no durable protocol marker; these are retained baseline fixture failures, assigned to core-retry's existing fixture work. No caller failure is hidden. Ratchet exits 0. `git grep -n 'caller: ToolCaller = NoThreadCaller()' -- src | wc -l` returns 0 on this branch. -Acceptance is a fresh scratch-root native thread on the reviewed installed runtime: actual comms_set_goal records the goal under that thread. Source tests and ratchet are necessary but not installed acceptance. Codex alone builds/installs native and operates the fleet pointer. +Acceptance is a fresh scratch-root native thread on the reviewed installed runtime: actual comms_set_goal records the goal under that thread. Source tests and ratchet are necessary but not installed acceptance. Installs run through tools/upgrade_bus.py, by whichever unit the commander assigns, on either machine. diff --git a/src/agent_comms/candidate_maintenance.py b/src/agent_comms/candidate_maintenance.py index c2186860b6..9a0fb16953 100644 --- a/src/agent_comms/candidate_maintenance.py +++ b/src/agent_comms/candidate_maintenance.py @@ -1,28 +1,22 @@ -"""Best-effort indexing of successful appends after the outer locks release. - -Losing a hint or crashing the daemon leaves the optional index unavailable; -canonical bus and sealed SQL still drive original delivery. Byte positions -include observations as well as messages, without inventing message sequences. -""" +"""Index successful appends after outer locks release, within the owning root's lifetime.""" from __future__ import annotations import logging import sqlite3 import threading -from contextlib import ExitStack from pathlib import Path from .bus_route_counts import BusRouteCounts from .coordination_errors import CoordinationError from .routing import DeliveryMessage -from .wire_log import CurrentSnapshotOpening, WireLog from .wake_candidate_index import FirstBuild, ProjectionUnavailableError, WakeCandidateIndex +from .wire_log import CurrentSnapshotOpening, WireLog _LOG = logging.getLogger(__name__) _MAX_ACTIVE_ROOTS = 32 _guard = threading.Lock() -_pending: dict[Path, tuple[WireLog, int]] = {} +_pending: dict[Path, CandidateMaintenance] = {} class CandidateMaintenanceFailed(RuntimeError): # noqa: N818 - named worker failure @@ -33,72 +27,82 @@ def __init__(self, root: Path) -> None: super().__init__(f"Candidate maintenance failed for {root}") -def schedule_candidate_catchup(log: WireLog, through_bytes: int) -> None: - """Prepare writer indexes, then signal their existing maintenance worker.""" - if through_bytes <= 0: - raise ValueError("candidate notification needs an actual committed byte position") - BusRouteCounts(log.path).prepare() - root = log.path.parent - with _guard: - current = _pending.get(root) - if current is not None: - _pending[root] = (current[0], max(current[1], through_bytes)) - return - if len(_pending) >= _MAX_ACTIVE_ROOTS: - raise CandidateMaintenanceFailed(root) - _pending[root] = (log, through_bytes) - try: - threading.Thread( - target=_drain_candidate, - args=(root,), - name="agent-comms-candidate-maintenance", - daemon=True, - ).start() - except RuntimeError as error: - _pending.pop(root, None) - raise CandidateMaintenanceFailed(root) from error - - -def _retire_candidate(root: Path) -> None: - with _guard: - _pending.pop(root, None) +class CandidateMaintenance(threading.Thread): + """One root's coalesced cursor and actual worker; retirement joins this owner.""" + def __init__(self, log: WireLog, through_bytes: int) -> None: + super().__init__(name="agent-comms-candidate-maintenance", daemon=True) + self.log = log + self.through_bytes = through_bytes -def _drain_candidate(root: Path) -> None: - with ExitStack() as cleanup: - cleanup.callback(_retire_candidate, root) + def run(self) -> None: + root = self.log.path.parent try: - with _guard: - log, _bytes = _pending[root] - # Source ownership is checked off the producer path. Maintenance cannot - # create the protocol marker or authorize delivery. - with log.checked_read() as source: + with self.log.checked_read() as source: metadata = source.marker root_id = metadata.root_id - index = WakeCandidateIndex(log) - routes = BusRouteCounts(log.path) + index = WakeCandidateIndex(self.log) + routes = BusRouteCounts(self.log.path) while True: with _guard: - target = _pending[root][1] + target = self.through_bytes hint = index.notify_committed_append(root_id=root_id, through_bytes=target) - result = index.catch_up_committed_append( - hint, start=FirstBuild() - ) - with log.opened_wire_snapshot(opening=CurrentSnapshotOpening()) as snapshot: + result = index.catch_up_committed_append(hint, start=FirstBuild()) + with self.log.opened_wire_snapshot(opening=CurrentSnapshotOpening()) as snapshot: route_result = routes.sync( snapshot, - lambda record: DeliveryMessage.from_wire(record, root_id=snapshot.metadata.root_id), + lambda record: DeliveryMessage.from_wire( + record, root_id=snapshot.metadata.root_id + ), ) if result.caught_up and route_result.complete: with _guard: - if _pending[root][1] <= result.checkpoint_bytes: - # Disarm cleanup before a successor can own this root. - cleanup.pop_all() + if self.through_bytes <= result.checkpoint_bytes: _pending.pop(root) return except ( - OSError, ValueError, sqlite3.Error, CoordinationError, ProjectionUnavailableError, + OSError, + ValueError, + sqlite3.Error, + CoordinationError, + ProjectionUnavailableError, ) as error: failure = CandidateMaintenanceFailed(root) _LOG.exception("%s", failure) raise failure from error + finally: + with _guard: + if root in _pending and _pending[root] is self: + _pending.pop(root) + + +def schedule_candidate_catchup(log: WireLog, through_bytes: int) -> None: + """Prepare writer indexes, then coalesce into the root's actual worker.""" + if through_bytes <= 0: + raise ValueError("candidate notification needs an actual committed byte position") + BusRouteCounts(log.path).prepare() + root = log.path.parent + with _guard: + if root in _pending: + worker = _pending[root] + worker.through_bytes = max(worker.through_bytes, through_bytes) + return + if len(_pending) >= _MAX_ACTIVE_ROOTS: + raise CandidateMaintenanceFailed(root) + worker = CandidateMaintenance(log, through_bytes) + _pending[root] = worker + try: + worker.start() + except RuntimeError as error: + _pending.pop(root) + raise CandidateMaintenanceFailed(root) from error + + +def drain_candidate_catchup(root: Path) -> None: + """A stopped root cannot be renamed or removed while its worker still writes.""" + while True: + with _guard: + if root not in _pending: + return + worker = _pending[root] + worker.join() diff --git a/src/agent_comms/check_root.py b/src/agent_comms/check_root.py index 02de2fb15a..7ace5f24d1 100644 --- a/src/agent_comms/check_root.py +++ b/src/agent_comms/check_root.py @@ -12,6 +12,7 @@ from typing import ClassVar from .attempt_recovery import RecoveryMonitorCapability +from .candidate_maintenance import drain_candidate_catchup from .child_process import NoProcess from .cli_result import CliResult from .comms import Comms @@ -74,6 +75,7 @@ def open(self) -> Iterator[Comms]: ): copied.owners.stop(thread.name) finally: + drain_candidate_catchup(root) RecoveryMonitorCapability.retire_check_root(root) def _populate(self, copied: Comms) -> None: diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index f8a8137f7d..19dcab0564 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -44,7 +44,7 @@ from ..goals import OrderCheck, ProblemGoalOrder from ..message_reference import MessageReference from ..native_tool_call import SelectedToolDenied -from ..plan_phases import OrderPhase +from ..plan_phases import OrderPhase, PlanPhase from ..progress_claims import MergedClaim, WorkingClaim from ..recorded_time import RecordedEventTime from ..repo import Repo @@ -169,6 +169,7 @@ class OrderFile: order: PendingProblemOrder defect: str why_stupid: str + unplanned_reason: str = "" problems: tuple[str, ...] = () slop: tuple[Slop, ...] = () @@ -181,7 +182,8 @@ def read(cls, text: str): @property def prose(self) -> str: - return f'{self.defect}\n{self.why_stupid}' + reason = f'\nUnplanned reason: {self.unplanned_reason}' if self.unplanned_reason else '' + return f'{self.defect}\n{self.why_stupid}{reason}' def problems_found(self, clean: bool) -> list[str]: problems = [f'{name}: {error}; fix: state the defect plainly without wording rules on code' @@ -325,6 +327,20 @@ def send(self, unit, due, document: OrderFile, clean=False, reassign=None, new=N problems.append('due: fix: supply HH:MM local time (00:00 through 23:59)') group = [new[0]] if new else list(document.problems) listed = {t.id: t for t in self.todos()} + history = self.comms.goals.todos.history_for_commander(self.incarnation) + problems.extend(order.phase.order_refusals(history, document.unplanned_reason)) + held = tuple(todo for todo in listed.values() if todo.holder) + busy = tuple(todo for plan in PlanPhase.members_with(PlanPhase) + for todo in plan.serving(held) if todo.holder == unit and todo.id not in group) + if busy: + snapshot = self.comms.registry.snapshot() + occupied = {todo.holder for todo in held} | {self.commander} | self.parked + idle = sorted(thread.name for thread in snapshot.threads.values() + if thread.name not in occupied and thread.role.executor(thread) + and snapshot.statuses[thread.name].active) + problems.append(f'{unit} already holds plan orders ' + f'{", ".join(todo.id for todo in busy)}; ' + f'running units holding no order: {", ".join(idle) or "none"}') if not group or any(not re.fullmatch(r'P[1-9]\d*', pid) for pid in group): problems.append('problems: fix: supply an array of P IDs, or use order problem') if new and document.problems: @@ -480,8 +496,7 @@ def problem(self, order_file, unit, due, where, disk=False, def close(self, pid, works, artifact=''): """Core owns closing: the installed evidence, the artifact and the no-open-order rule live on the Todo.""" evidence = artifact or self.comms.goals.todos.get(pid).order.closing_artifact() - ThreadCaller(self.commander).invoke(CommsProblemCloseTool(problem=pid, - works=works, artifact=evidence), self.comms) + self.call(CommsProblemCloseTool(problem=pid, works=works, artifact=evidence)) self.write_board() print(f'{pid} closed') @@ -802,6 +817,23 @@ def parse(cls, argv: list[str], description: str) -> CommanderVerb: return cls.member_named(args.command).from_arguments(args) +class StaleOrderGate(CommanderVerb): + """All directive verbs read the configured commander's same retained board.""" + + def gate(self, ctx: Commander) -> None: + super().gate(ctx) + stale = tuple(todo for todo in ctx.held_orders() if todo.overdue(time.time() - 30 * 60)) + if stale: + lines = [] + for todo in stale: + due = time.localtime(todo.deadline.recorded_at("An order deadline")) + lines.append(f'{todo.id}: holder {todo.holder}, deadline ' + f'{time.strftime("%Y-%m-%d %H:%M", due)}') + sys.exit('REFUSED, nothing recorded: orders more than 30 minutes overdue:\n ' + + '\n '.join(lines) + + '\nMove their deadlines or withdraw them before sending more work.') + + class RuledVerb(CommanderVerb): """HARD RULE: no commander action, mine or a fork's, on rules not re-read in 20 min.""" @@ -855,7 +887,7 @@ class OrderVerb(CommanderVerb): @dataclass(frozen=True, kw_only=True) -class SendVerb(OrderVerb, DirectiveVerb): +class SendVerb(OrderVerb, DirectiveVerb, StaleOrderGate): help = "record the order in Core and send it" unit: str = option("unit") due: str = option("due", help="HH:MM local time") @@ -874,7 +906,7 @@ def run(self, ctx: Commander) -> None: @dataclass(frozen=True, kw_only=True) -class StrikeVerb(OrderVerb, DirectiveVerb): +class StrikeVerb(OrderVerb, DirectiveVerb, StaleOrderGate): help = "a send-back, a missed deadline or a false claim" unit: str = option("unit") why: str = option("--why") @@ -971,9 +1003,11 @@ class PullRequestVerb(OrderVerb, declared_name="pr"): def run(self, ctx: Commander) -> None: todo = ctx.comms.goals.todos.get(self.pid) - checkout = ctx.comms.registry.storage_rule.order_checkout(todo.holder, todo.id) + checkout = Repo.of(Path.cwd()) + if checkout.github.lower() != todo.repo.lower(): + sys.exit(f'Refused: current checkout belongs to {checkout.github}, not {todo.repo}.') print(todo.order.open_pull_request( - todo, checkout, self.title, Path(self.body_file).read_text())) + todo, checkout.checkout, self.title, Path(self.body_file).read_text())) @dataclass(frozen=True, kw_only=True) @@ -999,7 +1033,7 @@ def run(self, ctx: Commander) -> None: @dataclass(frozen=True, kw_only=True) -class SayVerb(OrderVerb, DirectiveVerb): +class SayVerb(OrderVerb, DirectiveVerb, StaleOrderGate): help = "a non-order message; refused if it orders work" units: str = option("units", metavar="UNIT[,UNIT]") body_file: str = option("body_file", metavar="BODYFILE") @@ -1009,7 +1043,7 @@ def run(self, ctx: Commander) -> None: @dataclass(frozen=True, kw_only=True) -class ProblemVerb(OrderVerb, DirectiveVerb): +class ProblemVerb(OrderVerb, DirectiveVerb, StaleOrderGate): help = "Core records the problem with its order, or nothing" order: str = option("order", metavar="ORDER.json") to: str = option("--to") diff --git a/src/agent_comms/commander_parity/__init__.py b/src/agent_comms/commander_parity/__init__.py index 22073f86a4..efe7d431bc 100644 --- a/src/agent_comms/commander_parity/__init__.py +++ b/src/agent_comms/commander_parity/__init__.py @@ -3,7 +3,7 @@ from __future__ import annotations from abc import ABC, abstractmethod -from collections.abc import Iterable +from collections.abc import Callable, Iterable from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, ClassVar @@ -43,6 +43,7 @@ class ParityRow( current: ClassVar[str] native: ClassVar[str] phases: ClassVar[tuple[type[PlanPhase], ...]] + comparison: ClassVar[Callable[[], ParityComparison]] = UnverifiedParity @classmethod def compare(cls, todos: Iterable[Todo]) -> ParityComparison: @@ -51,7 +52,7 @@ def compare(cls, todos: Iterable[Todo]) -> ParityComparison: @classmethod def verify(cls) -> ParityComparison: """Run this declared workflow now; a run is not durable cutover authority.""" - return UnverifiedParity() + return cls.comparison() @classmethod def rows(cls) -> tuple[type[ParityRow], ...]: diff --git a/src/agent_comms/commander_parity/lifecycle.py b/src/agent_comms/commander_parity/lifecycle.py new file mode 100644 index 0000000000..2c0e441cb0 --- /dev/null +++ b/src/agent_comms/commander_parity/lifecycle.py @@ -0,0 +1,487 @@ +"""Exercise current order commands against real native owners on a selected bus copy.""" + +from __future__ import annotations + +import contextlib +import io +import json +import os +import shlex +import sys +import time +from abc import ABC, abstractmethod +from collections.abc import Iterator +from dataclasses import dataclass, replace +from pathlib import Path +from typing import ClassVar +from uuid import uuid4 + +from pydantic import PrivateAttr + +from ..change_report import FileLine +from ..check_root import CheckRoot +from ..commander.config import CommanderConfig +from ..commander.order import Commander, OrderFile, hhmm +from ..commander.request import RequestAddVerb +from ..field_codec import FieldCodec +from ..goals import CountCheckResult, OrderCheck, ProblemGoalOrder +from ..todos import ( + ApplicationOrderRun, + MergedProgress, + Moving, + NoCommitProgress, + PendingProblemOrder, + Todo, +) +from ..tool_caller import ThreadCaller, ToolCaller +from ..tool_result import ToolResult +from ..tools import ( + CommsChangeProblemTool, + CommsFireTool, + CommsOrderTool, + CommsProblemCloseTool, + CommsTodoCreateTool, + CommsTodoListTool, + ProblemOrdered, + TodoChanged, + TodoList, + ToolRequest, +) +from . import ParityComparison, UnverifiedParity + + +@dataclass(frozen=True) +class LifecycleObservation(ParityComparison): + owner: str + invocations: int + checks: tuple[tuple[str, bool], ...] + output: str + + @property + def passed(self) -> bool: + return bool(self.invocations and self.checks) and all(value for _, value in self.checks) + + def describe(self) -> str: + return ( + f"owner={self.owner}; calls={self.invocations}; " + f"checks={dict(self.checks)}; output={self.output!r}" + ) + + +type LifecycleRequest = ToolRequest[ToolCaller] | ToolRequest[ThreadCaller] + + +@dataclass(frozen=True) +class NativeInvocation: + request: LifecycleRequest + result: ToolResult + + +class MeasuredCommander(Commander): + """Keep actual native returns at the existing command boundary, not frame locals.""" + + _invocations: list[NativeInvocation] = PrivateAttr(default_factory=list) + + @property + def invocations(self) -> tuple[NativeInvocation, ...]: + return tuple(self._invocations) + + def call(self, tool: LifecycleRequest) -> ToolResult: + result: ToolResult = super().call(tool) + self._invocations.append(NativeInvocation(tool, result)) + return result + + +@dataclass(frozen=True) +class LifecycleRun: + commander: MeasuredCommander + worker: str + decisions: Path + document: Path + artifact: Path + + @classmethod + @contextlib.contextmanager + def open(cls, *, extra_threads: tuple[str, ...] = ()) -> Iterator[LifecycleRun]: + config = MeasuredCommander.read(CommanderConfig.path()) + source = config.comms + actor = ToolCaller.from_actor(os.environ.get("AGENT_COMMS_THREAD")).acting_name( + config.commander + ) + caller = source.registry.require(actor) + if actor == config.commander: + held = source.goals.todos.held_for_commander(config.incarnation) + if not held: + raise ValueError("Lifecycle comparison requires an existing native order holder") + worker = source.registry.require(held[-1].held().owner) + else: + worker = caller + directory = caller.storage_roots.scratch / "commander-parity" / uuid4().hex + threads = tuple(dict.fromkeys((config.commander, worker.name, *extra_threads))) + with CheckRoot(source, threads, directory).open() as copied: + root = copied.root + commander = MeasuredCommander( + commander=config.commander, + root=root, + cli=config.cli, + state=root / "commander-state", + board=root / "board.md", + narrative=root / "narrative.md", + repos=config.repos, + worktrees=root / "projects", + scratch=root / "scratch", + disk=root, + ) + with contextlib.redirect_stdout(io.StringIO()): + RequestAddVerb( + text=( + "Compare current order commands with their real native owners in this " + "disposable copied bus, preserving the source bus and its sessions." + ) + ).run(commander) + decisions = root / "decisions.json" + decisions.write_text(json.dumps([worker.name])) + command = shlex.join( + ( + sys.executable, + "-c", + "import json;from pathlib import Path;" + f"print(len(json.loads(Path({str(decisions)!r}).read_text())))", + ) + ) + order = PendingProblemOrder( + owner=FileLine(str(Path(__file__).resolve()), 1), + check=OrderCheck( + str(Path(__file__).resolve().parents[3]), command, CountCheckResult(0) + ), + edits=( + "Remove the copied workflow decision and verify its native order completes.", + ), + banned=("Do not change the source bus or its saved sessions.",), + report="Report the actual copied goal and native order state.", + run=ApplicationOrderRun( + "Read the copied native goal and confirm its recorded source location " + "and assigned worker." + ), + ) + document = root / "order.json" + document.write_text( + json.dumps( + FieldCodec.encode( + OrderFile( + order, + "The copied workflow decision prevents the native completion check " + "from reaching zero.", + "This is stupid because an unfinished copied decision must not count " + "as delivered work.", + ) + ) + ) + ) + yield cls(commander, worker.name, decisions, document, root / "application.json") + + @property + def due(self) -> str: + return time.strftime("%H:%M", time.localtime(time.time() + 3600)) + + def create(self) -> Todo: + self.commander.problem( + str(self.document), self.worker, self.due, f"{Path(__file__).resolve()}:1" + ) + rows = tuple(self.commander.comms.goals.todos.for_commander(self.commander.incarnation)) + if len(rows) != 1: + raise ValueError(f"A fresh copied workflow must create one problem, got {len(rows)}") + return rows[0] + + def complete(self, todo: Todo) -> Todo: + goal = self.commander.comms.registry.require(self.worker).goal + self.artifact.write_text(json.dumps(FieldCodec.encode(goal))) + self.decisions.write_text("[]") + self.commander.done(todo.id, "op", seen=str(self.artifact)) + return self.commander.comms.goals.todos.get(todo.id) + + +class LifecycleCase(ABC): + native_owner: ClassVar[type[LifecycleRequest]] + additional_owners: ClassVar[tuple[type[LifecycleRequest], ...]] = () + + @classmethod + @abstractmethod + def exercise(cls, run: LifecycleRun) -> tuple[Todo, ...]: ... + + @classmethod + @abstractmethod + def checks( + cls, run: LifecycleRun, rows: tuple[Todo, ...], native: tuple[ToolResult, ...], output: str + ) -> tuple[tuple[str, bool], ...]: ... + + @classmethod + def read(cls) -> ParityComparison: + owners = (cls.native_owner, *cls.additional_owners) + with LifecycleRun.open() as run: + output = io.StringIO() + with contextlib.redirect_stdout(output): + rows = cls.exercise(run) + native = tuple( + invocation.result + for invocation in run.commander.invocations + if isinstance(invocation.request, owners) + ) + text = output.getvalue() + return LifecycleObservation( + ", ".join(f"{owner.__module__}.{owner.__qualname__}" for owner in owners), + len(native), + cls.checks(run, rows, native, text), + text, + ) + + +class CreateProblemCase(LifecycleCase): + native_owner: ClassVar[type[ToolRequest[ThreadCaller]]] = CommsTodoCreateTool + + @classmethod + def exercise(cls, run: LifecycleRun) -> tuple[Todo, ...]: + return (run.create(),) + + @classmethod + def checks( + cls, run: LifecycleRun, rows: tuple[Todo, ...], native: tuple[ToolResult, ...], output: str + ) -> tuple[tuple[str, bool], ...]: + if not native or not isinstance(native[-1], ProblemOrdered): + return (("native order result returned", False),) + result, saved = native[-1], rows[-1] + goal = run.commander.comms.registry.require(run.worker).goal + return ( + ("printed native identity", output.rstrip().endswith(cls.printed_identity(result))), + ( + "native source retained", + saved.order.checked().owner == result.todo.order.checked().owner, + ), + ("native assignment retained", saved.assignment == result.todo.assignment), + ("native goal published", goal.require().id == result.goal.id), + ) + + @classmethod + def printed_identity(cls, result: ProblemOrdered) -> str: + return result.todo.id + + +class AssignOrderCase(CreateProblemCase): + native_owner = CommsOrderTool + + @classmethod + def exercise(cls, run: LifecycleRun) -> tuple[Todo, ...]: + before = run.create() + run.commander.fire( + run.worker, "The copied workflow is released before its next native assignment." + ) + document = replace(OrderFile.read(run.document.read_text()), problems=(before.id,)) + run.commander.send(run.worker, run.due, document, clean=True) + return before, run.commander.comms.goals.todos.get(before.id) + + @classmethod + def checks( + cls, run: LifecycleRun, rows: tuple[Todo, ...], native: tuple[ToolResult, ...], output: str + ) -> tuple[tuple[str, bool], ...]: + inherited = super().checks(run, rows, native, output) + return ( + *inherited, + ("same problem reassigned", rows[0].id == rows[-1].id), + ("prior assignment retained", bool(rows[-1].released)), + ) + + @classmethod + def printed_identity(cls, result: ProblemOrdered) -> str: + return result.todo.order.checked().id + + +class InstalledEvidenceCase(LifecycleCase): + native_owner = CommsProblemCloseTool + + @classmethod + def exercise(cls, run: LifecycleRun) -> tuple[Todo, ...]: + created = run.create() + delivered = run.complete(created) + run.commander.close( + created.id, + "Read the copied native goal and confirm its recorded source location " + "and assigned worker.", + str(run.artifact), + ) + return delivered, run.commander.comms.goals.todos.get(created.id) + + @classmethod + def checks( + cls, run: LifecycleRun, rows: tuple[Todo, ...], native: tuple[ToolResult, ...], output: str + ) -> tuple[tuple[str, bool], ...]: + return ( + ("delivery alone unaccepted", not rows[0].order.accepted), + ("installed evidence accepted", rows[-1].order.accepted), + ("closed in native store", not rows[-1].state.open), + ("actual artifact exists", run.artifact.is_file()), + ("current close printed", f"{rows[-1].id} closed" in output), + ) + + +class BuildBoundProofCase(InstalledEvidenceCase): + @classmethod + def read(cls) -> ParityComparison: + closing = super().read() + return UnverifiedParity( + "Missing native owner: a build-bound application-run record linking the " + "runtime and Core/Toad revisions to closing evidence. The actual close " + "owner currently accepts a bare artifact path. " + closing.describe() + ) + + +class StrikeLimitCase(LifecycleCase): + native_owner = CommsChangeProblemTool + additional_owners = (CommsFireTool,) + + @classmethod + def exercise(cls, run: LifecycleRun) -> tuple[Todo, ...]: + created = run.create() + for number in range(ProblemGoalOrder.StrikeLimit): + run.commander.strike( + run.worker, + created.id, + "The copied decision remains unfinished at its recorded review " + f"number {number + 1}.", + ) + run.commander.fire( + run.worker, "The copied holder exhausted its declared live strikes on this problem." + ) + return created, run.commander.comms.goals.todos.get(created.id) + + @classmethod + def checks( + cls, run: LifecycleRun, rows: tuple[Todo, ...], native: tuple[ToolResult, ...], output: str + ) -> tuple[tuple[str, bool], ...]: + saved = rows[-1] + return ( + ("declared limit recorded", saved.strike_count == ProblemGoalOrder.StrikeLimit), + ("one native fire", sum(isinstance(result, TodoList) for result in native) == 1), + ( + "assignment released", + saved.id + not in { + todo.id + for todo in run.commander.comms.goals.todos.held_for_commander( + run.commander.incarnation + ) + }, + ), + ("current strike count printed", f"{ProblemGoalOrder.StrikeLimit} strike(s)" in output), + ) + + +class OrderProgressMutationCase(LifecycleCase): + native_owner = CommsChangeProblemTool + + @classmethod + def exercise(cls, run: LifecycleRun) -> tuple[Todo, ...]: + before = run.create() + moved = time.strftime("%H:%M", time.localtime(time.time() + 7200)) + run.commander.due( + before.id, + moved, + "The copied workflow needs time to record its actual native application evidence.", + ) + return before, run.commander.comms.goals.todos.get(before.id) + + @classmethod + def checks( + cls, run: LifecycleRun, rows: tuple[Todo, ...], native: tuple[ToolResult, ...], output: str + ) -> tuple[tuple[str, bool], ...]: + before, after = rows + messages = run.commander.comms.bus.inbox(run.worker) + unit = run.commander.comms.registry.require(run.worker) + goal = unit.goal.require() + instructions = goal.order.instructions( + goal.order.todos(run.commander.comms.goals, unit.incarnation, goal.id) + ) + native_deadlines = tuple( + result.todo.deadline for result in native if isinstance(result, TodoChanged) + ) + return ( + ( + "native deadline retained", + bool(native_deadlines) and native_deadlines[-1] == after.deadline, + ), + ("deadline actually changed", before.deadline != after.deadline), + ("assignment unchanged", before.assignment == after.assignment), + ("unit got the actual update", bool(messages)), + ( + "unit messages hide clocks", + all( + hhmm(clock) not in message.body + for clock in (before.deadline, after.deadline) + for message in messages + ), + ), + ( + "unit goal hides deadline", + "Deadline:" not in instructions and hhmm(after.deadline) not in instructions, + ), + ("commander still sees deadline", f"due {hhmm(after.deadline)}" in output), + ) + + +@dataclass(frozen=True) +class ProgressComparison(ParityComparison): + current: dict[str, str] + native: dict[str, str] + first_seconds: int + later_seconds: int + merged_no_strikes: tuple[bool, ...] + workflow: ParityComparison + + @property + def passed(self) -> bool: + return all( + ( + self.workflow.passed, + bool(self.native), + self.current == self.native, + self.first_seconds == 600, + self.later_seconds == 900, + bool(self.merged_no_strikes), + all(self.merged_no_strikes), + ) + ) + + def describe(self) -> str: + return ( + f"{self.workflow.describe()}; live_progress={len(self.native)}; " + f"read_equal={self.current == self.native}; " + f"intervals={self.first_seconds}/{self.later_seconds}; " + f"live_merged_no_strikes={self.merged_no_strikes}" + ) + + +def verify_order_progress() -> ParityComparison: + context = MeasuredCommander.read(CommanderConfig.path()) + current = {key: value.line for key, value in context.progress().items()} + results = tuple( + invocation.result + for invocation in context.invocations + if isinstance(invocation.request, CommsTodoListTool) + ) + if len(results) != 1 or not isinstance(results[0], TodoList): + return UnverifiedParity("Current progress did not return one native TodoList measurement") + result = results[0] + observed = {key: value.line for key, value in result.progress.items()} + todos = {todo.id: todo for todo in result.todos} + merged = tuple( + not progress.stalled(todos[pid].order.checked().at.recorded_at("An order"), time.time()) + for pid, progress in result.progress.items() + if isinstance(progress, MergedProgress) + ) + return ProgressComparison( + current, + observed, + NoCommitProgress.seconds, + Moving.seconds, + merged, + OrderProgressMutationCase.read(), + ) diff --git a/src/agent_comms/commander_parity/rows.py b/src/agent_comms/commander_parity/rows.py index ab22a2feb2..3772660101 100644 --- a/src/agent_comms/commander_parity/rows.py +++ b/src/agent_comms/commander_parity/rows.py @@ -11,8 +11,17 @@ StoryPhase, ) from . import ParityComparison, ParityRow, UnverifiedParity +from .lifecycle import ( + AssignOrderCase, + BuildBoundProofCase, + CreateProblemCase, + InstalledEvidenceCase, + StrikeLimitCase, + verify_order_progress, +) from .load import LoadComparison from .measurements import RuleComparison, StorageComparison, measured_load +from .story import StoryCase class CreateProblemParityRow(ParityRow): @@ -22,6 +31,8 @@ class CreateProblemParityRow(ParityRow): native = "`CommsTodoCreateTool`" phases = (OneOrderOwnerPhase,) + comparison = CreateProblemCase.read + class AssignOrderParityRow(ParityRow): number = 2 @@ -30,6 +41,8 @@ class AssignOrderParityRow(ParityRow): native = "`CommsOrderTool`, `CommsTodoTool` assignment and release" phases = (OneOrderOwnerPhase,) + comparison = AssignOrderCase.read + class DeliveredCommitParityRow(ParityRow): number = 3 @@ -46,6 +59,8 @@ class InstalledEvidenceParityRow(ParityRow): native = "`CommsProblemCloseTool`" phases = (OneOrderOwnerPhase,) + comparison = InstalledEvidenceCase.read + class BoardParityRow(ParityRow): number = 5 @@ -70,6 +85,8 @@ class StrikeLimitParityRow(ParityRow): native = "`Strike` rows, author-declared `StrikeLimit`, one `comms_fire`" phases = (RecordPlanePhase,) + comparison = StrikeLimitCase.read + class VerdictParityRow(ParityRow): number = 8 @@ -117,6 +134,7 @@ class PublicStoryParityRow(ParityRow): current = "`fleet-narrative.md`; `judge.py roast`" native = "`comms_order_story`, `comms_fleet_story`; send-back reason posted once" phases = (StoryPhase,) + comparison = StoryCase.read class ScheduledWakeParityRow(ParityRow): @@ -278,6 +296,8 @@ class BuildBoundProofParityRow(ParityRow): native = "Build run records runtime and Core/Toad revisions; close requires delivered commit" phases = (RecordPlanePhase,) + comparison = BuildBoundProofCase.read + class DeletionTargetParityRow(ParityRow): number = 31 @@ -312,6 +332,8 @@ class OrderProgressParityRow(ParityRow): ) phases = (OneOrderOwnerPhase, ControlPlanePhase) + comparison = staticmethod(verify_order_progress) + class MultipleCommandersParityRow(ParityRow): number = 34 diff --git a/src/agent_comms/commander_parity/story.py b/src/agent_comms/commander_parity/story.py new file mode 100644 index 0000000000..e38256d920 --- /dev/null +++ b/src/agent_comms/commander_parity/story.py @@ -0,0 +1,103 @@ +"""Row 13 runs the shared copied lifecycle, then reads its original Todo facts.""" + +from dataclasses import replace + +from ..commander.order import OrderFile +from ..goal_actions import CreateProblemGoalAction +from ..order_story import FleetStory, OrderStory +from ..todos import SendBackVerdict, StrikeReason, Todo +from ..tool_result import ToolResult +from ..tools import ( + CommsFleetStoryTool, + CommsOrderStoryTool, + CommsTodoCreateTool, + CommsVerdictTool, +) +from .lifecycle import LifecycleCase, LifecycleRun + + +class StoryCase(LifecycleCase): + native_owner = CommsOrderStoryTool + additional_owners = (CommsFleetStoryTool,) + strike_reason = "The copied worker failed to remove the first unfinished decision." + back_reason = "The copied worker duplicated the decision instead of repairing its owner." + + @classmethod + def exercise(cls, run: LifecycleRun) -> tuple[Todo, ...]: + created = run.create() + verdict = SendBackVerdict( + created.revision, + run.commander.incarnation, + created.held(), + "story-review", + StrikeReason(cls.back_reason), + ) + run.commander.call(CommsVerdictTool(problem=created.id, operation=verdict)) + document = replace(OrderFile.read(run.document.read_text()), problems=(created.id,)) + sent_back = run.commander.comms.goals.todos.get(created.id) + run.commander.send( + run.worker, + run.due, + document, + clean=True, + replaces=sent_back.messages[-1].message_id, + ) + run.commander.strike(run.worker, created.id, cls.strike_reason) + saved = run.commander.comms.goals.todos.get(created.id) + run.commander.call( + CommsTodoCreateTool( + owner=run.worker, + order=CreateProblemGoalAction( + problem="P2", + order=replace(document.order, at=saved.order.checked().at), + request=run.commander.comms.registry.require(run.commander.commander) + .goal.require() + .id, + deadline=saved.deadline, + commander=run.commander.commander, + repo=saved.repo, + text="The second problem shares the same real sent order.", + ), + ) + ) + for _ in range(2): + story = run.commander.call(CommsOrderStoryTool(order_id=saved.order.checked().id)) + assert isinstance(story, OrderStory) + print(story.line) + fleet = run.commander.call(CommsFleetStoryTool()) + assert isinstance(fleet, FleetStory) + print("\n".join(fleet.lines)) + return created, saved, run.commander.comms.goals.todos.get(created.id) + + @classmethod + def checks( + cls, run: LifecycleRun, rows: tuple[Todo, ...], native: tuple[ToolResult, ...], output: str + ) -> tuple[tuple[str, bool], ...]: + orders = tuple(result for result in native if isinstance(result, OrderStory)) + fleets = tuple(result for result in native if isinstance(result, FleetStory)) + return ( + ("two actual order reads", len(orders) == 2), + ("one actual fleet read", len(fleets) == 1), + ("record unchanged by reads", rows[-2] == rows[-1]), + ("standalone and implied strike saved", len(rows[-1].strikes) == 2), + ("send-back saved once", len(rows[-1].verdicts) == 1), + ("order reads stable", len(orders) == 2 and orders[0] == orders[1]), + ( + "linked problems share one order story", + bool(orders) + and tuple(problem.problem for problem in orders[-1].problems) + == (rows[-1].id, "P2"), + ), + ( + "original problem history retained", + bool(orders) and orders[-1].problems[0] == rows[-1].story(), + ), + ("strike reason once", bool(orders) and orders[-1].line.count(cls.strike_reason) == 1), + ("send-back reason once", bool(orders) and orders[-1].line.count(cls.back_reason) == 1), + ("one fleet line per open order", bool(fleets) and len(fleets[-1].lines) == 1), + ( + "fleet agrees with order read", + bool(fleets and orders) and fleets[-1].lines == (orders[-1].line,), + ), + ("actual output printed", bool(orders) and orders[-1].line in output), + ) diff --git a/src/agent_comms/order_story.py b/src/agent_comms/order_story.py new file mode 100644 index 0000000000..660473b408 --- /dev/null +++ b/src/agent_comms/order_story.py @@ -0,0 +1,70 @@ +"""Transient story views rendered by the original Todo record owners.""" + +from __future__ import annotations + +from abc import abstractmethod +from collections.abc import Iterable +from dataclasses import dataclass +from typing import TYPE_CHECKING, Self + +from .recorded_time import EventTime +from .tool_result import ToolResult + +if TYPE_CHECKING: + from .todos import Todo + + +@dataclass(frozen=True) +class StoryLine: + at: EventTime + text: str + + +class StorySource: + def story(self) -> tuple[StoryLine, ...]: + return () + + +class StoryEvent(StorySource): + at: EventTime + + @abstractmethod + def story_text(self) -> str: ... + + def story(self) -> tuple[StoryLine, ...]: + return (StoryLine(self.at, self.story_text()),) + + +@dataclass(frozen=True) +class ProblemStory: + problem: str + state: str + events: tuple[StoryLine, ...] + + @property + def line(self) -> str: + return f"{self.problem}: {self.state}; " + "; ".join( + " ".join(event.text.split()) for event in self.events + ) + + +@dataclass(frozen=True) +class OrderStory(ToolResult): + order_id: str + problems: tuple[ProblemStory, ...] + + @classmethod + def collect(cls, todos: Iterable[Todo]) -> tuple[Self, ...]: + grouped: dict[str, list[ProblemStory]] = {} + for todo in todos: + grouped.setdefault(todo.order.checked().id, []).append(todo.story()) + return tuple(cls(order_id, tuple(problems)) for order_id, problems in grouped.items()) + + @property + def line(self) -> str: + return " | ".join(problem.line for problem in self.problems) + + +@dataclass(frozen=True) +class FleetStory(ToolResult): + lines: tuple[str, ...] diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 04865c669b..50676f12ab 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -23,9 +23,23 @@ class OrderPhase(DeclaredFamily, affix="Phase"): """What an order serves: one plan phase, or nothing in the plan.""" + @classmethod + def order_refusals(cls, todos: Iterable[Todo], reason: str) -> tuple[str, ...]: + return () + class UnplannedPhase(OrderPhase): - pass + @classmethod + def order_refusals(cls, todos: Iterable[Todo], reason: str) -> tuple[str, ...]: + rows = tuple(todos) + ready = tuple(phase for phase in PlanPhase.members_with(PlanPhase) if phase.ready(rows)) + if ready and len(reason.split()) < 8: + return ( + "Plan phases are ready: " + + ", ".join(phase.declared_name for phase in ready) + + ". Name a phase, or provide an explicit unplanned_reason of 8+ words.", + ) + return () class PlanPhase(OrderPhase, declared_name=None): diff --git a/src/agent_comms/recorded_time.py b/src/agent_comms/recorded_time.py index b4464897df..75a073e15e 100644 --- a/src/agent_comms/recorded_time.py +++ b/src/agent_comms/recorded_time.py @@ -57,6 +57,10 @@ def capture(cls, timestamp: float) -> EventTime: """Capture a recorded Unix-seconds fact, never the read time.""" return RecordedEventTime(timestamp) + def chronology(self) -> tuple[int, float]: + """Undated facts sort after recorded events, without inventing a timestamp.""" + return (1, 0.0) + def since(self, timestamp: float) -> bool: return False @@ -107,6 +111,9 @@ def __post_init__(self): def require_recorded(self, label: str) -> None: pass + def chronology(self) -> tuple[int, float]: + return (0, self.at) + def since(self, timestamp: float) -> bool: return self.at >= timestamp diff --git a/src/agent_comms/repo.py b/src/agent_comms/repo.py index b6cfe20dbb..79df05b074 100644 --- a/src/agent_comms/repo.py +++ b/src/agent_comms/repo.py @@ -17,10 +17,10 @@ class Repo(BaseModel): github: str source: str = "src" - def contains(self, sha: str) -> bool: - """Only a successful ancestry proof in this installed checkout permits closure.""" + def contains(self, sha: str, revision: str = "HEAD") -> bool: + """Prove ancestry in the requested fetched revision without moving the checkout.""" result = subprocess.run( - ("git", "-C", str(self.checkout), "merge-base", "--is-ancestor", sha, "HEAD"), + ("git", "-C", str(self.checkout), "merge-base", "--is-ancestor", sha, revision), capture_output=True, timeout=60, ) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index efb2744795..a1122786a5 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -41,6 +41,7 @@ ) from .channel_targets import CommandTarget from .goals import CheckedCommit, CountCheckResult, OrderCheck, ProblemGoalOrder +from .order_story import ProblemStory, StoryEvent, StoryLine, StorySource from .recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime from .thread_identity import RecordedThreadIncarnation, StallWatcherIncarnation, ThreadIncarnation from .threads import Thread @@ -79,7 +80,7 @@ def artifact(self) -> str: @dataclass(frozen=True, kw_only=True) -class OrderDelivery(DeclaredFamily, affix="Delivery"): +class OrderDelivery(DeclaredFamily, StoryEvent, affix="Delivery"): at: EventTime = field(default_factory=UnrecordedEventTime) evidence: DeliveryEvidence = field(default_factory=NoDeliveryEvidence, metadata={"wire_omit_default": True}) @@ -92,6 +93,9 @@ def work(self) -> str: def line(self) -> str: return self.work + def story_text(self) -> str: + return self.work + async def run(self, check, *command): result = await BoundedRun.run(tuple(command), cwd=check.directory, timeout=120) if result.outcome != ExitedOutcome(0): @@ -116,6 +120,10 @@ async def checked(self, check: OrderCheck) -> ChildResult: ... @abstractmethod async def delivery_time(self, check): ... + def merged_commits(self, repo: Repo) -> tuple[str, ...]: + """Operations have no commit whose ancestry could prove a merge.""" + return () + def check_failed(self, goals: Goals, todo: Todo, error: OrderCheckFailedError) -> None: """Only a commit delivery can fail a check on delivered source.""" @@ -139,6 +147,9 @@ def delivered(self) -> str: def work(self) -> str: return f"delivered commit {self.delivered}" + def merged_commits(self, repo: Repo) -> tuple[str, ...]: + return (self.delivered,) if repo.contains(self.delivered, "origin/main") else () + def check_failed(self, goals: Goals, todo: Todo, error: OrderCheckFailedError) -> None: if self.commit != self.delivered: return # A failed integration check is not evidence that the unit's commit failed. @@ -729,6 +740,7 @@ def stalled_between(self, ordered: float, since: float, now: float) -> bool: @dataclass(frozen=True) class Moving(Stalling): """Work whose last commit, push or PR change was observed at ``at``.""" + seconds: ClassVar[int] = 15 * 60 at: float = field(kw_only=True) readings: DistanceReadings = field(default=DistanceReadings(), kw_only=True, compare=False) @@ -1333,7 +1345,7 @@ def place(self) -> str: @dataclass(frozen=True, kw_only=True) -class ProblemOrder(DeclaredFamily, affix="ProblemOrder"): +class ProblemOrder(DeclaredFamily, StorySource, affix="ProblemOrder"): """A problem's order state; an unassigned finding is not a fabricated order.""" pending: ClassVar[bool] = False line: ClassVar[str] = "No order" @@ -1435,6 +1447,9 @@ def phase(self) -> type[OrderPhase]: def withdrawals(self, todo: Todo) -> tuple[str, ...]: return (f"{todo.id} folded into {self.into}: {self.why}",) + def story(self) -> tuple[StoryLine, ...]: + return self.original.story() + @dataclass(frozen=True, kw_only=True) class CheckedProblemOrder(ProblemOrder, declared_name=None): @@ -1443,6 +1458,9 @@ class CheckedProblemOrder(ProblemOrder, declared_name=None): def checked(self) -> CheckedProblemOrder: return self + def story(self) -> tuple[StoryLine, ...]: + return (StoryLine(self.at, f"Ordered {self.id}: {'; '.join(self.edits)}"),) + id: str owner: FileLine check: OrderCheck @@ -1514,6 +1532,10 @@ def copy_as[Kind: ProblemOrder](self, kind: type[Kind], **evidence) -> Kind: def merged(self, sha: str) -> MergedProblemOrder: return self.copy_as(MergedProblemOrder, sha=sha) + def observed_merges(self, repo: Repo, source: ProgressSource) -> tuple[ProblemOrder, ...]: + return tuple(self.merged(merged.sha) for merged in MergedProgress.found(source) + if repo.contains(merged.sha)) + @property def pull_request_reference(self) -> str: return f"{self.pull_request_label} {self.id}" @@ -1550,7 +1572,7 @@ def work_instructions(self) -> tuple[str, ...]: def instructions(self, todo): return "\n".join((f"Problem: {todo.id}: {todo.text}", *self.finding.instructions(), f"Order: {self.id}", self.pull_request_instruction(todo.id), - *self.work_instructions(), f"Deadline: {FieldCodec.encode(todo.deadline)}")) + *self.work_instructions())) def delivered(self, delivery): delivery.at.require_delivery() @@ -1640,6 +1662,9 @@ class DeliveredProblemOrder(CheckedProblemOrder): def line(self) -> str: return f"Delivered: {self.delivery.line}, waiting on install" + def story(self) -> tuple[StoryLine, ...]: + return (*super().story(), *self.delivery.story()) + def require_transfer(self, todo: Todo) -> None: owner = (todo.assignment or todo.released[-1].previous).owner raise ValueError(f"{owner} already has {self.delivery.work}; {owner} merges it.") @@ -1649,6 +1674,11 @@ def delivered(self, delivery): raise ValueError("The order already records a different delivery.") return self + def observed_merges(self, repo: Repo, source: ProgressSource) -> tuple[ProblemOrder, ...]: + commits = self.delivery.merged_commits(repo) + return (tuple(self.merged(commit) for commit in commits) + or super().observed_merges(repo, source)) + def closing_artifact(self) -> str: return self.delivery.evidence.artifact() @@ -1711,6 +1741,9 @@ class WithdrawnProblemOrder(CheckedProblemOrder): pending: ClassVar[bool] = False line: ClassVar[str] = "Withdrawn" + def require_close(self) -> None: + """Retiring an erroneous order does not claim installed work.""" + def withdrawals(self, todo: Todo) -> tuple[str, ...]: superseded = f" (superseded by {self.superseded_by})" if self.superseded_by else "" return (f"{todo.id}: order {self.id} withdrawn from {self.previous.owner}, " @@ -2003,7 +2036,7 @@ def actor(self) -> ThreadIncarnation: @dataclass(frozen=True) -class StrikeTodo(TodoOperation): +class StrikeTodo(TodoOperation, StoryEvent): actor: ThreadIncarnation previous: Assignment reason: str @@ -2019,6 +2052,9 @@ def apply(self, current: Todo) -> Todo: return current return replace(current, strikes=(*current.strikes, self), revision=current.revision + 1) + def story_text(self) -> str: + return f"{self.actor.name} struck {self.previous.owner}: {self.reason}" + def prior_strikes(self, current: Todo) -> tuple[StrikeTodo, ...]: return current.live_strikes(self.previous.owner) @@ -2143,13 +2179,19 @@ def addressed(self, target: str, body: str) -> str: @dataclass(frozen=True) -class OrderVerdict(TodoCommand, declared_name=None): +class OrderVerdict(TodoCommand, StoryEvent, declared_name=None): """The commander's verdict on one tool call the holder made for its order, kept on the order.""" audience: ClassVar[type[VerdictAudience]] = VerdictAudience actor: ThreadIncarnation previous: Assignment at: EventTime = field(default_factory=lambda: EventTime.capture(time.time()), kw_only=True) + def story_text(self) -> str: + return f"{self.actor.name} judged {self.previous.owner}: {self.line}" + + def implied_strikes(self) -> tuple[StrikeTodo, ...]: + return () + @abstractmethod def judged(self, current: Todo) -> Todo: ... @@ -2168,6 +2210,9 @@ def apply(self, current: Todo) -> Todo: class ToolCallVerdict(OrderVerdict, TodoOperation, declared_name=None): tool_call_id: str + def story_text(self) -> str: + return f"{super().story_text()}; tool call {self.tool_call_id}" + def apply(self, current: Todo) -> Todo: current.require_commander(self.actor) _text(self.tool_call_id, "Judged tool call ID", 200) @@ -2178,6 +2223,9 @@ def apply(self, current: Todo) -> Todo: class QuotedVerdict(OrderVerdict, declared_name=None): quote: str + def story_text(self) -> str: + return f'{super().story_text()}; quoted: "{self.quote}"' + def require_quote(self, text: str) -> None: quoted = " ".join(self.quote.split()) if len(quoted.split()) < 4 or quoted not in " ".join(text.split()): @@ -2260,6 +2308,9 @@ class SupersededVerdict(ToolCallVerdict): reason: str line = "Superseded" + def story_text(self) -> str: + return f"{super().story_text()} by {self.replacement}: {self.reason}" + def prepare(self, goals, current): replacing = [todo for todo in goals.todos.held_for_commander(self.actor) if todo.sent_in(self.replacement)] @@ -2284,9 +2335,15 @@ class SendBackVerdict(ToolCallVerdict): def line(self) -> str: return f"Sent back: {self.reason.text}" + def implied_strikes(self) -> tuple[StrikeTodo, ...]: + return (StrikeTodo(self.expected_revision, self.actor, self.previous, self.reason.text, + self.at),) + + def story_text(self) -> str: + return f"{super().story_text()} (strike recorded)" + def judged(self, current: Todo) -> Todo: - strike = StrikeTodo(self.expected_revision, self.actor, self.previous, self.reason.text, - self.at) + (strike,) = self.implied_strikes() return strike.apply(current) def record(self, goals: Goals, problem: str) -> Todo: @@ -2294,13 +2351,17 @@ def record(self, goals: Goals, problem: str) -> Todo: @dataclass(frozen=True) -class VoidStrikeTodo(TodoOperation): +class VoidStrikeTodo(TodoOperation, StoryEvent): """The commander takes back a strike recorded in error; the strike stays on the record.""" actor: ThreadIncarnation strike: StrikeTodo reason: str at: EventTime = field(default_factory=lambda: EventTime.capture(time.time())) + def story_text(self) -> str: + return (f"{self.actor.name} voided a strike against {self.strike.previous.owner}: " + f"{self.reason}") + def apply(self, current: Todo) -> Todo: current.require_revision(self.expected_revision) current.require_commander(self.actor) @@ -2312,8 +2373,11 @@ def apply(self, current: Todo) -> Todo: @dataclass(frozen=True) -class ReleaseTodo(TodoOperation): +class ReleaseTodo(TodoOperation, StoryEvent): """The commander takes an open problem off its unit; the order and its strikes stay.""" + + def story_text(self) -> str: + return f"{self.actor.name} released {self.previous.owner}: {self.reason.text}" actor: ThreadIncarnation previous: Assignment reason: StrikeReason @@ -2519,6 +2583,17 @@ def ordered(cls, todo_id: str, repo: str, text: str, creator: ThreadIncarnation, def pending(self): return self.order.pending + def story(self) -> ProblemStory: + implied = frozenset(strike for verdict in self.verdicts + for strike in verdict.implied_strikes()) + sources: tuple[StorySource, ...] = ( + self.order, *(strike for strike in self.strikes if strike not in implied), + *self.voided, *self.released, *self.verdicts, + ) + events = tuple(sorted((event for source in sources for event in source.story()), + key=lambda event: event.at.chronology())) + return ProblemStory(self.id, self.order.line, events) + def overdue(self, now: float) -> bool: """An undelivered order remains past due until its result is recorded.""" return self.pending and self.deadline.recorded_at("An order deadline") < now @@ -2789,6 +2864,13 @@ def _one(cls, rows: tuple[Todo, ...], todo_id: str) -> Todo: def _current(cls, db: sqlite3.Connection, todo_id: str) -> Todo: return cls._one(cls._rows(db, "id=?", (todo_id,)), todo_id) + def for_order(self, order_id: str) -> tuple[Todo, ...]: + """A sent order can cover several linked problems; its ID is not a Todo key.""" + rows = self._select("json_extract(\"order\",'$.id')=?", (order_id,)) + if not rows: + raise TodoError(f"Unknown saved order: {order_id!r}") + return rows + def get(self, todo_id: str) -> Todo: _text(todo_id, "Todo ID", 128) return self._one(self._select("id=?", (todo_id,)), todo_id) @@ -2834,12 +2916,11 @@ def reconcile_merges(self, commander: ThreadIncarnation, repos: Sequence[Repo], checked = tuple((todo, order) for todo in self.held_for_commander(commander) if isinstance(order := todo.order, CheckedProblemOrder)) pulls = PullRequests(tuple(todo for todo, _ in checked)) - completed = tuple(DoneTodoState.closed(todo, order.merged(merged.sha)) + completed = tuple(DoneTodoState.closed(todo, merged) for todo, order in checked for repo in repos if repo.github.lower() == todo.repo - for merged in MergedProgress.found( - ProgressSource.of(todo, storage, pulls)) - if repo.contains(merged.sha)) + for merged in order.observed_merges( + repo, ProgressSource.of(todo, storage, pulls))) if not completed: return with self.writing() as db: diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index cf39fb398f..80f1ee08b8 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -79,6 +79,7 @@ CurrentTaskScopeSelection, Constraint, Decision, Subtask, TaskAttachment, TaskChange, TaskScopeSelection, OriginalTaskChange, ) +from .order_story import FleetStory, OrderStory from .order_work import OrderWork from .turn_digests import LastTurns, OrderEditsRead, ThreadTurns, TurnsSince, TurnDigestRead, UnjudgedEdits from .relationships import RelationshipEdit @@ -940,6 +941,31 @@ def apply(self, comms: Comms, caller: ToolCaller) -> ThreadTurns: return comms.transcripts.turns(caller.subject(self.thread), LastTurns(self.last)) +@dataclass(frozen=True, kw_only=True) +class CommsOrderStoryTool(ReadTool, ToolRequest): + label = "Order story" + description = ("Read a saved order's dated facts from its Todo, oldest first. " + "Send-back and its implied strike share one reason. " + "Overwritten order bodies cannot be reconstructed.") + order_id: str = tool_field("Immutable order ID") + + def apply(self, comms: Comms, caller: ToolCaller) -> OrderStory: + (story,) = OrderStory.collect(comms.goals.todos.for_order(self.order_id)) + return story + + +@dataclass(frozen=True, kw_only=True) +class CommsFleetStoryTool(ThreadReadTool): + label = "Fleet story" + description = ("Read one story line for every pending order of this commander, " + "including linked problems.") + + def apply(self, comms: Comms, caller: ToolCaller) -> FleetStory: + actor = comms.registry.require(caller.subject(self.thread)).incarnation + todos = (todo for todo in comms.goals.todos.for_commander(actor) if todo.pending) + return FleetStory(tuple(story.line for story in OrderStory.collect(todos))) + + @dataclass(frozen=True, kw_only=True) class CommsOrderWorkTool(ReadTool, ToolRequest): label = "Order work" diff --git a/tests/test_candidate_maintenance.py b/tests/test_candidate_maintenance.py index 645154eaf8..2bde1e99f0 100644 --- a/tests/test_candidate_maintenance.py +++ b/tests/test_candidate_maintenance.py @@ -53,15 +53,15 @@ def test_ordinary_private_send_deferred_index_catches_up_without_wake_authority( _SCHEDULE, ) indexed = threading.Event() - original = maintenance._drain_candidate + original = maintenance.CandidateMaintenance.run - def observe(root): - original(root) + def observe(worker): + original(worker) # Candidate catch-up alone is not completion: the same worker # still owns route preparation and its final pending-root check. indexed.set() - monkeypatch.setattr(maintenance, "_drain_candidate", observe) + monkeypatch.setattr(maintenance.CandidateMaintenance, "run", observe) message = comms.messaging.send("sender", "beta", "selected original").message assert indexed.wait(timeout=5), "deferred checkpoint did not catch up" page = WakeCandidateIndex(comms.bus.log).page( @@ -111,14 +111,15 @@ def probe(): def test_failed_drain_is_named_and_retires_pending_state(monkeypatch, tmp_path, caplog) -> None: comms, _root_id = _wire(tmp_path) root = comms.bus.log.path.parent - maintenance._pending[root] = (comms.bus.log, 1) + worker = maintenance.CandidateMaintenance(comms.bus.log, 1) + maintenance._pending[root] = worker def fail(*args, **kwargs): raise ValueError("synthetic index failure") monkeypatch.setattr(WakeCandidateIndex, "catch_up_committed_append", fail) with pytest.raises(maintenance.CandidateMaintenanceFailed) as failed: - maintenance._drain_candidate(root) + worker.run() assert failed.value.root == root assert root not in maintenance._pending assert "Candidate maintenance failed for" in caplog.text @@ -152,8 +153,9 @@ def test_one_background_drain_catches_up_a_backlog_of_oversized_records( ).checkpoint_bytes == first_bytes through_bytes = index.log.path.stat().st_size hint = index.notify_committed_append(root_id=root_id, through_bytes=through_bytes) - maintenance._pending[index.log.path.parent] = (index.log, through_bytes) - maintenance._drain_candidate(comms.bus.log.path.parent) + worker = maintenance.CandidateMaintenance(index.log, through_bytes) + maintenance._pending[index.log.path.parent] = worker + worker.run() assert not maintenance._pending assert index.catch_up_committed_append(hint).caught_up page = index.page( @@ -172,7 +174,7 @@ def test_source_replacement_during_replay_does_not_kill_the_worker(monkeypatch, replay = WakeCandidateIndex._replay_prefix replaced, finished = threading.Event(), threading.Condition() failures = [] - drain = maintenance._drain_candidate + drain = maintenance.CandidateMaintenance.run def replace_once(cls, source, *args, **kwargs): batch = replay(source, *args, **kwargs) @@ -185,9 +187,9 @@ def replace_once(cls, source, *args, **kwargs): replaced.set() return batch - def observed(path): + def observed(worker): try: - drain(path) + drain(worker) except Exception as error: failures.append(error) finally: @@ -199,7 +201,7 @@ def retired(): return root not in maintenance._pending monkeypatch.setattr(WakeCandidateIndex, "_replay_prefix", classmethod(replace_once)) - monkeypatch.setattr(maintenance, "_drain_candidate", observed) + monkeypatch.setattr(maintenance.CandidateMaintenance, "run", observed) monkeypatch.setattr(maintenance, "schedule_candidate_catchup", _SCHEDULE) comms.messaging.send("sender", "beta", "first original") assert replaced.wait(5) @@ -215,11 +217,11 @@ def test_concurrent_ordinary_publication_keeps_maintenance_alive(monkeypatch, tm root = comms.bus.log.path.parent finished = threading.Condition() failures = [] - drain = maintenance._drain_candidate + drain = maintenance.CandidateMaintenance.run - def observed_drain(path): + def observed_drain(worker): try: - drain(path) + drain(worker) except Exception as error: failures.append(error) raise @@ -231,7 +233,7 @@ def retired(): with maintenance._guard: return root not in maintenance._pending - monkeypatch.setattr(maintenance, "_drain_candidate", observed_drain) + monkeypatch.setattr(maintenance.CandidateMaintenance, "run", observed_drain) monkeypatch.setattr(maintenance, "schedule_candidate_catchup", _SCHEDULE) messages = [comms.messaging.send("sender", "beta", str(n) + "x" * (128 * 1024)).message for n in range(128)] diff --git a/tests/test_candidate_publication_notifications.py b/tests/test_candidate_publication_notifications.py index 7c761dda2d..39d1a8a15a 100644 --- a/tests/test_candidate_publication_notifications.py +++ b/tests/test_candidate_publication_notifications.py @@ -65,11 +65,11 @@ def test_nested_release_scope_runs_only_after_outer_release(): def notifications(monkeypatch): def activate(comms): calls, finished = [], threading.Event() - drain = candidate_maintenance._drain_candidate + drain = candidate_maintenance.CandidateMaintenance.run - def complete(root): + def complete(worker): try: - drain(root) + drain(worker) finally: finished.set() @@ -94,7 +94,7 @@ def probe(): calls.append(through_bytes) _SCHEDULE(log, through_bytes) - monkeypatch.setattr(candidate_maintenance, "_drain_candidate", complete) + monkeypatch.setattr(candidate_maintenance.CandidateMaintenance, "run", complete) monkeypatch.setattr(candidate_maintenance, "schedule_candidate_catchup", notify) return calls, finished @@ -250,7 +250,8 @@ def test_retiring_worker_does_not_delete_successor_notification(tmp_path, monkey publish_initial(comms, "sender", "member000", "first") assert index.maintain(start=RebuildIndex()) root = index.log.path.parent - candidate_maintenance._pending[root] = (index.log, index.log.path.stat().st_size) + worker = candidate_maintenance.CandidateMaintenance(index.log, index.log.path.stat().st_size) + candidate_maintenance._pending[root] = worker original_guard = candidate_maintenance._guard successor = [] @@ -269,11 +270,11 @@ def __exit__(self, *_): monkeypatch.setattr(candidate_maintenance, "_guard", InterleavedLock()) monkeypatch.setattr(candidate_maintenance, "schedule_candidate_catchup", _SCHEDULE) monkeypatch.setattr(threading.Thread, "start", lambda _: None) - candidate_maintenance._drain_candidate(root) + worker.run() assert successor - assert candidate_maintenance._pending[root][1] == index.log.path.stat().st_size + assert candidate_maintenance._pending[root].through_bytes == index.log.path.stat().st_size # Complete the retained successor without launching another daemon. - candidate_maintenance._drain_candidate(root) + candidate_maintenance._pending[root].run() assert root not in candidate_maintenance._pending assert index._verified_checkpoint(root_id).last_seq == 2 finally: diff --git a/tests/test_commander_problem_orders.py b/tests/test_commander_problem_orders.py index 7ff1bd1645..7950ec26f1 100644 --- a/tests/test_commander_problem_orders.py +++ b/tests/test_commander_problem_orders.py @@ -534,6 +534,7 @@ def order_body(repo, check=None, edit='Set count.txt to 1.', extra='', deletes=N body.write_text(json.dumps({ 'defect': 'The counter file holds zero where the fleet needs one', 'why_stupid': 'The counter is wrong and that is stupid.', + 'unplanned_reason': 'This isolated order checks command behavior rather than advancing the plan dependency chain.', 'order': {'kind': 'pending', 'owner': f'{__file__}:1', 'check': FieldCodec.encode(checked), 'edits': [edit], 'deletes': FieldCodec.encode(deletes), diff --git a/tests/test_order_story.py b/tests/test_order_story.py new file mode 100644 index 0000000000..b4dda0dc42 --- /dev/null +++ b/tests/test_order_story.py @@ -0,0 +1,83 @@ +"""Native stories use one copied Todo record, including a resent send-back.""" + +from dataclasses import replace + +import pytest + +from agent_comms.comms import Comms +from agent_comms.field_codec import FieldCodec +from agent_comms.goal_actions import CreateProblemGoalAction, OrderGoalAction, SetGoalAction +from agent_comms.goals import RequestGoalOrder +from agent_comms.order_story import FleetStory, OrderStory +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.todos import ( + AcceptedVerdict, SendBackVerdict, StrikeReason, StrikeTodo, TodoError, UnorderedProblemOrder, +) +from agent_comms.tool_caller import NoThreadCaller, ThreadCaller +from agent_comms.tools import ( + CommsFleetStoryTool, CommsOrderStoryTool, CommsTodoCreateTool, CommsOrderTool, CommsVerdictTool, +) +from cutover.backup import StoreBackup +from test_todos import order + + +def test_order_story_copied_bus_strike_and_send_back_once(comms, tmp_path): + comms.messaging.initialize_private_initial_protocol() + for name in ("lead", "worker"): + comms.registry.declare(comms.registry.new_thread(name, frozenset(), str(tmp_path))) + comms.goals.update_goal("lead", SetGoalAction( + text="Repair the worker", goal_id="request", order=RequestGoalOrder())) + caller = ThreadCaller("lead") + caller.invoke(CommsTodoCreateTool(owner="worker", order=CreateProblemGoalAction( + problem="P1", order=replace(order(tmp_path), edits=("Repair the existing record owner",)), + request="request", deadline=RecordedEventTime(1000), + commander="lead", repo="org/repo", text="Repair the worker")), comms) + todo = comms.goals.todos.get("P1") + actor = comms.registry.require("lead").incarnation + reason = "The worker changed the wrong owner and left the decision duplicated." + standalone = "The worker omitted the requested application check during the final review." + comms.goals.todos.apply("P1", StrikeTodo(todo.revision, actor, todo.held(), standalone)) + todo = comms.goals.todos.get("P1") + caller.invoke(CommsVerdictTool(problem="P1", operation=AcceptedVerdict( + todo.revision, actor, todo.held(), "accepted-call", quote="Repair the existing record owner")), comms) + todo = comms.goals.todos.get("P1") + caller.invoke(CommsVerdictTool(problem="P1", operation=SendBackVerdict( + todo.revision, actor, todo.held(), "bad-call", StrikeReason(reason))), comms) + todo = comms.goals.todos.get("P1") + caller.invoke(CommsOrderTool(owner="worker", order=OrderGoalAction( + problem="P1", expected_revision=todo.revision, assigner=actor, commander="lead", + request="request", deadline=RecordedEventTime(2000), + order=replace(order(tmp_path), id="resent-order"))), comms) + copied_root = tmp_path / "copy" + StoreBackup.copy_tree(comms.root, copied_root) + copied = Comms(copied_root) + saved = copied.goals.todos.get("P1") + stored_at = copied.goals.todos.path.stat().st_mtime_ns + result = CommsOrderStoryTool.invoke(copied, {"order_id": "resent-order"}, caller=NoThreadCaller()) + story = FieldCodec.decode(OrderStory, result) + assert story.line.count(reason) == story.line.count(standalone) == 1 + assert story.line.count("Ordered resent-order:") == story.line.count("Accepted") == 1 + assert story.line.count("bad-call") == story.line.count("accepted-call") == 1 + assert "strike recorded" in story.line + assert len(saved.strikes) == len(saved.verdicts) == 2 + assert CommsOrderStoryTool.invoke(copied, {"order_id": "resent-order"}, caller=caller) == result + fleet = FieldCodec.decode(FleetStory, CommsFleetStoryTool.invoke( + copied, {"thread": "lead"}, caller=NoThreadCaller())) + assert fleet.lines == (story.line,) + assert copied.goals.todos.get("P1") == saved == comms.goals.todos.get("P1") + assert copied.goals.todos.path.stat().st_mtime_ns == stored_at + with copied.goals.todos.writing() as db: + for problem in ("P2", "P3"): + replace(saved, id=problem, assignment=None, order=UnorderedProblemOrder(), + strikes=(), verdicts=(), messages=()).insert(db) + copied.goals.todos.link(actor, "P2", "worker", saved.order.checked().id) + linked = FieldCodec.decode(OrderStory, CommsOrderStoryTool.invoke( + copied, {"order_id": "resent-order"}, caller=caller)) + assert tuple(problem.problem for problem in linked.problems) == ("P1", "P2") + assert linked.problems[0] == story.problems[0] + assert linked.line.count(reason) == linked.line.count(standalone) == 1 + linked_fleet = FieldCodec.decode(FleetStory, CommsFleetStoryTool.invoke( + copied, {"thread": "lead"}, caller=caller)) + assert linked_fleet.lines == (linked.line,) + with pytest.raises(TodoError, match="Unknown saved order"): + CommsOrderStoryTool.invoke(copied, {"order_id": "absent-order"}, caller=caller) diff --git a/tests/test_stale_order_gates.py b/tests/test_stale_order_gates.py new file mode 100644 index 0000000000..0e053ea2a1 --- /dev/null +++ b/tests/test_stale_order_gates.py @@ -0,0 +1,158 @@ +"""Real TodoStore rows exercise the gate family and merged-delivery observer.""" +import json +import os +import subprocess +import time +from dataclasses import replace +from pathlib import Path + +import pytest + +from agent_comms.commander.order import OrderFile, OrderVerb, StaleOrderGate +from agent_comms.plan_phases import PlanPhase, RecordPlanePhase, UnplannedPhase +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.repo import Repo +from agent_comms.todos import ( + AssignTodo, CheckedProblemOrder, CommitDelivery, OperationDelivery, WithdrawTodo, +) +from agent_comms.tools import CommsOrderDoneTool +from test_commander_problem_orders import create, order_body, working + + +def test_stale_gate_covers_its_verb_family_but_allows_board_repairs( + commander, commander_script, tmp_path): + comms, _ = commander + context, _ = commander_script + todo = create(commander, 'P244', deadline=time.time() - 1900) + body = str(tmp_path / 'must-not-be-read') + blocked = ( + ['send', 'worker', '23:59', body], ['say', 'worker', body], + ['problem', body, '--to', 'worker', '--due', '23:59', '--where', f'{__file__}:1'], + ['strike', 'worker', '--why', 'P244 missed its deadline'], + ) + verbs = tuple(OrderVerb.parse(args, '') for args in blocked) + assert {type(verb) for verb in verbs} == set(OrderVerb.members_with(StaleOrderGate)) + for verb in verbs: + with pytest.raises(SystemExit) as refused: + verb.apply(context) + text = str(refused.value) + assert 'P244: holder worker, deadline ' in text + assert 'more than 30 minutes overdue' in text + why = 'The current source needs one more bounded check before delivery.' + repairs = ( + ['due', 'P244', '23:59', '--why', why], ['withdraw', 'P244', '--why', why], + ['done', 'worker', 'order-P244', '--sha', 'a' * 40], + ['close', 'P244', '--works', why], + ['check', 'P244', '--check', f'{tmp_path} :: printf 0 -> 0', '--why', why], + ['serve', 'P244', 'record_plane'], ['drop', 'worker', 'old', '--by', 'new'], + ) + for args in repairs: + OrderVerb.parse(args, '').gate(context) + actor = comms.registry.require('lead').incarnation + comms.goals.todos.apply(todo.id, AssignTodo( + todo.revision, todo.held(), RecordedEventTime(time.time() + 3600), todo.order, todo.goal)) + verbs[0].gate(context) + moved = comms.goals.todos.get(todo.id) + comms.goals.todos.apply(todo.id, WithdrawTodo( + moved.revision, actor, moved.held(), why, None)) + verbs[0].gate(context) + assert not comms.goals.todos.get(todo.id).state.open + + +@pytest.mark.parametrize('phase', (UnplannedPhase, *PlanPhase.members_with(PlanPhase))) +def test_phase_gate_uses_the_phase_family_and_requires_an_explicit_unplanned_reason( + commander, commander_script, phase): + comms, _ = commander + context, repo = commander_script + (repo / 'count.txt').write_text('0\n') + payload = json.loads(Path(order_body(repo)).read_text()) + payload['order']['serves'] = phase.declared_name + if phase is UnplannedPhase: + del payload['unplanned_reason'] + with pytest.raises(SystemExit, match='Plan phases are ready'): + context.send('worker', '23:59', OrderFile.read(json.dumps(payload)), + clean=True, new=('P1', 'Repair the owner')) + assert comms.goals.todos.for_commander(comms.registry.require('lead').incarnation) == () + payload['unplanned_reason'] = ( + 'This urgent retained incident is outside the declared plan dependency chain.') + context.send('worker', '23:59', OrderFile.read(json.dumps(payload)), + clean=True, new=('P1', 'Repair the owner')) + assert comms.goals.todos.get('P1').order.serves is phase + + +def test_second_plan_order_gate_names_the_busy_order_and_running_unowned_units( + commander, commander_script): + comms, _ = commander + context, repo = commander_script + todo = create(commander, 'P1') + with comms.goals.todos.writing() as db: + replace(todo, order=replace(todo.order, serves=RecordPlanePhase)).upsert(db) + comms.registry.declare(comms.registry.new_thread('free-unit', frozenset(), str(repo))) + comms.owners.acquire_thread('free-unit', owner_pid=os.getpid()) + (repo / 'count.txt').write_text('0\n') + payload = json.loads(Path(order_body(repo)).read_text()) + payload['order']['serves'] = RecordPlanePhase.declared_name + with pytest.raises(SystemExit) as refused: + context.send('worker', '23:59', OrderFile.read(json.dumps(payload)), + clean=True, new=('P2', 'Repair another owner')) + assert 'worker already holds plan orders P1' in str(refused.value) + assert 'running units holding no order: free-unit' in str(refused.value) + assert [todo.id for todo in comms.goals.todos.for_commander( + comms.registry.require('lead').incarnation)] == ['P1'] + + +def test_pr_uses_actual_owned_checkout_instead_of_an_invented_problem_directory( + commander, commander_script, monkeypatch, tmp_path): + comms, _ = commander + context, repo = commander_script + todo = create(commander, 'P287') + body = tmp_path / 'pull-request.md' + body.write_text('The retained order owns its immutable pull request identity.') + seen = [] + def publish(order, current, checkout, title, text): + seen.append((current.id, checkout, order.pull_request_body(text))) + return 'https://github.com/OpenHCSDev/agent-comms/pull/1' + monkeypatch.setattr(CheckedProblemOrder, 'open_pull_request', publish) + monkeypatch.chdir(repo) + OrderVerb.parse(['pr', 'P287', '--title', 'P287', '--body-file', str(body)], '').apply(context) + assert seen == [('P287', repo, todo.order.pull_request_body(body.read_text()))] + subprocess.run(['git', '-C', str(repo), 'remote', 'set-url', 'origin', + 'https://github.com/another/project.git'], check=True) + with pytest.raises(SystemExit, match=f'not {todo.repo}'): + OrderVerb.parse(['pr', 'P287', '--title', 'P287', '--body-file', str(body)], '').apply(context) + assert len(seen) == 1 # A different repository must never reach publication. + + +@pytest.mark.parametrize('commit_delivery', (True, False)) +def test_observer_closes_only_delivered_commits_present_in_fetched_main( + commander, tmp_path, commit_delivery): + comms, caller = commander + root = comms.root + origin = tmp_path / 'origin.git' + subprocess.run(['git', 'init', '-q', '--bare', str(origin)], check=True) + def git(*args): + return subprocess.check_output(['git', '-C', str(root), '-c', 'user.name=t', + '-c', 'user.email=t@t', *args], text=True).strip() + git('init', '-q', '-b', 'main') + git('remote', 'add', 'origin', str(origin)) + git('commit', '-q', '--allow-empty', '-m', 'delivered source') + sha = git('rev-parse', 'HEAD') + git('push', '-q', 'origin', 'HEAD:main') + todo = create(commander, 'P1', deadline=time.time() + 3600) + with comms.goals.todos.writing() as db: + replace(todo, order=replace(todo.order, serves=RecordPlanePhase)).upsert(db) + delivery = (CommitDelivery(commit=sha, delivered_commit=sha) if commit_delivery + else OperationDelivery()) + CommsOrderDoneTool(problem='P1', order_id=todo.order.id, delivery=delivery, + claims=working(comms)).apply(comms, caller) + assert comms.goals.todos.get('P1').state.open + repo = Repo(checkout=root, github=todo.repo) + actor = comms.registry.require('lead').incarnation + comms.goals.todos.reconcile_merges(actor, (repo,), + comms.registry.storage_rule) + observed = comms.goals.todos.get('P1') + assert observed.state.open is not commit_delivery + assert observed.order.serves is RecordPlanePhase + comms.goals.todos.reconcile_merges(actor, (repo,), + comms.registry.storage_rule) + assert comms.goals.todos.get('P1') == observed diff --git a/tests/test_tool_replies.py b/tests/test_tool_replies.py index d6814b664f..05dbf355df 100644 --- a/tests/test_tool_replies.py +++ b/tests/test_tool_replies.py @@ -19,7 +19,8 @@ from agent_comms.tools import ( CommsArchiveTool, CommsCancelRestartTool, CommsChannelsTool, CommsCollaborationTool, CommsCollaborationsTool, CommsConstraintTool, CommsDecisionTool, CommsDeleteViewTool, - CommsEditGoalTool, CommsGoalHistoryTool, CommsGoalTool, CommsInboxTool, + CommsEditGoalTool, CommsFleetStoryTool, CommsGoalHistoryTool, CommsGoalTool, CommsInboxTool, + CommsOrderStoryTool, CommsCorrectWordingTool, CommsFireTool, CommsModelTool, CommsOrderCheckTool, CommsOrderDoneTool, CommsOrderServeTool, CommsOrderTool, CommsOrderWorkTool, CommsPinChannelTool, CommsPinThreadTool, CommsProblemCloseTool, CommsRenameSelfTool, CommsRestartQueueTool, CommsRuleReviewTool, CommsResumeGoalTool, CommsSendTool, CommsSetChannelMetadataTool, CommsSetGoalTool, @@ -103,6 +104,8 @@ def scenario(comms: Comms, project, monkeypatch): OrderCheck(str(comms.root), "printf 1", CountCheckResult(1))), "why": "The recorded check counted the wrong file"}, lead yield CommsOrderWorkTool, {"problem": "P1"}, lead + yield CommsOrderStoryTool, {"order_id": comms.goals.todos.get("P1").order.checked().id}, lead + yield CommsFleetStoryTool, {"thread": "lead"}, lead (comms.root / "installed-evidence").write_text("Verified installed path") yield CommsOrderDoneTool, {"problem": "P1", "order_id": comms.goals.todos.get("P1").order.id, "delivery": FieldCodec.encode(OperationDelivery()),