diff --git a/docs/refactor/phase7-dispatch.md b/docs/refactor/phase7-dispatch.md new file mode 100644 index 0000000000..56bedeae88 --- /dev/null +++ b/docs/refactor/phase7-dispatch.md @@ -0,0 +1,103 @@ +# Phase 7: dispatch checked, ready work + +Owner: pi-cn-phase7. Rebased onto main `65f58db93` under the direct conflict instruction. +P270's merged-order and installed-proof integration belongs to pi-connector (#2036). + +## Implemented ownership + +`PlanPhase.task` declares a `PhaseTask`. `UnspecifiedPhaseTask` stays visible +but cannot invent an order or start a unit. `CheckedPhaseTask` carries the +existing `PendingProblemOrder`, including its source, check, edits and brief. +Production full-phase task briefs are still separate work; the dispatch proof +is not their completion. + +`PlanIdle.enforce` calls that task. The existing `UnownedWheelSpin` is the sole +owner selecting unowned rows: pair eligible idle units first, then allocate a +fresh destination only for the remaining unowned rows. Its stall watcher still +uses idle pairing only, so no spare leaves the old holder and strikes untouched. +`UnownedWork` and `WorkDispatch.reissue` are deleted. + +`WorkDispatch.publish` owns assignment publication for both destination lifetimes. +Fresh dispatch preflights storage, allocates its checkout and stopped native +registration, publishes, then calls `OwnerLifecycle.start`. `ReusedWorkDispatch` +publishes to an existing unit without allocating or starting another process. +The existing `WheelSpin` base builds the reissued action and runs both rule kinds; +the watcher retains its own assigner identity and revision-checked strike/release. +Failed or uncertain startup remains retained, never replayed or adopted. + +The task's exact problem identity and accepted installed order determine its +completion. Other closed subtasks, withdrawn orders and merely delivered +orders do not complete it. Finite positive deadline intervals are required. +The existing assignment trace requirement still applies on transfer. + +## Source relationships + +The before/after census used refactor-audit's existing `Package.load` parser +over Core, tools and Toad production roots. The retained census reports zero +parse omissions. NRA's parser was unavailable in the shared environment; +that limitation was not treated as a successful NRA scan. + +- `CommsStandingRulesTool.apply` and `commander.rotate.broken_rules` use + `StandingRule.broken/enforce`; no second rotation entrypoint was added. +- `PlanPhase` and `PlanIdle` consume the dependency-light `PhaseTask` owner. +- `CheckedPhaseTask` stays with dispatch; no compatibility re-export exists. +- `UnownedWheelSpin.watch/after_stall` remain the connector's entrypoints. + `tests/test_stall_watcher_strike.py` remains a consumer, including refusal, + blocked orders, no spare and root-release behavior. +- Only `UnownedWheelSpin.violations` selects unowned rows. There is no second + rule or generic standing-rule import of dispatch; that import would recreate + the storage/rule/thread cycle. +- The current source census is `P256-stall-owner-before.txt`: both retained and + main implementations, tests, tools and Toad were parsed through `Package.load`. + The first Toad path was absent; the corrected owned Toad checkout was appended. + Dynamic receiver resolution is not a behavioral proof. +- `ProblemOrder` owns refusal of unsupported reissue operations. +- `Registration` and `OwnerLifecycle` retain identity and process ownership; + no launch codec, second task store or copied fork history was introduced. +- Toad does not import the changed owner classes. Dynamic family discovery + and provider execution are runtime facts, not proved by the AST census. + +The CLI proof uses declaration-owned commands, reply states and the existing +`FieldCodec`. Observation uses `WireWatch`, not sleep polling. The configured +commander is read from its owner rather than compared with another literal. + +## Real native result + +Latest fresh run: `P256-native-20261019-1`, exit 0, source driver `643131ca3`. +No fake transport, provider, events, order or reply was used. + +- Unit: `unit-p1792441766503708626-r0` +- Configured model: `openai-codex/gpt-6.1-sol` +- Problem: `P1792441766503708626`; order: `ee1774f76fcb` +- Actual first reply: `ROTATION_REPLY_c9124a08ed18496fbf81c8a2ddff4ec3` +- Second rotation duplicated neither unit nor order. +- The actual reply passed the canonical order check. The unit stopped only + after its native turn ended; original native history remains intact. + +The later typed receipt decode was checked against that original saved order; +the read-only reply check also returned 1. This establishes source dispatch +through the configured native process, not candidate installation or full +production-phase completion. No new provider input is needed to retain it. + +## Source checks + +Source growth alone is not a defect under the owner's corrected ruling. No new +budget-only or formatting-only commit is used for this reconciliation. Both +production roots still need their structural checks; the checker and baselines +are not modified here. + +## Retained evidence and limits + +Owned scratch: `/home/ts/.cache/agent-scratch/pi-cn-phase7/`. Keep the original +native run immutable, including its private bus, order, reply and native history. +Before/after source observations and bounded test/ratchet results are in the +same owned directory. Disposable test/build output may be retired after borrower +checks; user sessions and uncertain inputs are not disposable output. + +Earlier native proof `run-20261019-control2` also passed. The initial control1 +attempt was refused before spawn for an unavailable prepared native package; +it was not replayed or hidden. No candidate install or PR merge was performed. + +The old root staging patch is removed: it described a superseded task API. +The full ControlPlane brief belongs to PR #1932's body and shared task owners, +not a second inactive implementation copied into the repository. diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index 7cd4eb20bd..8b6b85eb00 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -3,6 +3,7 @@ from __future__ import annotations import time +from abc import abstractmethod from collections.abc import Callable, Iterable, Iterator from contextlib import AbstractContextManager from dataclasses import dataclass, field @@ -22,6 +23,7 @@ from ..threads import Thread from ..todos import FlatTrend, OrderProgress, PullRequests, Todo from ..turn_digests import OrderEditCountRead, TurnDigestRead, TurnsSince +from ..work_dispatch import ReusedWorkDispatch, WorkDispatch from .config import CommanderConfig @@ -83,16 +85,22 @@ def idle(cls, comms: Comms, commander: str) -> list[Thread]: and TurnDigestRead(comms.root, thread, bound).read().turns), key=lambda thread: thread.name) - def assign(self, comms: Comms, commander: str, unit: Thread, - assigner: ThreadIncarnation) -> None: - """The order sent again to ``unit`` through its Goal, with the order's own text.""" - todo = self.todo - order = todo.order.reissued(time.time()) - comms.goals.update_goal(unit.name, OrderGoalAction( - problem=todo.id, expected_revision=todo.revision, order=order, - request=todo.goal.request(), deadline=todo.deadline, commander=commander, - assigner=assigner, traced=todo.order.owner)) - comms.messaging.send(commander, unit.name, f"{self.line()}\n{order.instructions(todo)}") + @staticmethod + def order_for(todo: Todo, commander: str, assigner: ThreadIncarnation) -> OrderGoalAction: + return OrderGoalAction( + problem=todo.id, expected_revision=todo.revision, + order=todo.order.reissued(time.time()), request=todo.goal.request(), + deadline=todo.deadline, commander=commander, assigner=assigner, + traced=todo.order.owner) + + @abstractmethod + def dispatches(self, comms: Comms, commander: str) -> Iterable[WorkDispatch]: ... + + def enforce(self, comms: Comms, commander: str) -> None: + for dispatch in self.dispatches(comms, commander): + unit = dispatch.start(comms) + comms.messaging.send(commander, unit.name, + f"{self.line()}\n{dispatch.action.order.instructions(self.todo)}") @dataclass(frozen=True) @@ -121,33 +129,44 @@ def line(self) -> str: " to done did not fall across two readings -> Core strikes it and gives the order" " to an idle unit") - def enforce(self, comms: Comms, commander: str) -> None: + def dispatches(self, comms: Comms, commander: str) -> Iterable[WorkDispatch]: for unit in self.idle(comms, commander)[:1]: - self.assign(comms, commander, unit, StallWatcherIncarnation.watcher()) + yield ReusedWorkDispatch(self.order_for( + self.todo, commander, StallWatcherIncarnation.watcher()), self.todo.revision, unit) @dataclass(frozen=True) class UnownedWheelSpin(WheelSpin, UndatedStandingRule): """A unit holds no next order while the board has unowned rows.""" - unit: str + dispatch: WorkDispatch + + @property + def unit(self) -> str: + return self.dispatch.name @classmethod def violations(cls, comms: Comms, commander: str) -> tuple[UnownedWheelSpin, ...]: - return cls.paired(comms, commander, (todo for todo in - comms.goals.todos.for_commander(comms.registry.require(commander).incarnation) - if not todo.holder)) + actor = comms.registry.require(commander).incarnation + rows = tuple(todo for todo in comms.goals.todos.for_commander(actor) + if todo.pending and not todo.holder) + paired = cls.paired(comms, commander, rows, actor) + assigned = {rule.todo.id for rule in paired} + return paired + tuple(cls(todo, WorkDispatch( + cls.order_for(todo, commander, actor), todo.revision)) + for todo in rows if todo.id not in assigned) @classmethod - def paired(cls, comms: Comms, commander: str, - todos: Iterable[Todo]) -> tuple[UnownedWheelSpin, ...]: + def paired(cls, comms: Comms, commander: str, todos: Iterable[Todo], + assigner: ThreadIncarnation) -> tuple[UnownedWheelSpin, ...]: """Each pending row with the first idle unit not struck on it.""" idle, found = cls.idle(comms, commander), [] for todo in (todo for todo in todos if todo.pending): for unit in idle: if not todo.live_strikes(unit.name): idle.remove(unit) - found.append(cls(todo, unit.name)) + found.append(cls(todo, ReusedWorkDispatch( + cls.order_for(todo, commander, assigner), todo.revision, unit))) break return tuple(found) @@ -155,9 +174,8 @@ def line(self) -> str: return (f"WHEEL SPIN: {self.unit} holds no order while {self.todo.id} is unowned" f" -> Core gives it to {self.unit}") - def enforce(self, comms: Comms, commander: str) -> None: - self.assign(comms, commander, comms.registry.require(self.unit), - comms.registry.require(commander).incarnation) + def dispatches(self, comms: Comms, commander: str) -> Iterable[WorkDispatch]: + return (self.dispatch,) @classmethod def after_stall(cls, comms: Comms) -> None: @@ -165,9 +183,8 @@ def after_stall(cls, comms: Comms) -> None: stalled = comms.goals.todos.stalled(comms.registry.storage_rule) for commander in {todo.creator_identity.name for todo in stalled}: mine = (todo for todo in stalled if todo.creator_identity.name == commander) - for rule in cls.paired(comms, commander, mine): - rule.assign(comms, commander, comms.registry.require(rule.unit), - StallWatcherIncarnation.watcher()) + for rule in cls.paired(comms, commander, mine, StallWatcherIncarnation.watcher()): + rule.enforce(comms, commander) @classmethod def watch_pass(cls, opened_wire: Callable[[], AbstractContextManager[Comms]]) -> None: diff --git a/src/agent_comms/standing_rules.py b/src/agent_comms/standing_rules.py index 3fbf3b08a9..18800b3d29 100644 --- a/src/agent_comms/standing_rules.py +++ b/src/agent_comms/standing_rules.py @@ -217,10 +217,13 @@ def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: if phase.ready(todos) and not phase.ordered(todos) ) + def enforce(self, comms: Comms, commander: str) -> None: + self.phase.task.start(comms, comms.registry.require(commander), self.phase) + def line(self) -> str: return ( f"PLAN IDLE: phase {self.phase.declared_name} ready, no open order" - " -> dispatch it this wake" + f" -> {self.phase.task.instruction}" ) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index ea58133091..675bdecef0 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -1313,7 +1313,6 @@ def instructions(self) -> tuple[str, ...]: class ProblemOrder(DeclaredFamily, affix="ProblemOrder"): """A problem's order state; an unassigned finding is not a fabricated order.""" pending: ClassVar[bool] = False - accepted: ClassVar[bool] = False line: ClassVar[str] = "No order" pull_request_label: ClassVar[str] = "Agent-Comms-Order:" finding: ProblemFinding = field(default_factory=UnlocatedProblemFinding, @@ -1338,6 +1337,10 @@ def phase(self) -> type[OrderPhase]: def for_phase(self, phase: type[OrderPhase]) -> ProblemOrder: raise ValueError("Only a checked order can declare a plan phase.") + @property + def accepted(self) -> bool: + return False + def require_transfer(self, todo: Todo) -> None: raise ValueError("This problem cannot receive another order.") @@ -1634,9 +1637,12 @@ def waiting_on_install(self, todo: Todo, read: dict[str, OrderProgress]) -> tupl class InstalledProblemOrder(DeliveredProblemOrder): works: str artifact: str - accepted: ClassVar[bool] = True line: ClassVar[str] = "Installed" + @property + def accepted(self) -> bool: + return True + def __post_init__(self): super().__post_init__() if not self.works.strip() or not self.artifact.strip(): @@ -1647,6 +1653,9 @@ def installed(self, works, artifact): raise ValueError("The order already records different installed evidence.") return self + def merged(self, sha: str) -> ProblemOrder: + return self.copy_as(InstalledMergedProblemOrder, sha=sha) + def require_close(self): pass @@ -1654,6 +1663,11 @@ def waiting_on_install(self, todo: Todo, read: dict[str, OrderProgress]) -> tupl return () +@dataclass(frozen=True, kw_only=True) +class InstalledMergedProblemOrder(MergedProblemOrder, InstalledProblemOrder): + """An installed merge retains the application's existing evidence, without a wrapper.""" + + @dataclass(frozen=True, kw_only=True) class WithdrawnProblemOrder(CheckedProblemOrder): """An order the commander sent in error and retired before delivery; it strikes no one.""" @@ -2419,6 +2433,8 @@ def assigned(self, proposed: Assignment, traced: FileLine | None = None) -> Todo if self.assignment is not None: self.assignment.require_trace(proposed.owner, traced) proposed = self.assignment.successor(proposed, self) + elif self.released: + self.released[-1].previous.require_trace(proposed.owner, traced) return replace(self, assignment=proposed, revision=self.revision + 1) def reassigned(self, assigned: Todo, actor: ThreadIncarnation) -> Todo: diff --git a/src/agent_comms/work_dispatch.py b/src/agent_comms/work_dispatch.py index e9e880a3ec..cdcac546fb 100644 --- a/src/agent_comms/work_dispatch.py +++ b/src/agent_comms/work_dispatch.py @@ -1,4 +1,4 @@ -"""Checked work starts through Core's existing order and native lifecycle owners. +"""Core assigns checked work to its unit, then starts it through the native lifecycle. A phase declares an existing PendingProblemOrder, not a second task format. Work without that declaration stays visible; readiness is not permission to invent an @@ -7,6 +7,7 @@ from __future__ import annotations +import math import subprocess import time from collections.abc import Iterable @@ -16,7 +17,6 @@ from .goal_actions import ( CreateProblemGoalAction, GoalActionContext, - OrderGoalAction, ProblemOrderGoalAction, ) from .native_settings import NativeThreadDefaults @@ -34,16 +34,24 @@ @dataclass(frozen=True) class WorkDispatch: - """One checked native order: allocate its unit, record its claim, then start it.""" + """Allocate a unit, assign its checked native order, then start it.""" action: ProblemOrderGoalAction revision: int + @property + def name(self) -> str: + return f"unit-{self.action.problem.lower()}-r{self.revision}" + + def publish(self, comms: Comms, child: Thread) -> Thread: + """Assign the native order before its unit can begin work.""" + comms.goals.update_goal(child.name, self.action) + return comms.registry.require(child.name) + def start(self, comms: Comms) -> Thread: defaults = NativeThreadDefaults() - # A concurrent rotation of this same row contends on one git checkout, - # not on two newly generated unit names. Never adopt a partial attempt. - name = f"unit-{self.action.problem.lower()}-r{self.revision}" + # A concurrent rotation contends on one checkout. Never adopt a partial attempt. + name = self.name storage = comms.registry.storage_rule checkout = storage.order_checkout(name, self.action.problem) candidate = comms.registry.new_thread( @@ -86,28 +94,24 @@ def start(self, comms: Comms) -> Thread: StoppedThreadStatus(), ) # A failed publication leaves a stopped unit and its checkout, not an - # untracked running task. A failed/uncertain start never releases the claim. - comms.goals.update_goal(child.name, self.action) + # untracked running task. A failed/uncertain start never releases its order. + self.publish(comms, child) comms.owners.start(child.name) return comms.registry.require(child.name) - @classmethod - def reissue(cls, comms: Comms, commander: Thread, todo: Todo) -> None: - """Retained checked work needs no already-idle unit.""" - current = comms.goals.todos.get(todo.id) - current.require_revision(todo.revision) - if current.holder or not current.pending: - return - action = OrderGoalAction( - problem=current.id, - expected_revision=current.revision, - order=current.order.reissued(time.time()), - request=current.goal.request(), - deadline=current.deadline, - commander=commander.name, - assigner=commander.incarnation, - ) - cls(action, revision=current.revision).start(comms) + +@dataclass(frozen=True) +class ReusedWorkDispatch(WorkDispatch): + """Publish to an existing unit without allocating or starting another process.""" + + unit: Thread + + @property + def name(self) -> str: + return self.unit.name + + def start(self, comms: Comms) -> Thread: + return self.publish(comms, comms.registry.require(self.name)) @dataclass(frozen=True) @@ -119,19 +123,19 @@ class CheckedPhaseTask(PhaseTask): text: str order: PendingProblemOrder deadline_seconds: float - instruction = "Core starts the declared checked task in its own unit checkout" + instruction = "Core assigns the declared checked task to a unit and starts it" def __post_init__(self) -> None: - if not 0 < self.deadline_seconds <= 24 * 60 * 60: - raise ValueError("A phase task needs a positive deadline of at most one day.") + if not math.isfinite(self.deadline_seconds) or self.deadline_seconds <= 0: + raise ValueError("A phase task needs a finite positive deadline interval.") def start(self, comms: Comms, commander: Thread, phase: type[PlanPhase]) -> None: history = comms.goals.todos.history_for_commander(commander.incarnation) if not phase.ready(history) or phase.ordered(history): return - # An existing unowned row belongs to UnownedWork, which reissues that - # row's actual order instead of replacing it with a phase template. - if any(todo.state.open for todo in phase.serving(history)): + # This task's existing row belongs to UnownedWheelSpin. Other phase subtasks + # neither supply this task's check nor replace its explicit identity. + if any(todo.id == self.problem for todo in history): return now = time.time() order = replace(self.order.reissued(now), serves=phase) diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index 5fdf42d87d..19813c61c7 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -109,8 +109,9 @@ def test_blocked_held_orders_are_not_struck_or_released(commander, commander_scr comms = Comms(copied) standing.UnownedWheelSpin.after_stall(comms) with pytest.raises(TodoConflict): - standing.UnownedWheelSpin(blocked, 'spare').assign( - comms, 'lead', comms.registry.require('spare'), StallWatcherIncarnation.watcher()) + action = standing.WheelSpin.order_for(blocked, 'lead', StallWatcherIncarnation.watcher()) + dispatch = standing.ReusedWorkDispatch(action, blocked.revision, comms.registry.require('spare')) + standing.UnownedWheelSpin(blocked, dispatch).enforce(comms, 'lead') assert comms.goals.todos.get('P1') == blocked == original.goals.todos.get('P1') diff --git a/tests/test_standing_rules.py b/tests/test_standing_rules.py index bfcd5ba98f..4053c77e2d 100644 --- a/tests/test_standing_rules.py +++ b/tests/test_standing_rules.py @@ -107,7 +107,8 @@ def idle(): assert RecordPlanePhase in idle() and ReviewQueuePhase not in idle() assert PlanIdle(RecordPlanePhase).line() == ( - "PLAN IDLE: phase record_plane ready, no open order -> dispatch it this wake") + "PLAN IDLE: phase record_plane ready, no open order -> " + "no checked task declared; supply its source, check and brief") create(commander, "P1") assert comms.goals.todos.get("P1").order.serves is UnplannedPhase assert RecordPlanePhase in idle() @@ -241,10 +242,17 @@ def progress(): taken = StrikeReason("The unit was taken off P2 so the test leaves it unowned.") comms.goals.fire("worker", taken, "lead") assert comms.goals.todos.get("P2").holder is None - read = CommsStandingRulesTool().apply(comms, ThreadCaller("lead")) - assert ("WHEEL SPIN: worker holds no order while P2 is unowned -> Core gives it to worker" - in read.broken) - assert comms.goals.todos.get("P2").holder == "worker" + [unowned] = module.UnownedWheelSpin.violations(comms, "lead") + assert unowned.todo.id == "P2" + # Firing releases P2 without striking its recently active worker on that order. + # Reuse that eligible idle unit instead of allocating a second process. + assert type(unowned.dispatch) is module.ReusedWorkDispatch + assert unowned.unit == "worker" + assert unowned.line() == ( + f"WHEEL SPIN: {unowned.unit} holds no order while P2 is unowned" + f" -> Core gives it to {unowned.unit}") + # Actual unit launch/first reply is a separate native application proof. + assert comms.goals.todos.get("P2").holder is None def test_unjudged_follows_the_unjudged_edits(commander, standing): diff --git a/tests/test_work_dispatch.py b/tests/test_work_dispatch.py new file mode 100644 index 0000000000..33008f549e --- /dev/null +++ b/tests/test_work_dispatch.py @@ -0,0 +1,123 @@ +"""State-level sanity checks, not the required native startup/first-reply proof.""" + +from dataclasses import replace + +import pytest + +from agent_comms.commander.standing import UnownedWheelSpin +from agent_comms.field_codec import FieldCodec +from agent_comms.goal_actions import CreateProblemGoalAction +from agent_comms.plan_phases import RecordPlanePhase +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.todos import DoneTodoState, OperationDelivery, ProblemOrder, StrikeReason +from agent_comms.phase_tasks import UnspecifiedPhaseTask +from agent_comms.work_dispatch import CheckedPhaseTask, ReusedWorkDispatch, WorkDispatch + +from test_commander_problem_orders import create, order +from test_standing_rules import working + + +def test_phase_task_accepts_only_its_full_problem_with_installed_evidence(commander): + comms, _ = commander + full, subtask = (create(commander, name) for name in ("P1", "P2")) + task = CheckedPhaseTask(full.id, full.repo, full.text, full.order, 1800) + accepted = full.order.delivered( + OperationDelivery(at=RecordedEventTime(200)), + ).installed("The installed application now completes the declared full phase workflow.", "run.txt") + assert not task.done((full, subtask)) + assert not task.done((replace(subtask, state=DoneTodoState, order=accepted),)) + assert not task.done((replace(full, state=DoneTodoState),)) + assert not task.done((replace(full, order=accepted),)) + assert task.done((replace(full, state=DoneTodoState, order=accepted),)) + assert not UnspecifiedPhaseTask().done((replace(full, state=DoneTodoState, order=accepted),)) + for original, expected in ((full.order, False), (accepted, True)): + encoded = FieldCodec.encode(original.merged("a" * 40)) + assert "original" not in encoded + if expected: + assert encoded["works"] == original.works + assert encoded["artifact"] == original.artifact + merged = FieldCodec.decode(ProblemOrder, encoded) + assert task.done((replace(full, state=DoneTodoState, order=merged),)) is expected + + +@pytest.mark.parametrize("interval", [0, -1, float("nan"), float("inf")]) +def test_checked_task_requires_a_finite_positive_deadline(commander, interval): + comms, _ = commander + with pytest.raises(ValueError, match="finite positive"): + CheckedPhaseTask("P1", "org/repo", "Full phase", order(comms.root, "full"), interval) + + +def test_unspecified_phase_cannot_allocate_or_fabricate_an_order(commander): + comms, _ = commander + lead = comms.registry.require("lead") + before = comms.registry.all_threads() + task = UnspecifiedPhaseTask() + task.start(comms, lead, RecordPlanePhase) + assert comms.registry.all_threads() == before + assert comms.goals.todos.history_for_commander(lead.incarnation) == () + + +def test_invalid_request_refuses_dispatch_before_allocating_anything(commander): + comms, _ = commander + checked = order(comms.root, "full-phase-order") + action = CreateProblemGoalAction( + problem="P1", repo="OpenHCSDev/agent-comms", text="Repair the full phase", + order=checked, request="not-the-active-request", commander="lead", + deadline=RecordedEventTime(1000), + ) + name = "unit-p1-r0" + path = comms.registry.storage_rule.order_checkout(name, "P1") + assert not path.exists() + before = comms.registry.all_threads() + with pytest.raises(ValueError): + WorkDispatch(action, revision=0).start(comms) + assert not path.exists() + assert comms.registry.all_threads() == before + + +def test_one_rule_pairs_idle_units_before_allocating_unowned_fallbacks(commander): + comms, _ = commander + create(commander, "P1") + create(commander, "P2") + comms.goals.fire("worker", StrikeReason("The unit stopped work on both orders and left their changes unfinished."), "lead") + spare = comms.registry.declare(comms.registry.new_thread( + "spare", frozenset({"comms"}), str(comms.root), created_at=3.0)) + working(comms, spare.name, spare.created_at) + rules = UnownedWheelSpin.violations(comms, "lead") + assert {rule.todo.id for rule in rules} == {"P1", "P2"} + assert len({rule.unit for rule in rules}) == 2 + reused = [rule for rule in rules if isinstance(rule.dispatch, ReusedWorkDispatch)] + fresh = [rule for rule in rules if type(rule.dispatch) is WorkDispatch] + assert len(reused) == len(fresh) == 1 + assert reused[0].unit == spare.name + assert not fresh[0].todo.holder + # Selection must not allocate or launch either destination. + assert set(comms.registry.all_threads()) == {"lead", "worker", "spare"} + + +def test_released_order_preserves_the_two_owner_trace_requirement(commander): + comms, _ = commander + lead, worker = (comms.registry.require(name) for name in ("lead", "worker")) + second, third = ( + comms.registry.declare(comms.registry.new_thread(name, frozenset(), str(comms.root))) + for name in ("second", "third") + ) + todo = create(commander, "P1") + reason = StrikeReason("The previous owner stopped work and left the checked order unfinished.") + comms.goals.todos.fire(worker.name, lead.incarnation, reason) + released = comms.goals.todos.get(todo.id) + assigned = comms.goals.todos.assign( + todo.id, expected_revision=released.revision, owner=second, assigner=lead.incarnation, + generation="second-generation", goal=todo.goal, + deadline=todo.deadline, order=todo.order.reissued(200), + ) + assert assigned.holder == second.name + comms.goals.todos.fire(second.name, lead.incarnation, reason) + released = comms.goals.todos.get(todo.id) + with pytest.raises(ValueError, match="pass --traced"): + comms.goals.todos.assign( + todo.id, expected_revision=released.revision, owner=third, assigner=lead.incarnation, + generation="third-generation", goal=todo.goal, + deadline=todo.deadline, order=todo.order.reissued(300), + ) + assert comms.goals.todos.get(todo.id) == released diff --git a/tools/prove_commander_dispatch.py b/tools/prove_commander_dispatch.py new file mode 100644 index 0000000000..6b91de9b82 --- /dev/null +++ b/tools/prove_commander_dispatch.py @@ -0,0 +1,287 @@ +"""Run real checked-order native dispatch on a new scratch bus and retain its reply. + +No fake transport, provider or history is used. Failed attempts remain retained. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import shlex +import subprocess +import sys +import time +from abc import abstractmethod +from contextlib import aclosing +from dataclasses import dataclass +from pathlib import Path +from typing import Annotated, ClassVar +from uuid import uuid4 + +from agent_comms.change_report import FileLine +from agent_comms.cli_bridges.shared import DeclaredOptions +from agent_comms.commander.config import CommanderConfig +from agent_comms.commander.order import CommanderContext +from agent_comms.comms import Comms +from agent_comms.declared_family import DeclaredFamily +from agent_comms.field_codec import FieldCodec, PathText, option +from agent_comms.goal_actions import SetGoalAction +from agent_comms.goals import CountCheckResult, OrderCheck, RequestGoalOrder +from agent_comms.messages import Message +from agent_comms.plan_phases import RecordPlanePhase +from agent_comms.progress_claims import WorkingClaim +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.repo import Repo +from agent_comms.storage_rules import StorageRule +from agent_comms.thread_execution import ExternalThreadExecution +from agent_comms.thread_status import StoppedThreadStatus +from agent_comms.todos import MechanicalOrderRun, OperationDelivery, PendingProblemOrder +from agent_comms.tool_caller import ThreadCaller +from agent_comms.tools import CommsHistoryTool, CommsOrderDoneTool, CommsStandingRulesTool +from agent_comms.wire_watch import WatchSource, WireWatch +from agent_comms.work_dispatch import CheckedPhaseTask + + +class Reply(DeclaredFamily, affix="Reply"): + arrived: ClassVar[bool] = False + + @classmethod + def read(cls, comms: Comms, commander: str, unit: str, token: str) -> Reply: + history = CommsHistoryTool(peer=unit, limit=20).apply(comms, ThreadCaller(commander)) + for message in history.messages: + if message.sender == unit and token in message.body: + return ReceivedReply(message) + return AwaitedReply() + + @abstractmethod + def require_message(self) -> Message: ... + + +class AwaitedReply(Reply): + def require_message(self) -> Message: + raise TimeoutError("The native unit has not replied; retain its original attempt") + + +@dataclass(frozen=True) +class ReceivedReply(Reply): + message: Message + arrived = True + + def require_message(self) -> Message: + return self.message + + +async def first_reply(comms: Comms, commander: str, unit: str, token: str) -> Message: + async with asyncio.timeout(180): + async with aclosing( + WireWatch.observations( + WatchSource.modified_file(comms.bus.log.path), + WatchSource.modified_file(comms.registry.store.path), + ) + ) as changes: + async for _ in changes: + observed = Reply.read(comms, commander, unit, token) + if observed.arrived: + return observed.require_message() + if not comms.registry.require(unit).process_alive: + raise RuntimeError(f"Unit exited before replying; inspect {comms.root}") + raise TimeoutError(f"No first reply; preserve {comms.root}") + + +async def stop_after_turn(comms: Comms, unit: str) -> None: + async with asyncio.timeout(180): + async with aclosing( + WireWatch.observations( + WatchSource.modified_file(comms.registry.store.path), + ) + ) as changes: + async for _ in changes: + turn = comms.registry.require(unit).turn_record.read() + if not turn.executing: + turn.require_idle() + comms.owners.stop(unit) + return + + +@dataclass(frozen=True, kw_only=True) +class ProofCommand(DeclaredFamily, DeclaredOptions, affix="ProofCommand"): + root: Annotated[Path, PathText] = option("--root") + + @abstractmethod + def apply(self, project: Path) -> int: ... + + +@dataclass(frozen=True, kw_only=True) +class CheckProofCommand(ProofCommand): + help = "check the actual native reply" + commander: str = option("--commander") + unit: str = option("--unit") + token: str = option("--token") + + def apply(self, project: Path) -> int: + print(int(Reply.read(Comms(self.root), self.commander, self.unit, self.token).arrived)) + return 0 + + +@dataclass(frozen=True, kw_only=True) +class RunProofCommand(ProofCommand): + help = "create a fresh scratch root and run its real rotation proof" + + def apply(self, project: Path) -> int: + root = self.root.expanduser().resolve() + allowed = StorageRule().scratch_base + if not root.is_relative_to(allowed) or root.exists(): + raise ValueError("Proof needs a NEW root under the owner's agent-scratch directory") + home = root.parent / "home" + home.mkdir(parents=True) + comms = Comms(root) + comms.messaging.initialize_private_initial_protocol() + # A fresh interpreter acquires the scratch commander configuration. + # Keep the actual storage declaration across its isolated HOME. + (root / "storage.json").write_text( + json.dumps(FieldCodec.encode(comms.registry.storage_rule)) + ) + commander = comms.registry.declare( + comms.registry.new_thread( + "dispatch-proof-commander", + frozenset(), + str(project), + model="anthropic/claude-sonnet-4-5", + execution=ExternalThreadExecution, + ), + StoppedThreadStatus(), + ) + comms.goals.update_goal( + commander.name, + SetGoalAction( + text="Verify one rotation-started native unit on this scratch bus only.", + goal_id=uuid4().hex, + order=RequestGoalOrder(), + ), + ) + CommanderConfig( + commander=commander.name, + root=root, + cli=Path(sys.executable).with_name("agent-comms"), + state=root.parent / "state", + board=root.parent / "board.md", + narrative=root.parent / "narrative.md", + repos=(), + worktrees=comms.registry.storage_rule.checkout_base, + scratch=root.parent, + disk=root.parent, + ).write(home) + return subprocess.run( + (sys.executable, __file__, "observe", "--root", str(root)), + env={**os.environ, "HOME": str(home)}, + check=False, + ).returncode + + +@dataclass(frozen=True, kw_only=True) +class ObserveProofCommand(ProofCommand): + help = "observe the freshly configured scratch rotation" + + def apply(self, project: Path) -> int: + root = self.root.resolve() + context = CommanderContext.read_root(root) + comms = context.comms + commander = comms.registry.require(context.commander) + problem = f"P{time.time_ns()}" + unit = f"unit-{problem.lower()}-r0" + token = f"ROTATION_REPLY_{uuid4().hex}" + check_command = shlex.join( + ( + sys.executable, + str(Path(__file__).resolve()), + "check", + "--root", + str(root), + "--commander", + commander.name, + "--unit", + unit, + "--token", + token, + ) + ) + order = PendingProblemOrder( + id=uuid4().hex, + owner=FileLine(__file__, 1), + check=OrderCheck(str(project), check_command, CountCheckResult(1)), + edits=( + f"Use comms_send to send exactly {token} to {commander.name}, then return. " + "Verify your first real reply; make no source edits.", + ), + banned=("No helper agents, live-bus writes, source edits or extra tool calls.",), + report="Send the requested token once and return; the proof commander checks delivery.", + at=RecordedEventTime(1), + first_call_deadline=RecordedEventTime(121), + run=MechanicalOrderRun(), + ) + RecordPlanePhase.task = CheckedPhaseTask( + problem, + Repo.of(project).github, + "Verify rotation starts a native unit and its reply.", + order, + deadline_seconds=300, + ) + print(f"ROOT {root}", flush=True) + source = subprocess.check_output( + ("git", "-C", str(project), "rev-parse", "HEAD"), text=True + ) + print(f"SOURCE {source.strip()}", flush=True) + for line in context.call(CommsStandingRulesTool()).broken: + print(line, flush=True) + assigned = comms.goals.todos.get(problem) + received_order = FieldCodec.decode(PendingProblemOrder, FieldCodec.encode(assigned.order)) + started = comms.registry.require(assigned.held().owner) + print( + f"UNIT {started.name} model={started.model} pid={started.require_process().pid}", + flush=True, + ) + print( + f"ORDER {received_order.id} problem={problem} checkout={started.worktree}", flush=True + ) + (root.parent / "order.json").write_text(json.dumps(FieldCodec.encode(assigned), indent=2)) + before = comms.registry.all_threads() + for line in context.call(CommsStandingRulesTool()).broken: + print(line, flush=True) + assert comms.registry.all_threads().keys() == before.keys(), "duplicate unit" + after = FieldCodec.decode( + PendingProblemOrder, FieldCodec.encode(comms.goals.todos.get(problem).order) + ) + assert after.id == received_order.id, "duplicate order" + print("SECOND_ROTATION no duplicate unit or order", flush=True) + found = asyncio.run(first_reply(comms, commander.name, started.name, token)) + print(f"FIRST_REPLY {found.sender}: {found.body}", flush=True) + artifact = root.parent / "reply.json" + artifact.write_text(json.dumps(FieldCodec.encode(found), indent=2)) + CommsOrderDoneTool( + problem=problem, + order_id=received_order.id, + delivery=OperationDelivery(), + claims=(WorkingClaim(str(artifact)),), + ).apply(comms, ThreadCaller(commander.name)) + asyncio.run(stop_after_turn(comms, started.name)) + print("STOPPED after its turn ended; original history retained", flush=True) + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + for member in ProofCommand.members_with(ProofCommand): + member.add_parser(commands) + arguments = parser.parse_args() + return ( + ProofCommand.member_named(arguments.command) + .from_arguments(arguments) + .apply(Path(__file__).resolve().parents[1]) + ) + + +if __name__ == "__main__": + raise SystemExit(main())