P287: gate stale and unplanned commander work; reconcile merged deliveries

src/agent_comms/commander/order.py

@@ -41,7 +41,7 @@
 from ..goals import OrderCheck, ProblemGoalOrder
 from ..message_reference import MessageReference
 from ..native_tool_call import SelectedToolDenied
-from ..plan_phases import OrderPhase
+from ..plan_phases import OrderPhase, PlanPhase
 from ..progress_claims import MergedClaim, WorkingClaim
 from ..recorded_time import RecordedEventTime
 from ..repo import Repo
@@ -179,6 +179,7 @@ class OrderFile:
     order: PendingProblemOrder
     defect: str
     why_stupid: str
+    unplanned_reason: str = ""
     problems: tuple[str, ...] = ()
     slop: tuple[Slop, ...] = ()
 
@@ -191,7 +192,8 @@ def read(cls, text: str):
 
     @property
     def prose(self) -> str:
-        return f'{self.defect}\n{self.why_stupid}'
+        reason = f'\nUnplanned reason: {self.unplanned_reason}' if self.unplanned_reason else ''
+        return f'{self.defect}\n{self.why_stupid}{reason}'
 
     def problems_found(self, clean: bool) -> list[str]:
         problems = [f'{name}: {error}; fix: state the defect plainly without wording rules on code'
@@ -323,6 +325,20 @@ def send(self, unit, due, document: OrderFile, clean=False, reassign=None, new=N
             problems.append('due: fix: supply HH:MM local time (00:00 through 23:59)')
         group = [new[0]] if new else list(document.problems)
         listed = {t.id: t for t in self.todos()}
+        history = self.comms.goals.todos.history_for_commander(self.incarnation)
+        problems.extend(order.phase.order_refusals(history, document.unplanned_reason))
+        held = tuple(todo for todo in listed.values() if todo.holder)
+        busy = tuple(todo for plan in PlanPhase.members_with(PlanPhase)
+                     for todo in plan.serving(held) if todo.holder == unit and todo.id not in group)
+        if busy:
+            snapshot = self.comms.registry.snapshot()
+            occupied = {todo.holder for todo in held} | {self.commander} | self.parked
+            idle = sorted(thread.name for thread in snapshot.threads.values()
+                          if thread.name not in occupied and thread.role.executor(thread)
+                          and snapshot.statuses[thread.name].active)
+            problems.append(f'{unit} already holds plan orders '
+                            f'{", ".join(todo.id for todo in busy)}; '
+                            f'running units holding no order: {", ".join(idle) or "none"}')
         if not group or any(not re.fullmatch(r'P[1-9]\d*', pid) for pid in group):
             problems.append('problems: fix: supply an array of P<n> IDs, or use order problem')
         if new and document.problems:
@@ -804,6 +820,23 @@ def parse(cls, argv: list[str], description: str) -> CommanderVerb:
         return cls.member_named(args.command).from_arguments(args)
 
 
+class StaleOrderGate(CommanderVerb):
+    """All directive verbs read the configured commander's same retained board."""
+
+    def gate(self, ctx: Commander) -> None:
+        super().gate(ctx)
+        stale = tuple(todo for todo in ctx.held_orders() if todo.overdue(time.time() - 30 * 60))
+        if stale:
+            lines = []
+            for todo in stale:
+                due = time.localtime(todo.deadline.recorded_at("An order deadline"))
+                lines.append(f'{todo.id}: holder {todo.holder}, deadline '
+                             f'{time.strftime("%Y-%m-%d %H:%M", due)}')
+            sys.exit('REFUSED, nothing recorded: orders more than 30 minutes overdue:\n  '
+                     + '\n  '.join(lines)
+                     + '\nMove their deadlines or withdraw them before sending more work.')
+
+
 class RuledVerb(CommanderVerb):
     """HARD RULE: no commander action, mine or a fork's, on rules not re-read in 20 min."""
 
@@ -857,7 +890,7 @@ class OrderVerb(CommanderVerb):
 
 
 @dataclass(frozen=True, kw_only=True)
-class SendVerb(OrderVerb, DirectiveVerb):
+class SendVerb(OrderVerb, DirectiveVerb, StaleOrderGate):
     help = "record the order in Core and send it"
     unit: str = option("unit")
     due: str = option("due", help="HH:MM local time")
@@ -876,7 +909,7 @@ def run(self, ctx: Commander) -> None:
 
 
 @dataclass(frozen=True, kw_only=True)
-class StrikeVerb(OrderVerb, DirectiveVerb):
+class StrikeVerb(OrderVerb, DirectiveVerb, StaleOrderGate):
     help = "a send-back, a missed deadline or a false claim"
     unit: str = option("unit")
     why: str = option("--why")
@@ -972,9 +1005,11 @@ class PullRequestVerb(OrderVerb, declared_name="pr"):
 
     def run(self, ctx: Commander) -> None:
         todo = ctx.comms.goals.todos.get(self.pid)
-        checkout = ctx.comms.registry.storage_rule.order_checkout(todo.holder, todo.id)
+        checkout = Repo.of(Path.cwd())
+        if checkout.github.lower() != todo.repo.lower():
+            sys.exit(f'Refused: current checkout belongs to {checkout.github}, not {todo.repo}.')
         print(todo.order.open_pull_request(
-            todo, checkout, self.title, Path(self.body_file).read_text()))
+            todo, checkout.checkout, self.title, Path(self.body_file).read_text()))
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -1000,7 +1035,7 @@ def run(self, ctx: Commander) -> None:
 
 
 @dataclass(frozen=True, kw_only=True)
-class SayVerb(OrderVerb, DirectiveVerb):
+class SayVerb(OrderVerb, DirectiveVerb, StaleOrderGate):
     help = "a non-order message; refused if it orders work"
     units: str = option("units", metavar="UNIT[,UNIT]")
     body_file: str = option("body_file", metavar="BODYFILE")
@@ -1010,7 +1045,7 @@ def run(self, ctx: Commander) -> None:
 
 
 @dataclass(frozen=True, kw_only=True)
-class ProblemVerb(OrderVerb, DirectiveVerb):
+class ProblemVerb(OrderVerb, DirectiveVerb, StaleOrderGate):
     help = "Core records the problem with its order, or nothing"
     order: str = option("order", metavar="ORDER.json")
     to: str = option("--to")

src/agent_comms/plan_phases.py

@@ -23,9 +23,23 @@
 class OrderPhase(DeclaredFamily, affix="Phase"):
     """What an order serves: one plan phase, or nothing in the plan."""
 
+    @classmethod
+    def order_refusals(cls, todos: Iterable[Todo], reason: str) -> tuple[str, ...]:
+        return ()
+
 
 class UnplannedPhase(OrderPhase):
-    pass
+    @classmethod
+    def order_refusals(cls, todos: Iterable[Todo], reason: str) -> tuple[str, ...]:
+        rows = tuple(todos)
+        ready = tuple(phase for phase in PlanPhase.members_with(PlanPhase) if phase.ready(rows))
+        if ready and len(reason.split()) < 8:
+            return (
+                "Plan phases are ready: "
+                + ", ".join(phase.declared_name for phase in ready)
+                + ". Name a phase, or provide an explicit unplanned_reason of 8+ words.",
+            )
+        return ()
 
 
 class PlanPhase(OrderPhase, declared_name=None):

src/agent_comms/repo.py

@@ -17,10 +17,10 @@ class Repo(BaseModel):
     github: str
     source: str = "src"
 
-    def contains(self, sha: str) -> bool:
-        """Only a successful ancestry proof in this installed checkout permits closure."""
+    def contains(self, sha: str, revision: str = "HEAD") -> bool:
+        """Prove ancestry in the requested fetched revision without moving the checkout."""
         result = subprocess.run(
-            ("git", "-C", str(self.checkout), "merge-base", "--is-ancestor", sha, "HEAD"),
+            ("git", "-C", str(self.checkout), "merge-base", "--is-ancestor", sha, revision),
             capture_output=True,
             timeout=60,
         )

src/agent_comms/todos.py

@@ -110,6 +110,10 @@ async def checked(self, check: OrderCheck) -> ChildResult: ...
     @abstractmethod
     async def delivery_time(self, check): ...
 
+    def merged_commits(self, repo: Repo) -> tuple[str, ...]:
+        """Operations have no commit whose ancestry could prove a merge."""
+        return ()
+
     def strikes(
         self, todo: Todo, order_id: str, actor: ThreadIncarnation
     ) -> tuple[StrikeTodo, ...]:
@@ -137,6 +141,9 @@ def delivered(self) -> str:
     def work(self) -> str:
         return f"delivered commit {self.delivered}"
 
+    def merged_commits(self, repo: Repo) -> tuple[str, ...]:
+        return (self.delivered,) if repo.contains(self.delivered, "origin/main") else ()
+
     async def require_source(self, check: OrderCheck) -> None:
         raise ValueError("A commit delivery names the unit's delivered commit and the checked one.")
 
@@ -1477,6 +1484,10 @@ def copy_as[Kind: ProblemOrder](self, kind: type[Kind], **evidence) -> Kind:
     def merged(self, sha: str) -> ProblemOrder:
         return self.copy_as(MergedProblemOrder, sha=sha)
 
+    def observed_merges(self, repo: Repo, source: ProgressSource) -> tuple[ProblemOrder, ...]:
+        return tuple(self.merged(merged.sha) for merged in MergedProgress.found(source)
+                     if repo.contains(merged.sha))
+
     @property
     def pull_request_reference(self) -> str:
         return f"{self.pull_request_label} {self.id}"
@@ -1620,6 +1631,11 @@ def delivered(self, delivery):
             raise ValueError("The order already records a different delivery.")
         return self
 
+    def observed_merges(self, repo: Repo, source: ProgressSource) -> tuple[ProblemOrder, ...]:
+        commits = self.delivery.merged_commits(repo)
+        return (tuple(self.merged(commit) for commit in commits)
+                or super().observed_merges(repo, source))
+
     def closing_artifact(self) -> str:
         return self.delivery.evidence.artifact()
 
@@ -1682,6 +1698,9 @@ class WithdrawnProblemOrder(CheckedProblemOrder):
     pending: ClassVar[bool] = False
     line: ClassVar[str] = "Withdrawn"
 
+    def require_close(self) -> None:
+        """Retiring an erroneous order does not claim installed work."""
+
     def withdrawals(self, todo: Todo) -> tuple[str, ...]:
         superseded = f" (superseded by {self.superseded_by})" if self.superseded_by else ""
         return (f"{todo.id}: order {self.id} withdrawn from {self.previous.owner}, "
@@ -2665,12 +2684,11 @@ def reconcile_merges(self, commander: ThreadIncarnation, repos: Sequence[Repo],
         checked = tuple((todo, order) for todo in self.held_for_commander(commander)
                         if isinstance(order := todo.order, CheckedProblemOrder))
         pulls = PullRequests(tuple(todo for todo, _ in checked))
-        completed = tuple(DoneTodoState.closed(todo, order.merged(merged.sha))
+        completed = tuple(DoneTodoState.closed(todo, merged)
                           for todo, order in checked
                           for repo in repos if repo.github.lower() == todo.repo
-                          for merged in MergedProgress.found(
-                              ProgressSource.of(todo, storage, pulls))
-                          if repo.contains(merged.sha))
+                          for merged in order.observed_merges(
+                              repo, ProgressSource.of(todo, storage, pulls)))
         if not completed:
             return
         with self.writing() as db:

tests/test_commander_problem_orders.py

@@ -544,6 +544,7 @@ def order_body(repo, check=None, edit='Set count.txt to 1.', extra=''):
     body.write_text(json.dumps({
         'defect': 'The counter file holds zero where the fleet needs one',
         'why_stupid': 'The counter is wrong and that is stupid.',
+        'unplanned_reason': 'This isolated order checks command behavior rather than advancing the plan dependency chain.',
         'order': {'kind': 'pending', 'owner': f'{__file__}:1',
                   'check': FieldCodec.encode(checked), 'edits': [edit],
                   'run': {'kind': 'application', 'text': 'the installed app shows the counter at one after restart'},

tests/test_stale_order_gates.py

@@ -0,0 +1,158 @@
+"""Real TodoStore rows exercise the gate family and merged-delivery observer."""
+import json
+import os
+import subprocess
+import time
+from dataclasses import replace
+from pathlib import Path
+
+import pytest
+
+from agent_comms.commander.order import OrderFile, OrderVerb, StaleOrderGate
+from agent_comms.plan_phases import PlanPhase, RecordPlanePhase, UnplannedPhase
+from agent_comms.recorded_time import RecordedEventTime
+from agent_comms.repo import Repo
+from agent_comms.todos import (
+    AssignTodo, CheckedProblemOrder, CommitDelivery, OperationDelivery, WithdrawTodo,
+)
+from agent_comms.tools import CommsOrderDoneTool
+from test_commander_problem_orders import create, order_body, working
+
+
+def test_stale_gate_covers_its_verb_family_but_allows_board_repairs(
+        commander, commander_script, tmp_path):
+    comms, _ = commander
+    context, _ = commander_script
+    todo = create(commander, 'P244', deadline=time.time() - 1900)
+    body = str(tmp_path / 'must-not-be-read')
+    blocked = (
+        ['send', 'worker', '23:59', body], ['say', 'worker', body],
+        ['problem', body, '--to', 'worker', '--due', '23:59', '--where', f'{__file__}:1'],
+        ['strike', 'worker', '--why', 'P244 missed its deadline'],
+    )
+    verbs = tuple(OrderVerb.parse(args, '') for args in blocked)
+    assert {type(verb) for verb in verbs} == set(OrderVerb.members_with(StaleOrderGate))
+    for verb in verbs:
+        with pytest.raises(SystemExit) as refused:
+            verb.apply(context)
+        text = str(refused.value)
+        assert 'P244: holder worker, deadline ' in text
+        assert 'more than 30 minutes overdue' in text
+    why = 'The current source needs one more bounded check before delivery.'
+    repairs = (
+        ['due', 'P244', '23:59', '--why', why], ['withdraw', 'P244', '--why', why],
+        ['done', 'worker', 'order-P244', '--sha', 'a' * 40],
+        ['close', 'P244', '--works', why],
+        ['check', 'P244', '--check', f'{tmp_path} :: printf 0 -> 0', '--why', why],
+        ['serve', 'P244', 'record_plane'], ['drop', 'worker', 'old', '--by', 'new'],
+    )
+    for args in repairs:
+        OrderVerb.parse(args, '').gate(context)
+    actor = comms.registry.require('lead').incarnation
+    comms.goals.todos.apply(todo.id, AssignTodo(
+        todo.revision, todo.held(), RecordedEventTime(time.time() + 3600), todo.order, todo.goal))
+    verbs[0].gate(context)
+    moved = comms.goals.todos.get(todo.id)
+    comms.goals.todos.apply(todo.id, WithdrawTodo(
+        moved.revision, actor, moved.held(), why, None))
+    verbs[0].gate(context)
+    assert not comms.goals.todos.get(todo.id).state.open
+
+
+@pytest.mark.parametrize('phase', (UnplannedPhase, *PlanPhase.members_with(PlanPhase)))
+def test_phase_gate_uses_the_phase_family_and_requires_an_explicit_unplanned_reason(
+        commander, commander_script, phase):
+    comms, _ = commander
+    context, repo = commander_script
+    (repo / 'count.txt').write_text('0\n')
+    payload = json.loads(Path(order_body(repo)).read_text())
+    payload['order']['serves'] = phase.declared_name
+    if phase is UnplannedPhase:
+        del payload['unplanned_reason']
+        with pytest.raises(SystemExit, match='Plan phases are ready'):
+            context.send('worker', '23:59', OrderFile.read(json.dumps(payload)),
+                         clean=True, new=('P1', 'Repair the owner'))
+        assert comms.goals.todos.for_commander(comms.registry.require('lead').incarnation) == ()
+        payload['unplanned_reason'] = (
+            'This urgent retained incident is outside the declared plan dependency chain.')
+    context.send('worker', '23:59', OrderFile.read(json.dumps(payload)),
+                 clean=True, new=('P1', 'Repair the owner'))
+    assert comms.goals.todos.get('P1').order.serves is phase
+
+
+def test_second_plan_order_gate_names_the_busy_order_and_running_unowned_units(
+        commander, commander_script):
+    comms, _ = commander
+    context, repo = commander_script
+    todo = create(commander, 'P1')
+    with comms.goals.todos.writing() as db:
+        replace(todo, order=replace(todo.order, serves=RecordPlanePhase)).upsert(db)
+    comms.registry.declare(comms.registry.new_thread('free-unit', frozenset(), str(repo)))
+    comms.owners.acquire_thread('free-unit', owner_pid=os.getpid())
+    (repo / 'count.txt').write_text('0\n')
+    payload = json.loads(Path(order_body(repo)).read_text())
+    payload['order']['serves'] = RecordPlanePhase.declared_name
+    with pytest.raises(SystemExit) as refused:
+        context.send('worker', '23:59', OrderFile.read(json.dumps(payload)),
+                     clean=True, new=('P2', 'Repair another owner'))
+    assert 'worker already holds plan orders P1' in str(refused.value)
+    assert 'running units holding no order: free-unit' in str(refused.value)
+    assert [todo.id for todo in comms.goals.todos.for_commander(
+        comms.registry.require('lead').incarnation)] == ['P1']
+
+
+def test_pr_uses_actual_owned_checkout_instead_of_an_invented_problem_directory(
+        commander, commander_script, monkeypatch, tmp_path):
+    comms, _ = commander
+    context, repo = commander_script
+    todo = create(commander, 'P287')
+    body = tmp_path / 'pull-request.md'
+    body.write_text('The retained order owns its immutable pull request identity.')
+    seen = []
+    def publish(order, current, checkout, title, text):
+        seen.append((current.id, checkout, order.pull_request_body(text)))
+        return 'https://github.com/OpenHCSDev/agent-comms/pull/1'
+    monkeypatch.setattr(CheckedProblemOrder, 'open_pull_request', publish)
+    monkeypatch.chdir(repo)
+    OrderVerb.parse(['pr', 'P287', '--title', 'P287', '--body-file', str(body)], '').apply(context)
+    assert seen == [('P287', repo, todo.order.pull_request_body(body.read_text()))]
+    subprocess.run(['git', '-C', str(repo), 'remote', 'set-url', 'origin',
+                    'https://github.com/another/project.git'], check=True)
+    with pytest.raises(SystemExit, match=f'not {todo.repo}'):
+        OrderVerb.parse(['pr', 'P287', '--title', 'P287', '--body-file', str(body)], '').apply(context)
+    assert len(seen) == 1  # A different repository must never reach publication.
+
+
+@pytest.mark.parametrize('commit_delivery', (True, False))
+def test_observer_closes_only_delivered_commits_present_in_fetched_main(
+        commander, tmp_path, commit_delivery):
+    comms, caller = commander
+    root = comms.root
+    origin = tmp_path / 'origin.git'
+    subprocess.run(['git', 'init', '-q', '--bare', str(origin)], check=True)
+    def git(*args):
+        return subprocess.check_output(['git', '-C', str(root), '-c', 'user.name=t',
+            '-c', 'user.email=t@t', *args], text=True).strip()
+    git('init', '-q', '-b', 'main')
+    git('remote', 'add', 'origin', str(origin))
+    git('commit', '-q', '--allow-empty', '-m', 'delivered source')
+    sha = git('rev-parse', 'HEAD')
+    git('push', '-q', 'origin', 'HEAD:main')
+    todo = create(commander, 'P1', deadline=time.time() + 3600)
+    with comms.goals.todos.writing() as db:
+        replace(todo, order=replace(todo.order, serves=RecordPlanePhase)).upsert(db)
+    delivery = (CommitDelivery(commit=sha, delivered_commit=sha) if commit_delivery
+                else OperationDelivery())
+    CommsOrderDoneTool(problem='P1', order_id=todo.order.id, delivery=delivery,
+                      claims=working(comms)).apply(comms, caller)
+    assert comms.goals.todos.get('P1').state.open
+    repo = Repo(checkout=root, github=todo.repo)
+    actor = comms.registry.require('lead').incarnation
+    comms.goals.todos.reconcile_merges(actor, (repo,),
+                                    comms.registry.storage_rule)
+    observed = comms.goals.todos.get('P1')
+    assert observed.state.open is not commit_delivery
+    assert observed.order.serves is RecordPlanePhase
+    comms.goals.todos.reconcile_merges(actor, (repo,),
+                                    comms.registry.storage_rule)
+    assert comms.goals.todos.get('P1') == observed