diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 33e6ee33de..3ed66ffbe0 100644 --- a/src/agent_comms/commander/order.py +++ b/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 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") diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 04865c669b..50676f12ab 100644 --- a/src/agent_comms/plan_phases.py +++ b/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): diff --git a/src/agent_comms/repo.py b/src/agent_comms/repo.py index b6cfe20dbb..79df05b074 100644 --- a/src/agent_comms/repo.py +++ b/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, ) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 896e0be57f..13da3ca8d3 100644 --- a/src/agent_comms/todos.py +++ b/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: diff --git a/tests/test_commander_problem_orders.py b/tests/test_commander_problem_orders.py index 2bc690a82d..d89775727e 100644 --- a/tests/test_commander_problem_orders.py +++ b/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'}, diff --git a/tests/test_stale_order_gates.py b/tests/test_stale_order_gates.py new file mode 100644 index 0000000000..0e053ea2a1 --- /dev/null +++ b/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