diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 05042371b2..4241ab5d51 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -57,7 +57,7 @@ | 24 | Reassigning past two owners needs the commander's trace | scratch `order.py send --reassign --traced` | Transfer reads Todo owner history and requires traced source location | `record_plane` | | 25 | Unit wording gets a correction, not a strike | scratch `vibe_watch.py` | `PlainText` for commander verdicts/strikes; unit wording gets correction | `record_plane` | | 26 | Alerts carry only decisions | none; every line interrupts | `messages.AlertKind.require_publishable`, `tools.CommsDecisionTool` | `control_plane` | -| 27 | Momentum prediction per unit, compared with the outcome | none | Prediction record and later outcome on commander page; `CommitMomentum` is a partial candidate | `commander_page` | +| 27 | Momentum prediction per unit, compared with the outcome | none | Todo retains the latest heartbeat distance forecast and actual closure; OrderProgress shows both, with numeric error only for a measured falling count | `commander_page` | | 28 | Measured load and self-report | `capacity.py` unit count | `commander.capacity.Load.of` reports orders, overdue, edits and forks; self-report comparison unverified | `control_plane` | | 29 | No handoff of finished, checked work | none; scratch transfer moved P507 after its fix was pushed and checked | Transfer refuses delivered checked work; that owner merges; check is reused | `record_plane` | | 30 | A proof is bound to the build it ran on | none; screenshots and logs are bare paths | Build run records runtime and Core/Toad revisions; close requires delivered commit | `record_plane` | 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 5e5cdef9dc..90c37c8336 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -465,8 +465,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') @@ -477,12 +476,16 @@ def write_board(self): '| Problem | Owner file | Owner | Order | Due | Check | Progress | Order state ' '| Strikes | Owner last acted |', '|---|---|---|---|---|---|---|---|---|---|'] - listed, read = self.todos(), self.progress() + self.todos() # Reconcile installed merges before reading the board snapshot. + snapshot: TodoList = self.call(CommsTodoListTool()) + listed, read = snapshot.todos, snapshot.progress activity = {unit: self.last_activity(unit) for unit in {t.holder for t in listed if t.holder}} for t in sorted(listed, key=lambda t: int(t.id[1:]) if t.id[1:].isdigit() else 0): lines.append(f'| {t.id}: {t.text.replace("|", "|")} | {t.order.board_cells(t)} | ' f'{read[t.id].line if t.id in read else "-"} | {t.order.line} | ' f'{t.strike_count} | {activity[t.holder].clock() if t.holder else "-"} |') + lines += ['', '## Closed orders: predicted against actual', '', + *(f'- {todo.id}: {read[todo.id].line}' for todo in snapshot.closed)] history = self.comms.goals.todos.history_for_commander(self.incarnation) lines += ['', '## Withdrawn orders', '', *(f'- {line}' for t in history for line in t.order.withdrawals(t))] @@ -715,8 +718,8 @@ def due(self, pid: str, at: str, why: str) -> None: moved = RecordedEventTime(time.time()).at_clock(at) self.call(CommsChangeProblemTool(problem=pid, operation=AssignTodo( todo.revision, todo.assignment, moved, todo.order, todo.goal))) - self.on_record(f'{time.strftime("%Y-%m-%d %H:%M")} {todo.holder} ({pid} order {todo.order.id} due ' - f'{todo.deadline.clock()} moved to {moved.clock()}): {why}', 'Deadline', + self.on_record(f'{time.strftime("%Y-%m-%d %H:%M")} {todo.holder} ({pid} order {todo.order.id} ' + f'deadline changed): {why}', 'Deadline', VerdictAudience.of(todo.held().owner, todo.goal)) print(f'{pid}: due {moved.clock()}') diff --git a/src/agent_comms/commander_heartbeat.py b/src/agent_comms/commander_heartbeat.py index cc8a466762..4b0e830533 100644 --- a/src/agent_comms/commander_heartbeat.py +++ b/src/agent_comms/commander_heartbeat.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time from collections.abc import Callable from contextlib import AbstractContextManager from dataclasses import dataclass, field @@ -21,10 +22,16 @@ class CommanderReview: def wake(self, comms: Comms, sender: str, commander: Thread) -> None: rules = StandingRule.broken(comms, commander.name) + reviewed = comms.goals.todos.review_predictions( + commander.incarnation, comms.registry.storage_rule, time.time() + ) found = frozenset(rules) if found == self.violations: return - lines = ["COMMANDER HEARTBEAT: standing-rule violations changed."] + lines = [ + "COMMANDER HEARTBEAT: standing-rule violations changed.", + *(f"{todo.id} @{todo.holder}{todo.predicted_finish.line}" for todo in reviewed), + ] owed = comms.goals.todos.held_for_commander(commander.incarnation) units = sorted({todo.held().owner for todo in owed}) if units: 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..72f9b8053d --- /dev/null +++ b/src/agent_comms/commander_parity/lifecycle.py @@ -0,0 +1,486 @@ +"""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 +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, + 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( + clock.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 after.deadline.clock() not in instructions, + ), + ("commander still sees deadline", f"due {after.deadline.clock()}" 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/prediction.py b/src/agent_comms/commander_parity/prediction.py new file mode 100644 index 0000000000..3a7d991fd9 --- /dev/null +++ b/src/agent_comms/commander_parity/prediction.py @@ -0,0 +1,153 @@ +"""A retained forecast survives two real heartbeat reviews and native closing.""" + +from __future__ import annotations + +import asyncio +import json +import os +import subprocess +from dataclasses import replace +from pathlib import Path + +from ..commander_heartbeat import CommanderReview +from ..field_codec import FieldCodec +from ..goals import CheckedCommit, OrderCheck +from ..storage_rules import StorageRule +from ..todos import FinishedPrediction, MovingTrend, ReviewedPrediction, Todo +from ..tool_result import ToolResult +from ..tools import CommsProblemCloseTool, CommsTodoListTool, TodoList +from .lifecycle import LifecycleCase, LifecycleRun + + +class PredictionCase(LifecycleCase): + native_owner = CommsTodoListTool + additional_owners = (CommsProblemCloseTool,) + + @staticmethod + def git(checkout: Path, *arguments: str, **environment: str) -> str: + return subprocess.check_output( + ("git", "-C", str(checkout), *arguments), + env=os.environ | environment, + text=True, + stderr=subprocess.PIPE, + timeout=30, + ).strip() + + @classmethod + def record_check(cls, check: OrderCheck, checkout: Path) -> None: + result = asyncio.run(check.run()) + CheckedCommit(check.command, result.outcome, result.stdout.decode()).store( + str(checkout), cls.git(checkout, "rev-parse", "HEAD") + ) + + @classmethod + def exercise(cls, run: LifecycleRun) -> tuple[Todo, ...]: + created = run.create() + comms = run.commander.comms + root = comms.root + # The copy owns its policy and local Git history, never the source worker's checkout. + policy = replace(comms.registry.storage_rule, checkout_base=root / "projects") + (root / "storage.json").write_text(json.dumps(FieldCodec.encode(policy))) + # Registration captures configuration for its lifetime; load the copy's new policy. + comms.registry.storage_rule = StorageRule.load(root) + (root / "commander.json").write_text(run.commander.model_dump_json()) + checkout = policy.order_checkout(run.worker, created.id) + checkout.mkdir(parents=True) + cls.git(checkout, "init", "-q", "-b", "main") + cls.git(checkout, "config", "user.name", "Prediction comparison") + cls.git(checkout, "config", "user.email", "prediction@example.invalid") + origin = root / "origin.git" + subprocess.run(("git", "init", "-q", "--bare", str(origin)), check=True, timeout=30) + cls.git(checkout, "remote", "add", "origin", str(origin)) + source = checkout / "decisions.txt" + source.write_text("100") + cls.git(checkout, "add", source.name) + ordered = created.order.checked().at.recorded_at("The copied order") + base_date = f"@{int(ordered - 1)} +0000" + cls.git( + checkout, + "commit", + "-qm", + "Retained base", + GIT_AUTHOR_DATE=base_date, + GIT_COMMITTER_DATE=base_date, + ) + cls.git(checkout, "push", "-q", "origin", "main") + cls.git(checkout, "symbolic-ref", "refs/remotes/origin/HEAD", "refs/remotes/origin/main") + check = created.order.checked().check + run.decisions.write_text(json.dumps(["unfinished"] * 100)) + cls.record_check(check, checkout) + cls.git(checkout, "switch", "-qc", "work") + heartbeat = CommanderReview() + commander = comms.registry.require(run.commander.commander) + rows = [created] + for offset, remaining in ((2, 60), (4, 20)): + source.write_text(str(remaining)) + cls.git(checkout, "add", source.name) + date = f"@{int(ordered + offset)} +0000" + cls.git( + checkout, + "commit", + "-qm", + f"{remaining} decisions remain", + GIT_AUTHOR_DATE=date, + GIT_COMMITTER_DATE=date, + ) + run.decisions.write_text(json.dumps(["unfinished"] * remaining)) + cls.record_check(check, checkout) + heartbeat.wake(comms, run.worker, commander) + rows.append(comms.goals.todos.get(created.id)) + # The current command and native owner must expose the exact retained forecast. + run.commander.write_board() + measurement = run.commander.invocations[-1].result + if not isinstance(measurement, TodoList): + raise ValueError("Board did not read its native TodoList") + print(f"Review: {measurement.progress[created.id].line}") + # The fixture repository is retained in the disposable copy, no longer an open checkout. + checkout.rename(root / "completed-history") + run.complete(rows[-1]) + run.commander.close( + created.id, + "Read the copied native goal and confirm its recorded source location " + "and assigned worker.", + str(run.artifact), + ) + rows.append(comms.goals.todos.get(created.id)) + return tuple(rows) + + @classmethod + def checks( + cls, run: LifecycleRun, rows: tuple[Todo, ...], native: tuple[ToolResult, ...], output: str + ) -> tuple[tuple[str, bool], ...]: + reads = tuple(result for result in native if isinstance(result, TodoList)) + if len(rows) != 4 or len(reads) < 3: + return (("two native reviews and closing observed", False),) + pid = rows[0].id + # Each board invocation carries the exact native result; compare no second live read. + finishes = tuple(read.progress[pid].finish for read in reads if pid in read.progress) + reviewed = tuple(finish for finish in finishes if isinstance(finish, ReviewedPrediction)) + finished = tuple(finish for finish in finishes if isinstance(finish, FinishedPrediction)) + if len(reviewed) < 2 or not finished: + return (("reviewed and actual finish retained", False),) + first, last, closed = reviewed[0], reviewed[-1], finished[-1] + board = run.commander.board.read_text() + messages = run.commander.comms.bus.inbox(run.commander.commander) + return ( + ( + "both reviews derive numeric measured slopes", + isinstance(first.trend, MovingTrend) and isinstance(last.trend, MovingTrend), + ), + ("second review replaces first", first != last and rows[1].revision < rows[2].revision), + ("closure retains last prediction", closed.forecast == last), + ("actual finish recorded", closed.actual >= last.at), + ( + "native row actually closed", + not rows[-1].state.open and pid in {t.id for t in reads[-1].closed}, + ), + ("board shows predicted against actual", closed.line in board), + ("comparison shows signed error", "actual minus predicted" in closed.line), + ( + "heartbeat shows its saved forecast", + any("predicted finish" in message.body for message in messages), + ), + ) diff --git a/src/agent_comms/commander_parity/rows.py b/src/agent_comms/commander_parity/rows.py index ab22a2feb2..84d135c7c1 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 .prediction import PredictionCase 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 @@ -242,10 +259,11 @@ class MomentumParityRow(ParityRow): capability = "Momentum prediction per unit, compared with the outcome" current = "none" native = ( - "Prediction record and later outcome on commander page; " - "`CommitMomentum` is a partial candidate" + "Todo retains the latest heartbeat distance forecast and actual closure; " + "OrderProgress shows both, with numeric error only for a measured falling count" ) phases = (CommanderPagePhase,) + comparison = PredictionCase.read class MeasuredLoadParityRow(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/goals.py b/src/agent_comms/goals.py index 5cd5f236cf..be2daa6035 100644 --- a/src/agent_comms/goals.py +++ b/src/agent_comms/goals.py @@ -331,12 +331,13 @@ def recorded(self, sha: str, repository: str) -> tuple[ChildResult, ...]: return tuple(checked.result for checked in CheckedCommit.stored(repository, sha) if checked.command == self.command)[:1] - def at(self, revision: str, repository: str) -> ChildResult: + def at(self, revision: str, repository: str, checkout_base: Path) -> ChildResult: """Read the result on ``revision``, or explicitly run the check in a detached checkout.""" sha = CheckedCommit.git(repository, "rev-parse", f"{revision}^{{commit}}").strip() for recorded in self.recorded(sha, repository): return recorded - with tempfile.TemporaryDirectory(prefix="order-check-") as scratch: + checkout_base.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix="order-check-", dir=checkout_base) as scratch: tree = str(Path(scratch) / "tree") CheckedCommit.git(repository, "worktree", "add", "--detach", tree, sha) try: diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 22bc688b77..f301ec98be 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -98,14 +98,14 @@ async def verified(self, todo: Todo, registry: Registration) -> OrderDelivery: check = todo.order.check for each in (check, *todo.order.deletes.checks): await self.require_source(each) - each.result.require(await self.checked(each)) + each.result.require(await self.checked(each, registry.storage_rule.checkout_base)) return replace(self, at=await self.delivery_time(check)) @abstractmethod async def require_source(self, check): ... @abstractmethod - async def checked(self, check: OrderCheck) -> ChildResult: ... + async def checked(self, check: OrderCheck, checkout_base: Path) -> ChildResult: ... @abstractmethod async def delivery_time(self, check): ... @@ -140,8 +140,8 @@ def work(self) -> str: async def require_source(self, check: OrderCheck) -> None: raise ValueError("A commit delivery names the unit's delivered commit and the checked one.") - async def checked(self, check: OrderCheck) -> ChildResult: - return await asyncio.to_thread(check.at, self.commit, check.directory) + async def checked(self, check: OrderCheck, checkout_base: Path) -> ChildResult: + return await asyncio.to_thread(check.at, self.commit, check.directory, checkout_base) async def delivery_time(self, check: OrderCheck) -> EventTime: return EventTime.capture(float( @@ -281,7 +281,7 @@ def line(self) -> str: async def require_source(self, check): pass - async def checked(self, check: OrderCheck) -> ChildResult: + async def checked(self, check: OrderCheck, checkout_base: Path) -> ChildResult: return await check.run() async def delivery_time(self, check): @@ -471,6 +471,13 @@ class DistanceTrend(DeclaredFamily, affix="Trend"): @abstractmethod def projection(self) -> str: ... + @property + def prediction(self) -> str: + return f"unavailable{self.projection}" + + def error(self, actual: float) -> str: + return "" + @dataclass(frozen=True) class MovingTrend(DistanceTrend): @@ -481,6 +488,13 @@ class MovingTrend(DistanceTrend): def projection(self) -> str: return f"; distance {self.distance}, projected done {RecordedEventTime(self.done).clock()}" + @property + def prediction(self) -> str: + return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.done)) + + def error(self, actual: float) -> str: + return f"; actual minus predicted {actual - self.done:+.0f}s" + @dataclass(frozen=True) class FlatTrend(DistanceTrend): @@ -513,6 +527,64 @@ def projection(self) -> str: return "" +@dataclass(frozen=True) +class FinishPrediction(DeclaredFamily, affix="FinishPrediction"): + """The order's unreviewed, reviewed, or closed forecast, retained on its Todo.""" + + @property + @abstractmethod + def line(self) -> str: ... + + def error(self, actual: float) -> str: + return "" + + def closed(self, actual: float) -> FinishPrediction: + return FinishedPrediction(self, actual) + + def for_order(self, order_id: str) -> FinishPrediction: + return self + + +@dataclass(frozen=True) +class UnreviewedPrediction(FinishPrediction): + @property + def line(self) -> str: + return "" + + +@dataclass(frozen=True) +class ReviewedPrediction(FinishPrediction): + order_id: str + at: float + trend: DistanceTrend + + def for_order(self, order_id: str) -> FinishPrediction: + return self if order_id == self.order_id else UnreviewedPrediction() + + @property + def line(self) -> str: + return (f"; predicted finish {self.trend.prediction}" + f" (reviewed {RecordedEventTime(self.at).clock()})") + + def error(self, actual: float) -> str: + return self.trend.error(actual) + + +@dataclass(frozen=True) +class FinishedPrediction(FinishPrediction): + forecast: FinishPrediction + actual: float + + @property + def line(self) -> str: + forecast = self.forecast.line or "; predicted finish not reviewed" + actual = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.actual)) + return f"{forecast}; actual finish {actual}{self.forecast.error(self.actual)}" + + def closed(self, actual: float) -> FinishPrediction: + return self + + @dataclass(frozen=True) class ProgressCheckout(DeclaredFamily, affix="Checkout"): """The local history available when an order's progress is captured.""" @@ -635,12 +707,17 @@ def readings(self) -> DistanceReadings: class OrderProgress(DeclaredFamily, affix="Progress"): """Where an order's work stands, derived on each read from its unit's checkout and PRs. - Never stored: the checkout and GitHub are its owners.""" + Never stored: the checkout and GitHub are its owners. The finish view is the + prediction already recorded on the Todo, not a new estimate on every read.""" + + finish: FinishPrediction = field( + default_factory=UnreviewedPrediction, kw_only=True, compare=False) @classmethod def read(cls, todo: Todo, storage: StorageRule, pulls: PullRequests) -> OrderProgress: """The first found member by its declared precedence; with none found, no commit yet.""" - return cls.from_source(MeasuredProgressSource.of(todo, storage, pulls)) + return replace(cls.from_source(MeasuredProgressSource.of(todo, storage, pulls)), + finish=todo.predicted_finish) @classmethod def from_source(cls, source: ProgressSource) -> OrderProgress: @@ -657,7 +734,10 @@ def state(self) -> str: ... @property def line(self) -> str: - return self.state + self.distance_readings().trend.projection + return self.state + self.distance_readings().trend.projection + self.finish.line + + def forecast(self, order_id: str, at: float) -> FinishPrediction: + return ReviewedPrediction(order_id, at, self.distance_readings().trend) def distance_readings(self) -> DistanceReadings: """The order's distance readings; progress that read no commit has none.""" @@ -683,6 +763,13 @@ def missing_merges(self, source: ProgressSource) -> tuple[MergedProgress, ...]: return () +@dataclass(frozen=True) +class ClosedProgress(OrderProgress): + @property + def state(self) -> str: + return "closed" + + @dataclass(frozen=True) class FoundProgress(OrderProgress): """A state read finds in the checkout or on GitHub; the lowest ``precedence`` found wins.""" @@ -723,6 +810,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) @@ -1535,7 +1623,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() @@ -1862,6 +1950,10 @@ class TodoState(DeclaredFamily, affix="TodoState"): def open_names(cls) -> tuple[str, ...]: return tuple(member.declared_name for member in cls.members_with(cls) if member.open) + @classmethod + def progress(cls, current: Todo, storage: StorageRule, pulls: PullRequests) -> OrderProgress: + return OrderProgress.read(current, storage, pulls) + @classmethod def require_open(cls, current: Todo) -> None: pass @@ -1901,6 +1993,10 @@ class BlockedTodoState(TodoState): class DoneTodoState(TodoState): open: ClassVar[bool] = False + @classmethod + def progress(cls, current: Todo, storage: StorageRule, pulls: PullRequests) -> OrderProgress: + return ClosedProgress(finish=current.predicted_finish) + @classmethod def require_open(cls, current: Todo) -> None: raise TodoError(f"{current.id} is done; a strike or hunt is recorded on an open problem.") @@ -1922,6 +2018,7 @@ def decide(cls, current, command): def closed(cls, current: Todo, order: ProblemOrder) -> Todo: order.require_close() return replace(current, state=cls, order=order, assignment=None, + predicted_finish=current.predicted_finish.closed(time.time()), revision=current.revision + 1) @@ -1969,8 +2066,9 @@ def apply(self, current: Todo) -> Todo: self.order.require_writable() assigned = (current if current.assignment == proposed else current.state.assign(current, proposed, self.traced)) + forecast = current.predicted_finish.for_order(self.order.id) assigned = replace(assigned, deadline=self.deadline, order=self.order, goal=self.goal, - revision=current.revision + 1) + predicted_finish=forecast, revision=current.revision + 1) return self.actor.assigned(current, assigned) @property @@ -2377,6 +2475,7 @@ class Todo(TypedTable): verdicts: tuple[OrderVerdict, ...] = () corrections: tuple[WordingCorrection, ...] = () messages: tuple[OrderMessage, ...] = () + predicted_finish: FinishPrediction = field(default_factory=UnreviewedPrediction) unrequested_goal: ClassVar[GoalReference] = NoGoalReference() @@ -2642,6 +2741,7 @@ def link(self, commander: ThreadIncarnation, problem: str, unit: str, message: s raise ValueError(f'{problem} is already held by {target.holder}') replace(target, order=target.order.linked(source.order), assignment=source.held(), goal=source.goal, deadline=source.deadline, messages=source.messages, + predicted_finish=UnreviewedPrediction(), revision=target.revision + 1).upsert(db) def _select(self, where: str, parameters: tuple) -> tuple[Todo, ...]: @@ -2678,6 +2778,41 @@ def for_goal(self, owner: ThreadIncarnation, goal_id: str) -> tuple[Todo, ...]: "json_extract(assignment,'$.owner')=? AND json_extract(assignment,'$.owner_created')=? " "AND json_extract(assignment,'$.generation')=?", (owner.name, owner.created_at, goal_id)) + def review_predictions(self, commander: ThreadIncarnation, storage: StorageRule, + at: float) -> tuple[Todo, ...]: + """Read progress before locking; save only the still-current order. + + Counts that are missing or not falling remain explicit unknown forecasts. A + deadline or commit cadence is not evidence of remaining work. + """ + held = self.held_for_commander(commander) + pulls = PullRequests(held) + observed = tuple((todo, OrderProgress.read(todo, storage, pulls).forecast( + todo.order.checked().id, at)) for todo in held) + saved = [] + if observed: + with self.writing() as db: + for todo, prediction in observed: + current = self._current(db, todo.id) + if current == todo: + result = replace(current, predicted_finish=prediction, + revision=current.revision + 1) + result.upsert(db) + saved.append(result) + return tuple(saved) + + def recent_closed_for_commander(self, commander: ThreadIncarnation, + limit: int = 50) -> tuple[Todo, ...]: + """The bounded outcome tail shown beside the open fleet, never all history.""" + condition, parameters = Todo._condition("1", (), { + "creator": commander.name, "creator_created": commander.wire_birth, + "state": DoneTodoState}) + with self.snapshot() as db: + return tuple(Todo.read(db.execute( + f"SELECT * FROM todo WHERE {condition} " + "ORDER BY json_extract(predicted_finish, '$.actual') DESC, rowid DESC LIMIT ?", + (*parameters, limit)))) + def stalled(self, storage: StorageRule) -> tuple[Todo, ...]: """Observe active held orders without changing owners, strikes or releases.""" held = tuple(todo for todo in self._open( diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index cf39fb398f..b3ab9fb044 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -168,6 +168,7 @@ class TodoChanged(ToolResult): class TodoList(ToolResult): todos: tuple[Todo, ...] progress: dict[str, OrderProgress] = field(default_factory=dict) + closed: tuple[Todo, ...] = () @dataclass(frozen=True) @@ -824,9 +825,12 @@ class CommsTodoListTool(ThreadReadTool): def apply(self, comms: Comms, caller: ToolCaller) -> TodoList: actor = comms.registry.require(caller.subject(self.thread)).incarnation listed = tuple(comms.goals.todos.for_commander(actor)) + closed = comms.goals.todos.recent_closed_for_commander(actor) pulls, storage = PullRequests(listed), comms.registry.storage_rule - return TodoList(listed, {todo.id: OrderProgress.read(todo, storage, pulls) - for todo in listed if todo.holder}) + return TodoList(listed, {todo.id: todo.state.progress(todo, storage, pulls) + for todo in (*listed, *closed) + if todo.holder or not todo.state.open}, + closed) @dataclass(frozen=True, kw_only=True) 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_order_prediction.py b/tests/test_order_prediction.py new file mode 100644 index 0000000000..963e09b424 --- /dev/null +++ b/tests/test_order_prediction.py @@ -0,0 +1,37 @@ +"""Forecast acceptance uses the shared copied-bus lifecycle, real Git and native owners.""" + +import json + +from agent_comms.commander.config import CommanderConfig +from agent_comms.commander_parity.prediction import PredictionCase +from agent_comms.comms import Comms +from agent_comms.field_codec import FieldCodec +from agent_comms.storage_rules import StorageRule +from agent_comms.thread_execution import ExternalThreadExecution +from agent_comms.todos import TodoStore + + +def test_order_reviewed_twice_then_closed_retains_last_prediction_on_a_copied_bus( + tmp_path, monkeypatch, +): + source = Comms(tmp_path / "source") + source.messaging.initialize_private_initial_protocol() + policy = StorageRule(scratch_base=tmp_path / "scratch", checkout_base=tmp_path / "wt") + (source.root / "storage.json").write_text(json.dumps(FieldCodec.encode(policy))) + source = Comms(source.root) # Registration captures the declared policy on construction. + for name in ("lead", "worker"): + source.registry.declare(source.registry.new_thread( + name, frozenset({"comms"}), str(tmp_path), execution=ExternalThreadExecution, + )) + CommanderConfig( + commander="lead", root=source.root, + state=tmp_path / "state", board=tmp_path / "board.md", + narrative=tmp_path / "narrative.md", repos=(), worktrees=tmp_path / "wt", + scratch=tmp_path / "scratch", disk=tmp_path, + ).write(tmp_path) + # Boundary configuration, not a replacement of any owner or its return value. + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AGENT_COMMS_THREAD", "worker") + result = PredictionCase.read() + assert result.passed, result.describe() + assert not TodoStore(source.root).path.exists() # Original has no order or forecast writes. diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index da1e20fce6..25a51e2ab8 100644 --- a/tests/test_order_progress.py +++ b/tests/test_order_progress.py @@ -115,7 +115,7 @@ def test_uncounted_check_keeps_other_overdue_orders_visible(commander, commander replace(todo, order=broken).upsert(db) create(commander, 'P2', deadline=1) commit(unit) - broken.check.at(git(unit, 'rev-parse', 'HEAD'), str(unit)) + broken.check.at(git(unit, 'rev-parse', 'HEAD'), str(unit), unit.parent) progress = read((comms.goals.todos.get('P1'), unit)) assert progress.line.endswith("; check prints no integer: printf 'no count here'") @@ -213,7 +213,7 @@ def test_progress_reads_only_explicit_check_receipts(checkout, tmp_path): source = ProgressSource.of(todo, StorageRule(checkout_base=unit.parent), Listed()) assert not source.readings().readings assert not marker.exists() - check.at(git(unit, 'rev-parse', 'HEAD').strip(), str(unit)) + check.at(git(unit, 'rev-parse', 'HEAD').strip(), str(unit), unit.parent) assert marker.read_text() == 'x' assert len(source.readings().readings) == 1 assert source.readings().readings[0].distance == CountPrinted(todo.order.checked().untouched(())) diff --git a/tests/test_standing_rules.py b/tests/test_standing_rules.py index 4053c77e2d..9f64a92ac3 100644 --- a/tests/test_standing_rules.py +++ b/tests/test_standing_rules.py @@ -206,17 +206,17 @@ def commit(left): check = f"echo run >> {runs}; cat left" todo = create(commander, "P1", command=check, expected=0, deadline=time.time() + 3600) ordered_check = todo.order.checked().check - ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit)) + ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit), unit.parent) def progress(): return OrderProgress.read(comms.goals.todos.get("P1"), storage, PullRequests()) commit(2) - ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit)) + ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit), unit.parent) assert "; distance 2, projected done " in progress().line assert isinstance(progress().distance_readings().trend, MovingTrend) commit(2) - ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit)) + ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit), unit.parent) assert progress().line.endswith("; distance 2, not falling") assert runs.read_text().count("run") == 3 # once per commit: base, then each branch commit diff --git a/tools/cutover/carry.py b/tools/cutover/carry.py index 20fdcda02f..de092f6507 100644 --- a/tools/cutover/carry.py +++ b/tools/cutover/carry.py @@ -644,7 +644,7 @@ def retire(self) -> None: class TodoColumnCarry(TodoRows): - """Todo rows gain one declared record column, empty; every other column stays as it was.""" + """Todo rows gain one declared record column at its default; other columns stay as saved.""" column: ClassVar[str] @@ -694,7 +694,8 @@ def _carry(cls, db: sqlite3.Connection, step: Step, chain: CarryChain, *, apply: raise ValueError(f"Unexpected todo columns: {actual!r}") selected = ", ".join(f'? AS "{item.wire_name}"' if item in added else f'"{item.wire_name}"' for item in fields) - empty = tuple(item.encode(()) for item in added) + empty = tuple(item.encode(FieldCodec.field_plan(Todo, item.name).default()) + for item in added) # Rows decode only once every step before the last has carried them; this one # adds columns, so it counts rows without reading them as today's Todo. count = int(db.execute(f'SELECT count(*) FROM "{table}"').fetchone()[0]) diff --git a/tools/cutover/carry_order_prediction.py b/tools/cutover/carry_order_prediction.py new file mode 100644 index 0000000000..0dce4f20a9 --- /dev/null +++ b/tools/cutover/carry_order_prediction.py @@ -0,0 +1,11 @@ +"""Todo rows gain the declared finish forecast, initially unreviewed.""" + +from .carry import TodoColumnCarry + + +class OrderPrediction(TodoColumnCarry): + column = "predicted_finish" + + +if __name__ == "__main__": + OrderPrediction.main() diff --git a/tools/proofs/cn_progress_read_only.py b/tools/proofs/cn_progress_read_only.py index 5065895634..05d2364053 100644 --- a/tools/proofs/cn_progress_read_only.py +++ b/tools/proofs/cn_progress_read_only.py @@ -63,7 +63,7 @@ def verify() -> None: assert not MARKER.exists() sha = CheckedCommit.git(str(checkout), "rev-parse", "HEAD").strip() assert not check.recorded(sha, str(checkout)) - check.at(sha, str(checkout)) # Deliberately explicit, never a board read. + check.at(sha, str(checkout), comms.registry.storage_rule.checkout_base) assert MARKER.read_bytes() == MARKER_CONTENT assert len(board(todo).readings) == 1 assert MARKER.read_bytes() == MARKER_CONTENT