Phase 7: dispatch checked ready work through native orders

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.

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,53 +129,62 @@ 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)
 
     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:
         """Assign first: refused or unavailable replacements leave the old holder untouched."""
         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:

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

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,13 +1653,21 @@ 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
 
     def waiting_on_install(self, todo: Todo, read: dict[str, OrderProgress]) -> tuple[Todo, ...]:
         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: