diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 7e1eeae84b..888180377f 100644 --- a/plans/commander-native-plan.md +++ b/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: ` 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: ` 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`). diff --git a/src/agent_comms/phase_tasks.py b/src/agent_comms/phase_tasks.py new file mode 100644 index 0000000000..31d380dfd8 --- /dev/null +++ b/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 diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 2a78e0159f..c42d40d3c8 100644 --- a/src/agent_comms/plan_phases.py +++ b/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): diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 046202fe31..ae859510d7 100644 --- a/src/agent_comms/todos.py +++ b/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) diff --git a/src/agent_comms/work_dispatch.py b/src/agent_comms/work_dispatch.py new file mode 100644 index 0000000000..e9e880a3ec --- /dev/null +++ b/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 + ) diff --git a/tests/test_installed_merge.py b/tests/test_installed_merge.py index 4e435dbdbf..7618ebd849 100644 --- a/tests/test_installed_merge.py +++ b/tests/test_installed_merge.py @@ -2,14 +2,16 @@ from dataclasses import fields, replace import json import subprocess +import time import pytest from agent_comms.repo import Repo from agent_comms.field_codec import FieldCodec +from agent_comms.plan_phases import RecordPlanePhase from agent_comms.todos import (CheckedProblemOrder, DoneTodoState, MergedProblemOrder, OperationDelivery, MergedPullRequest, PullRequests, UnorderedProblemOrder) -from agent_comms.tools import CommsOrderDoneTool +from agent_comms.tools import CommsOrderDoneTool, CommsOrderServeTool from agent_comms.recorded_time import RecordedEventTime from cutover.carry_merged_orders import MergedOrders from test_commander_problem_orders import working @@ -29,6 +31,8 @@ def test_installed_merge_closes_pending_or_delivered_order_once( delivery=OperationDelivery(at=RecordedEventTime(500)), claims=working(comms)).apply(comms, caller) todo = comms.goals.todos.get(todo.id) + CommsOrderServeTool(problem=todo.id, serves=RecordPlanePhase).apply(comms, caller) + todo = comms.goals.todos.get(todo.id) original = todo.order commit(unit) git(unit, 'switch', '-q', 'main') @@ -43,8 +47,7 @@ def test_installed_merge_closes_pending_or_delivered_order_once( git(installed, 'fetch', '-q', 'origin', 'main') git(installed, 'checkout', '-q', '-B', 'main', base) repo = Repo(checkout=installed, github=todo.repo) - config = script.CONFIG.model_copy(update={'repos': (repo,)}) - monkeypatch.setattr(script, 'CONFIG', config) + actor = comms.registry.require(caller.name).incarnation merge = MergedPullRequest(1, 'fix/p1', sha, merge=sha, merged_at=500, updated_at=500, orders=(original.id,)) monkeypatch.setattr(PullRequests, 'read', lambda self, repo, state, order: @@ -52,14 +55,15 @@ def test_installed_merge_closes_pending_or_delivered_order_once( # The object being available locally is not installation: HEAD must contain it. assert not repo.contains(sha) - assert script.COMMANDER.orders(script.CONFIG.repos) == (todo,) + comms.goals.todos.reconcile_merges(actor, (repo,), comms.registry.storage_rule) + assert script.held_orders() == (todo,) assert comms.goals.todos.get(todo.id) == todo if not delivered: assert 'waiting on install' in deadline_lines(script)[0] git(installed, 'checkout', '-q', '-B', 'main', sha) assert repo.contains(sha) - assert script.COMMANDER.orders(script.CONFIG.repos) == () + comms.goals.todos.reconcile_merges(actor, (repo,), comms.registry.storage_rule) assert script.held_orders() == () closed = comms.goals.todos.get(todo.id) assert closed.state is DoneTodoState and closed.assignment is None @@ -70,6 +74,8 @@ def test_installed_merge_closes_pending_or_delivered_order_once( for item in fields(CheckedProblemOrder)) assert closed.order.id == original.id assert "original" not in FieldCodec.encode(closed.order) + assert closed.order.phase is RecordPlanePhase + assert RecordPlanePhase.serving((closed,)) == (closed,) assert closed.strikes == todo.strikes assert FieldCodec.decode(type(closed), FieldCodec.encode(closed)) == closed @@ -93,9 +99,10 @@ def no_child(*args, **kwargs): with monkeypatch.context() as repeat: repeat.setattr(subprocess, 'run', no_child) - assert script.COMMANDER.orders(script.CONFIG.repos) == () + comms.goals.todos.reconcile_merges(actor, (repo,), comms.registry.storage_rule) + assert script.held_orders() == () assert comms.goals.todos.get(todo.id) == closed - script.board() + script.write_board() assert f'| {todo.id}:' not in (tmp_path / 'board.md').read_text() assert deadline_lines(script) == [] assert comms.goals.todos.get(todo.id) == closed @@ -106,7 +113,7 @@ class TickComplete(Exception): def next_tick(seconds): raise TickComplete - monkeypatch.setattr(script.time, 'sleep', next_tick) + monkeypatch.setattr(time, 'sleep', next_tick) with pytest.raises(TickComplete): script.overdue() assert 'OVERDUE:' not in capsys.readouterr().out diff --git a/tests/test_phase_task_dependencies.py b/tests/test_phase_task_dependencies.py new file mode 100644 index 0000000000..02fd3986b7 --- /dev/null +++ b/tests/test_phase_task_dependencies.py @@ -0,0 +1,27 @@ +"""Phase declarations stay usable before action and storage modules are loaded.""" + +from pathlib import Path +import subprocess +import sys + +import pytest + +from agent_comms.phase_tasks import UnspecifiedPhaseTask +from agent_comms.todos import UnorderedProblemOrder + + +@pytest.mark.parametrize("module", ("plan_phases", "work_dispatch", "goal_actions", "todos")) +def test_import_order_does_not_reintroduce_dispatch_cycle(module): + checkout = Path(__file__).resolve().parents[1] + result = subprocess.run( + (sys.executable, "-c", f"import agent_comms.{module}"), + cwd=checkout, + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 0, result.stderr + + +def test_unspecified_phase_and_unordered_problem_cannot_start_checked_work(): + assert not UnspecifiedPhaseTask().done(()) + with pytest.raises(ValueError, match="Only an undelivered order"): + UnorderedProblemOrder().reissued(1) diff --git a/tests/test_plan_phases.py b/tests/test_plan_phases.py new file mode 100644 index 0000000000..669ccccd83 --- /dev/null +++ b/tests/test_plan_phases.py @@ -0,0 +1,147 @@ +from dataclasses import replace + +import pytest + +from agent_comms.field_codec import FieldCodec +from agent_comms.standing_rules import PlanIdle +from agent_comms.plan_phases import ( + CommanderPagePhase, + ControlPlanePhase, + CutoverPhase, + PlanPhase, + RecordPlanePhase, + SkillPhase, + StoryPhase, + UnplannedPhase, +) +from agent_comms.todos import ( + DoneTodoState, + FoldedProblemOrder, + InstalledProblemOrder, + ProblemOrder, + OperationDelivery, + UnorderedProblemOrder, +) +from agent_comms.tools import CommsOrderServeTool +from agent_comms.work_dispatch import CheckedPhaseTask + +from test_commander_problem_orders import create + + +def accepted(todo): + installed = todo.order.copy_as( + InstalledProblemOrder, delivery=OperationDelivery(), + works="The declared whole phase worked through its acceptance run.", + artifact="phase-acceptance.txt", + ) + return replace(todo, state=DoneTodoState, assignment=None, order=installed) + + +def task(todo): + return CheckedPhaseTask( + problem=todo.id, repo=todo.repo, text=todo.text, + order=todo.order, deadline_seconds=300, + ) + + +def test_views_wait_for_native_work_but_never_block_cutover(commander, monkeypatch): + comms, caller = commander + # ReviewQueuePhase is explicitly delivered as #1654, not inferred from a subtask. + required = (RecordPlanePhase, ControlPlanePhase) + phases = (*required, CutoverPhase, StoryPhase, CommanderPagePhase) + todos = [] + for index, phase in enumerate(phases, 1): + todo = create(commander, f"P{index}") + CommsOrderServeTool(problem=todo.id, serves=phase).apply(comms, caller) + todo = comms.goals.todos.get(todo.id) + monkeypatch.setattr(phase, "task", task(todo)) + todos.append(todo) + + record_done = tuple( + accepted(todo) if todo.order.serves is RecordPlanePhase else todo + for todo in todos + ) + assert StoryPhase.ready(record_done) + assert not CommanderPagePhase.ready(record_done) + + native_done = tuple( + accepted(todo) if todo.order.serves in required else todo + for todo in todos + ) + assert CutoverPhase.ready(native_done) + assert StoryPhase.ready(native_done) + assert CommanderPagePhase.ready(native_done) + assert not SkillPhase.ready(native_done) + + for unfinished in required: + snapshot = tuple( + todo if todo.order.serves in (unfinished, CutoverPhase) else accepted(todo) + for todo in todos + ) + assert not CutoverPhase.ready(snapshot) + + cutover_done = tuple( + accepted(todo) if todo.order.serves is CutoverPhase else todo + for todo in native_done + ) + assert StoryPhase.ready(cutover_done) + assert CommanderPagePhase.ready(cutover_done) + assert SkillPhase.ready(cutover_done) + assert not any( + dependency in (StoryPhase, CommanderPagePhase) + for phase in PlanPhase.members_with(PlanPhase) + for dependency in phase.after + ) + + +def test_unordered_findings_have_no_invented_phase_declaration(commander): + comms, caller = commander + todo = create(commander, "P1") + finding = replace(todo, assignment=None, order=UnorderedProblemOrder()) + with comms.goals.todos.writing() as db: + finding.upsert(db) + assert finding.order.phase is UnplannedPhase + assert not RecordPlanePhase.serving((finding,)) + assert [rule.phase for rule in PlanIdle.violations(comms, caller.name)] == [RecordPlanePhase] + with pytest.raises(ValueError, match="Only a checked order"): + CommsOrderServeTool(problem=finding.id, serves=StoryPhase).apply(comms, caller) + assert comms.goals.todos.get(finding.id) == finding + + +def test_merged_phase_is_inherited_and_round_trips_without_a_wrapper(commander): + todo = create(commander, "P1") + original = todo.order.for_phase(RecordPlanePhase) + order = original.merged("a" * 40) + saved = replace(todo, order=order, state=DoneTodoState, assignment=None) + assert RecordPlanePhase.serving((saved,)) == (saved,) + wire = FieldCodec.encode(order) + assert "original" not in wire + restored = FieldCodec.decode(ProblemOrder, wire) + assert restored == order and restored.phase is RecordPlanePhase + changed = order.for_phase(ControlPlanePhase) + assert changed.phase is ControlPlanePhase + assert original.phase is RecordPlanePhase + assert not order.accepted + folded = FoldedProblemOrder(original=original, into="P2", why="Keep one owner") + assert folded.phase is UnplannedPhase + + +def test_closed_related_subtask_does_not_accept_a_whole_phase(commander, monkeypatch): + comms, caller = commander + subtask = create(commander, "P149") + CommsOrderServeTool(problem=subtask.id, serves=StoryPhase).apply(comms, caller) + subtask = accepted(comms.goals.todos.get(subtask.id)) + assert not StoryPhase.done((subtask,)) + + whole = create(commander, "P999") + CommsOrderServeTool(problem=whole.id, serves=StoryPhase).apply(comms, caller) + whole = comms.goals.todos.get(whole.id) + monkeypatch.setattr(StoryPhase, "task", task(whole)) + assert not StoryPhase.done((subtask, whole)) + assert StoryPhase.done((subtask, accepted(whole))) + + withdrawn = whole.order.withdrawn(whole.held(), "The order was sent in error.", None) + assert not StoryPhase.done((replace(whole, state=DoneTodoState, order=withdrawn),)) + assert not UnorderedProblemOrder().accepted + wrong_phase = replace(accepted(whole), order=replace(accepted(whole).order, serves=ControlPlanePhase)) + assert not StoryPhase.done((wrong_phase,))