P216: gate commander cutover on declared native parity

plans/commander-native-plan.md

@@ -206,4 +206,4 @@ Toad is a view of Core, so the commander's page is one more view over Core's ans
 
 ## Order and dependencies
 
-The phases and what each waits for are declared once, in `src/agent_comms/plan_phases.py` (the `PlanPhase` family). An order names the phase it serves with a `Serves: <phase>` line, stored as `ProblemOrder.serves`; a phase is done when every order serving it is closed. Each rotation prints `PLAN IDLE` for every phase whose dependencies are done and that has no open order (`PlanIdle` in `src/agent_comms/standing_rules.py`).
+The phases and what each waits for are declared once, in `src/agent_comms/plan_phases.py` (the `PlanPhase` family). An order names the phase it serves with a `Serves: <phase>` line, stored as `ProblemOrder.serves`. A phase is done when its explicitly declared full-phase checked task has accepted installed evidence; closing related subtasks does not complete the phase. Earlier delivered phases record their delivery on their declarations. Each rotation prints `PLAN IDLE` for every phase whose dependencies are done and that has no open order (`PlanIdle` in `src/agent_comms/standing_rules.py`).

src/agent_comms/phase_tasks.py

@@ -0,0 +1,33 @@
+"""Phase task capabilities, independent of order storage and work dispatch."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import Iterable
+from typing import TYPE_CHECKING, ClassVar
+
+if TYPE_CHECKING:
+    from .comms import Comms
+    from .plan_phases import PlanPhase
+    from .threads import Thread
+    from .todos import Todo
+
+
+class PhaseTask(ABC):
+    instruction: ClassVar[str]
+
+    @abstractmethod
+    def start(self, comms: Comms, commander: Thread, phase: type[PlanPhase]) -> None: ...
+
+    @abstractmethod
+    def done(self, todos: Iterable[Todo]) -> bool: ...
+
+
+class UnspecifiedPhaseTask(PhaseTask):
+    instruction = "no checked task declared; supply its source, check and brief"
+
+    def start(self, comms: Comms, commander: Thread, phase: type[PlanPhase]) -> None:
+        """Readiness alone cannot authorize a fabricated order."""
+
+    def done(self, todos: Iterable[Todo]) -> bool:
+        return False

src/agent_comms/plan_phases.py

@@ -2,8 +2,9 @@
 
 This family is the one owner of the plan's order and dependencies;
 ``plans/commander-native-plan.md`` points here. An order names the phase it
-serves in ``ProblemOrder.serves``; an order outside the plan serves
-``UnplannedPhase``. A phase is done when every order serving it is closed.
+serves through ``ProblemOrder.phase``; an order outside the plan serves
+``UnplannedPhase``. A phase is done only when its declared full-phase task is accepted; closing
+related subtasks does not establish completion of the phase.
 """
 
 from __future__ import annotations
@@ -12,6 +13,7 @@
 from typing import TYPE_CHECKING, ClassVar
 
 from .declared_family import DeclaredFamily
+from .phase_tasks import PhaseTask, UnspecifiedPhaseTask
 
 if TYPE_CHECKING:
     from .todos import Todo
@@ -27,15 +29,15 @@ class UnplannedPhase(OrderPhase):
 
 class PlanPhase(OrderPhase, declared_name=None):
     after: ClassVar[tuple[type[PlanPhase], ...]] = ()
+    task: ClassVar[PhaseTask] = UnspecifiedPhaseTask()
 
     @classmethod
     def serving(cls, todos: Iterable[Todo]) -> tuple[Todo, ...]:
-        return tuple(todo for todo in todos if todo.order.serves is cls)
+        return tuple(todo for todo in todos if todo.order.phase is cls)
 
     @classmethod
     def done(cls, todos: Iterable[Todo]) -> bool:
-        serving = cls.serving(todos)
-        return bool(serving) and not any(todo.state.open for todo in serving)
+        return cls.task.done(cls.serving(todos))
 
     @classmethod
     def ordered(cls, todos: Iterable[Todo]) -> bool:
@@ -78,24 +80,31 @@ class RecordPlanePhase(PlanPhase):
     after = (RetainedStorePhase,)
 
 
-class ReviewQueuePhase(PlanPhase):
+class ReviewQueuePhase(MergedPhase):
     after = (RecordPlanePhase,)
+    delivered = "41e4ec7319432755d75dcd8dbbab35839a4eb005 (#1654)"
 
 
-class StoryPhase(PlanPhase):
-    after = (ReviewQueuePhase,)
+class ControlPlanePhase(PlanPhase):
+    after = (RecordPlanePhase,)
 
 
-class CommanderPagePhase(PlanPhase):
-    after = (StoryPhase,)
+class CutoverPhase(PlanPhase):
+    # The parity gate in the Markdown plan has no executable declaration yet.
+    # These dependencies do not establish parity or authorize deleting scripts.
+    after = (RecordPlanePhase, ReviewQueuePhase, ControlPlanePhase)
 
 
-class ControlPlanePhase(PlanPhase):
+class StoryPhase(PlanPhase):
+    """A view of the native records, not a prerequisite for their cutover."""
+
     after = (RecordPlanePhase,)
 
 
-class CutoverPhase(PlanPhase):
-    after = (ReviewQueuePhase, ControlPlanePhase)
+class CommanderPagePhase(PlanPhase):
+    """A view of the native records and control plane, independent of cutover."""
+
+    after = (RecordPlanePhase, ControlPlanePhase)
 
 
 class SkillPhase(PlanPhase):

src/agent_comms/todos.py

@@ -1313,6 +1313,7 @@ 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,
@@ -1330,6 +1331,13 @@ def pull_request_ids(cls, body: str) -> tuple[str, ...]:
     def checked(self) -> CheckedProblemOrder:
         raise ValueError("This problem has no checked order.")
 
+    @property
+    def phase(self) -> type[OrderPhase]:
+        return UnplannedPhase
+
+    def for_phase(self, phase: type[OrderPhase]) -> ProblemOrder:
+        raise ValueError("Only a checked order can declare a plan phase.")
+
     def require_transfer(self, todo: Todo) -> None:
         raise ValueError("This problem cannot receive another order.")
 
@@ -1413,6 +1421,13 @@ def checked(self) -> CheckedProblemOrder:
         default_factory=NoOrderDeletion, metadata={"wire_omit_default": True})
     serves: type[OrderPhase] = field(default=UnplannedPhase, metadata={"wire_omit_default": True})
 
+    @property
+    def phase(self) -> type[OrderPhase]:
+        return self.serves
+
+    def for_phase(self, phase: type[OrderPhase]) -> CheckedProblemOrder:
+        return replace(self, serves=phase)
+
     def __post_init__(self):
         problems = []
         if not self.edits or not self.banned:
@@ -1619,6 +1634,7 @@ 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"
 
     def __post_init__(self):
@@ -2167,7 +2183,7 @@ class ServeTodo(TodoOperation):
     def apply(self, current: Todo) -> Todo:
         current.require_revision(self.expected_revision)
         current.require_commander(self.actor)
-        return replace(current, order=replace(current.order, serves=self.serves),
+        return replace(current, order=current.order.for_phase(self.serves),
                        revision=current.revision + 1)
 
 

src/agent_comms/work_dispatch.py

@@ -0,0 +1,153 @@
+"""Checked work starts through Core's existing order and native lifecycle owners.
+
+A phase declares an existing PendingProblemOrder, not a second task format. Work
+without that declaration stays visible; readiness is not permission to invent an
+acceptance check. Failed startup state stays retained for explicit disposition.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import time
+from collections.abc import Iterable
+from dataclasses import dataclass, replace
+from typing import TYPE_CHECKING
+
+from .goal_actions import (
+    CreateProblemGoalAction,
+    GoalActionContext,
+    OrderGoalAction,
+    ProblemOrderGoalAction,
+)
+from .native_settings import NativeThreadDefaults
+from .phase_tasks import PhaseTask
+from .recorded_time import RecordedEventTime
+from .storage_rules import CheckoutStorageEffect
+from .thread_status import StoppedThreadStatus
+
+if TYPE_CHECKING:
+    from .comms import Comms
+    from .plan_phases import PlanPhase
+    from .threads import Thread
+    from .todos import PendingProblemOrder, Todo
+
+
+@dataclass(frozen=True)
+class WorkDispatch:
+    """One checked native order: allocate its unit, record its claim, then start it."""
+
+    action: ProblemOrderGoalAction
+    revision: int
+
+    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}"
+        storage = comms.registry.storage_rule
+        checkout = storage.order_checkout(name, self.action.problem)
+        candidate = comms.registry.new_thread(
+            name,
+            frozenset(),
+            str(checkout),
+            model=defaults.model,
+            thinking_level=defaults.thinking_level,
+        )
+        self.action.request_ref(GoalActionContext(comms.goals, candidate))
+        comms.registry.require_declaration_available(candidate, StoppedThreadStatus())
+        CheckoutStorageEffect(checkout).check(storage, candidate)
+        subprocess.run(
+            ("git", "-C", self.action.order.check.directory, "fetch", "origin", "main"),
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+        subprocess.run(
+            (
+                "git",
+                "-C",
+                self.action.order.check.directory,
+                "worktree",
+                "add",
+                "-b",
+                f"auto/{name}",
+                str(checkout),
+                "origin/main",
+            ),
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+        location = candidate.runtime.new_location(comms.root, checkout)
+        child = comms.registry.declare(
+            replace(candidate, session_file=location.read_key),
+            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)
+        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 CheckedPhaseTask(PhaseTask):
+    """An explicitly authored task; its order owns source, check, edits and run."""
+
+    problem: str
+    repo: str
+    text: str
+    order: PendingProblemOrder
+    deadline_seconds: float
+    instruction = "Core starts the declared checked task in its own unit checkout"
+
+    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.")
+
+    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)):
+            return
+        now = time.time()
+        order = replace(self.order.reissued(now), serves=phase)
+        action = CreateProblemGoalAction(
+            problem=self.problem,
+            repo=self.repo,
+            text=self.text,
+            order=order,
+            request=order.request(commander, (self.problem,)),
+            commander=commander.name,
+            deadline=RecordedEventTime(now + self.deadline_seconds),
+        )
+        WorkDispatch(action, revision=0).start(comms)
+
+    def done(self, todos: Iterable[Todo]) -> bool:
+        return any(
+            todo.id == self.problem and not todo.state.open and todo.order.accepted
+            for todo in todos
+        )