diff --git a/src/agent_comms/agent_activity.py b/src/agent_comms/agent_activity.py index afcc8c75a1..79a93089fe 100644 --- a/src/agent_comms/agent_activity.py +++ b/src/agent_comms/agent_activity.py @@ -4,7 +4,7 @@ import logging import time -from collections.abc import Iterable, Mapping +from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, replace from functools import partial from pathlib import Path @@ -261,32 +261,27 @@ def begin_turn( try: phase = PreparingPhase(detail) lease.record.transition(lease, phase) - self._record_turn_activity(leased.name, (phase,)) + self._emit_activity(phase.activity_event(leased.name)) except BaseException: self.registry.release_turn(lease) raise return RegistryOwner(thread=leased, claimed_generation=lease.claimed_generation) - def _record_turn_activity(self, name: str, phases: Iterable[TurnPhase]) -> None: - """Changed phases are real activity events, just like CLI status reports.""" - for phase in phases: - self._emit_activity(phase.activity_event(name)) + def _turn_phase_effects( + self, lease: TurnLeaseFence, phase: TurnPhase, + transition: Callable[[TurnLeaseFence, TurnPhase], tuple[TurnState, ...]], + ) -> tuple[TurnState, ...]: + with _store_lock(self._wire_lock_path, mode=SharedLock()): + effects = transition(lease, phase) + for state in effects: + state.active.phase.phase_effects(self, lease) + return effects def transition_turn(self, lease: TurnLeaseFence, phase: TurnPhase) -> tuple[TurnState, ...]: - with _store_lock(self._wire_lock_path, mode=SharedLock()): - effects = lease.record.transition(lease, phase) - self._record_turn_activity( - lease.identity.incarnation.name, (state.active.phase for state in effects) - ) - return effects + return self._turn_phase_effects(lease, phase, lease.record.transition) def observe_native_phase(self, lease: TurnLeaseFence, phase: TurnPhase) -> tuple[TurnState, ...]: - with _store_lock(self._wire_lock_path, mode=SharedLock()): - effects = lease.record.observe_native_phase(lease, phase) - self._record_turn_activity( - lease.identity.incarnation.name, (state.active.phase for state in effects) - ) - return effects + return self._turn_phase_effects(lease, phase, lease.record.observe_native_phase) def finish_turn(self, lease: TurnLeaseFence) -> TurnCompletion: """Persist this lease's terminal identity before publishing idle activity.""" diff --git a/src/agent_comms/commander/config.py b/src/agent_comms/commander/config.py index c4c11c4e7f..df6c020045 100644 --- a/src/agent_comms/commander/config.py +++ b/src/agent_comms/commander/config.py @@ -17,6 +17,7 @@ from ..errors import RelationViolationError from ..native_arguments import NativeArguments from ..threads import Thread +from ..todos import Todo class CommanderRole: @@ -40,6 +41,15 @@ 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.""" + result = subprocess.run( + ('git', '-C', str(self.checkout), 'merge-base', '--is-ancestor', sha, 'HEAD'), + capture_output=True, timeout=60) + if result.returncode not in (0, 1): + result.check_returncode() + return result.returncode == 0 + @classmethod def of(cls, checkout: Path) -> Repo: """Its GitHub name comes from the checkout's own origin remote.""" @@ -109,6 +119,12 @@ class Commander: def of(cls, config: CommanderConfig) -> Commander: return cls(config.commander, Comms(config.root)) + def orders(self, repos: tuple[Repo, ...]) -> tuple[Todo, ...]: + actor = self.comms.registry.require(self.name).incarnation + store = self.comms.goals.todos + store.reconcile_merges(actor, repos, self.comms.registry.storage_rule) + return store.for_commander(actor) + def units(self) -> frozenset[str]: """The threads this commander gives orders to; the owner, claude-ai and peer commanders are not among them.""" diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 83e875478f..3a04251f09 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -397,7 +397,7 @@ def board(): '| Problem | Owner file | Owner | Order | Due | Progress | Delivered ' '| Owner last acted |', '|---|---|---|---|---|---|---|---|'] - listed, read = todos(), progress() + listed, read = COMMANDER.orders(CONFIG.repos), progress() activity = {unit: last_activity(unit) for unit in {t.holder for t in listed if t.holder}} for t in sorted(listed, key=lambda t: int(t.id[1:]) if t.id[1:].isdigit() else 0): unit, last = t.holder, t.released[-1:] @@ -589,7 +589,7 @@ def replied(latest: Message | None) -> str: def deadline_lines(pending: list[Todo], read: dict[str, OrderProgress], since: float = 0.0) -> list[str]: - """Every past-due unrecorded order alarms with its state until recorded. + """Past-due unfinished orders alarm; installed merges have already closed. Stall crossings remain separate from deadline alarms.""" now = time.time() late, stalled = [], [] @@ -602,8 +602,7 @@ def deadline_lines(pending: list[Todo], read: dict[str, OrderProgress], since: f answer = reply(t.holder, ordered) due = f'due {hhmm(t.deadline)}, order {t.order.id}' late.append(f'{t.holder} ({due}, {work.line}{replied(answer)})') - alarm = (f'OVERDUE: {", ".join(late)} -> record merged work; ' - 'push or merge unfinished work now') + alarm = f'OVERDUE: {", ".join(late)} -> push or merge unfinished work now' return [alarm] * bool(late) + stalled @@ -618,7 +617,7 @@ def overdue(): installs: list[str] = [] while True: now = time.time() - held = held_orders() + held = tuple(todo for todo in COMMANDER.orders(CONFIG.repos) if todo.assignment) activity = {unit: last_activity(unit) for unit in {t.assignment.owner for t in held}} if missing := first_calls(held, activity): print('NO FIRST CALL: ' + ', '.join(missing) + ' -> stuck holding an order; push it now or reassign', diff --git a/src/agent_comms/pi_payloads.py b/src/agent_comms/pi_payloads.py index a252d61a68..a29d556e01 100644 --- a/src/agent_comms/pi_payloads.py +++ b/src/agent_comms/pi_payloads.py @@ -18,6 +18,7 @@ from .declared_family import DeclaredFamily, UnknownFamilyMember from .field_codec import FieldCodec, Id128Text, MalformedRecord, TextRepresentation from .native_file_artifact import NativeFileArtifact +from .turn_failure import ProviderTransportFailed from .native_session_reopen import NativeSessionIdentity from .pi_counts import NonnegativePiCount, PiCount, UnreportedCount from .pi_vocabulary import ( @@ -33,6 +34,7 @@ if TYPE_CHECKING: from .agent_events import ProviderUsage + from .turn_phase import TurnPhase from .turn_usage import UsageAccount @@ -504,6 +506,9 @@ def wire_member(cls, value): def refine_model_request_failure(self, failure): return failure + def turn_phase(self, phase: TurnPhase) -> TurnPhase: + return phase + @property def description(self) -> str: return "" @@ -516,9 +521,11 @@ class ProviderTransportFailureDiagnostic(PiDiagnostic): details: ProviderTransportDetails timestamp: int | None = None - def refine_model_request_failure(self, failure): - from .turn_failure import ProviderTransportFailed + def turn_phase(self, phase: TurnPhase) -> TurnPhase: + from .turn_phase import FailedPhase + return FailedPhase(error=self.error) + def refine_model_request_failure(self, failure): return ProviderTransportFailed(failure.text) @property diff --git a/src/agent_comms/pi_vocabulary.py b/src/agent_comms/pi_vocabulary.py index 020ac4dfa4..c219276dcd 100644 --- a/src/agent_comms/pi_vocabulary.py +++ b/src/agent_comms/pi_vocabulary.py @@ -152,6 +152,8 @@ async def apply(cls, session, message): return from .acp_failure import NativeMessageFailureSource + for diagnostic in message.diagnostics: + session.watchdog.phase = diagnostic.turn_phase(session.watchdog.phase) error = session.output.error( (message.error_message or "").strip() or f"Model request {cls.declared_name}", message.diagnostics, source=NativeMessageFailureSource(message), diff --git a/src/agent_comms/standing_rules.py b/src/agent_comms/standing_rules.py index 51efb755a6..3e0c9829b8 100644 --- a/src/agent_comms/standing_rules.py +++ b/src/agent_comms/standing_rules.py @@ -196,7 +196,7 @@ def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: def line(self) -> str: return ( f"BOARD DISAGREES: {self.problem} order {self.order} reads no commit, " - f"but its linked PR merged as {self.merge} -> repair the progress read this wake" + f"but its linked PR merged as {self.merge[:9]} -> repair the progress read this wake" ) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 281e53ab5d..19c45babcb 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -45,6 +45,7 @@ from .plan_phases import OrderPhase, UnplannedPhase if TYPE_CHECKING: + from .commander.config import Repo from .comms import Comms from .goal_management import Goals @@ -590,18 +591,18 @@ def untouched(self, now: float) -> bool: @dataclass(frozen=True) class MergedProgress(FoundProgress): - """A merge awaits its order record; its alarm does not itself draw a strike.""" + """A linked merge closes its order once the installed checkout contains it.""" precedence: ClassVar[int] = 1 sha: str @classmethod def found(cls, source: ProgressSource) -> Iterator[Self]: - return (cls(pr.merge[:9]) for pr in source.requests("merged") + return (cls(pr.merge) for pr in source.requests("merged") if pr.merge and (pr.merged_at or 0) >= source.since) @property def state(self) -> str: - return f"merged as {self.sha}, awaiting record" + return f"merged as {self.sha[:9]}, waiting on install" def waiting_on_install(self, todo: Todo) -> tuple[Todo, ...]: return (todo,) @@ -1238,6 +1239,9 @@ def changed(name: str) -> bool: def copy_as[Kind: ProblemOrder](self, kind: type[Kind], **evidence) -> Kind: return kind(**{item.name: getattr(self, item.name) for item in fields(self)}, **evidence) + def merged(self, sha: str) -> ProblemOrder: + return MergedProblemOrder(original=self, sha=sha) + @property def pull_request_reference(self) -> str: return f"{self.pull_request_label} {self.id}" @@ -1310,6 +1314,21 @@ def waiting_on_install(self, todo: Todo, read: dict[str, OrderProgress]) -> tupl return () +@dataclass(frozen=True, kw_only=True) +class MergedProblemOrder(ProblemOrder): + """An installed merge, retaining its original order without inventing a check or app proof.""" + original: CheckedProblemOrder + sha: str + line: ClassVar[str] = "Merged and installed" + + @property + def id(self) -> str: + return self.original.id + + def require_close(self) -> None: + pass + + @dataclass(frozen=True, kw_only=True) class PendingProblemOrder(CheckedProblemOrder): def reissued(self, at: float) -> PendingProblemOrder: @@ -1626,7 +1645,13 @@ def change(cls, current, command): @classmethod def decide(cls, current, command): current.require_commander(command.actor) - return replace(current, state=cls, assignment=None, revision=current.revision + 1) + return cls.closed(current, current.order) + + @classmethod + def closed(cls, current: Todo, order: ProblemOrder) -> Todo: + order.require_close() + return replace(current, state=cls, order=order, assignment=None, + revision=current.revision + 1) @dataclass(frozen=True) @@ -2297,6 +2322,26 @@ def for_goal(self, owner: ThreadIncarnation, goal_id: str) -> tuple[Todo, ...]: "json_extract(assignment,'$.owner')=? AND json_extract(assignment,'$.owner_created')=? " "AND json_extract(assignment,'$.generation')=?", (owner.name, owner.created_at, goal_id)) + def reconcile_merges(self, commander: ThreadIncarnation, repos: Sequence[Repo], + storage: StorageRule) -> None: + """Read before locking; close only orders whose observed revision still holds.""" + 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)) + 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)) + if not completed: + return + with self.writing() as db: + for result in completed: + current = self._current(db, result.id) + if current.revision + 1 == result.revision: + result.upsert(db) + def board_disagreements( self, commander: ThreadIncarnation, storage: StorageRule, ) -> Iterator[tuple[Todo, MergedProgress]]: diff --git a/src/agent_comms/turn_phase.py b/src/agent_comms/turn_phase.py index ffe7df4c92..d86a319f04 100644 --- a/src/agent_comms/turn_phase.py +++ b/src/agent_comms/turn_phase.py @@ -7,8 +7,11 @@ from .declared_family import DeclaredFamily from .activity import Activity, ActivityState, ObservedActivity +from .pi_payloads import ProviderDiagnosticError if TYPE_CHECKING: from . import pi_events as pi + from .agent_activity import AgentActivity + from .turn_lease import TurnLeaseFence class StallExempt: @@ -121,6 +124,11 @@ def response_output(self) -> TurnPhase: def stalled(self, accepted: bool) -> TurnPhase: return self if accepted else PromptAcceptancePhase() + def phase_effects(self, agents: AgentActivity, lease: TurnLeaseFence) -> None: + agents.set_activity( + lease.identity.incarnation.name, self.activity_state, self.activity_detail, + ) + class Excursion(TurnPhase, declared_name=None): pauses_input_clock = True @@ -146,6 +154,14 @@ class IdlePhase(ActivityWord, TurnPhase): activity_state = ActivityState.IDLE label = "Ready" + def phase_effects(self, agents: AgentActivity, lease: TurnLeaseFence) -> None: + agents.finish_turn(lease) + + +@dataclass(frozen=True, kw_only=True) +class FailedPhase(IdlePhase): + error: ProviderDiagnosticError + class CancellingPhase(TurnPhase): accepts_followup = False diff --git a/tests/test_installed_merge.py b/tests/test_installed_merge.py new file mode 100644 index 0000000000..53399af943 --- /dev/null +++ b/tests/test_installed_merge.py @@ -0,0 +1,125 @@ +"""Merge observation closes the real stored order, without a completion request.""" +from dataclasses import replace +import subprocess + +import pytest + +from agent_comms.commander.config import Repo +from agent_comms.field_codec import FieldCodec +from agent_comms.todos import (DoneTodoState, MergedProblemOrder, OperationDelivery, + PullRequest, PullRequests, UnorderedProblemOrder) +from agent_comms.tools import CommsOrderDoneTool +from test_commander_problem_orders import working +from test_order_progress import checkout, commit, deadline_lines, git + + +@pytest.mark.parametrize('delivered', (False, True)) +def test_installed_merge_closes_pending_or_delivered_order_once( + commander, commander_script, checkout, tmp_path, monkeypatch, capsys, delivered): + comms, caller = commander + script, _ = commander_script + todo, unit = checkout + if delivered: + CommsOrderDoneTool(problem=todo.id, order_id=todo.order.id, + delivery=OperationDelivery(), claims=working(comms)).apply(comms, caller) + todo = comms.goals.todos.get(todo.id) + original = todo.order + commit(unit) + git(unit, 'switch', '-q', 'main') + git(unit, '-c', 'user.name=t', '-c', 'user.email=t@t', 'merge', '--no-ff', + '-m', 'merge the order', 'fix/p1') + sha = git(unit, 'rev-parse', 'HEAD') + base = git(unit, 'rev-parse', 'HEAD^1') + installed = tmp_path / 'installed' + installed.mkdir() + git(installed, 'init', '-q', '-b', 'main') + git(installed, 'remote', 'add', 'origin', str(unit)) + 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) + merge = PullRequest(1, 'fix/p1', sha, sha, 500, 500, (original.id,)) + monkeypatch.setattr(PullRequests, 'read', lambda self, repo, state, order: + (merge,) if state == 'merged' else ()) + + # 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,) + 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) == () + assert script.held_orders() == () + closed = comms.goals.todos.get(todo.id) + assert closed.state is DoneTodoState and closed.assignment is None + assert closed.revision == todo.revision + 1 and not closed.pending + assert isinstance(closed.order, MergedProblemOrder) + assert closed.order.sha == sha and closed.order.original == original + assert closed.order.id == original.id + assert closed.strikes == todo.strikes + assert FieldCodec.decode(type(closed), FieldCodec.encode(closed)) == closed + + def no_child(*args, **kwargs): + pytest.fail('A repeated closed-order read launched a subprocess') + + with monkeypatch.context() as repeat: + repeat.setattr(subprocess, 'run', no_child) + assert script.COMMANDER.orders(script.CONFIG.repos) == () + assert comms.goals.todos.get(todo.id) == closed + script.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 + + class TickComplete(Exception): + pass + + def next_tick(seconds): + raise TickComplete + + monkeypatch.setattr(script.time, 'sleep', next_tick) + with pytest.raises(TickComplete): + script.overdue() + assert 'OVERDUE:' not in capsys.readouterr().out + + +def test_new_revision_is_not_closed_from_an_old_merge_observation( + commander, commander_script, checkout, monkeypatch): + comms, _ = commander + script, installed = commander_script + todo, _ = checkout + repo = Repo(checkout=installed, github=todo.repo) + merge = PullRequest(1, 'fix/p1', 'a' * 40, 'b' * 40, 500, 500, (todo.order.id,)) + monkeypatch.setattr(PullRequests, 'read', lambda self, repo, state, order: (merge,)) + newer = replace(todo, revision=todo.revision + 1, + order=replace(todo.order, id='replacement-order')) + + def observed(self, sha): + with comms.goals.todos.writing() as db: + newer.upsert(db) + return True + + monkeypatch.setattr(Repo, 'contains', observed) + comms.goals.todos.reconcile_merges(comms.registry.require('lead').incarnation, + (repo,), comms.registry.storage_rule) + assert comms.goals.todos.get(todo.id) == newer + + +def test_only_checked_orders_are_selected_for_merge_reconciliation(commander, checkout): + comms, _ = commander + todo, _ = checkout + unordered = replace(todo, order=UnorderedProblemOrder()) + with comms.goals.todos.writing() as db: + unordered.upsert(db) + comms.goals.todos.reconcile_merges(comms.registry.require('lead').incarnation, + (), comms.registry.storage_rule) + assert comms.goals.todos.get(todo.id) == unordered + + +def test_failed_git_is_not_a_negative_ancestry_result(tmp_path): + with pytest.raises(subprocess.CalledProcessError): + Repo(checkout=tmp_path / 'missing', github='test/repo').contains('a' * 40) diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index b676797076..96cbed66d2 100644 --- a/tests/test_order_progress.py +++ b/tests/test_order_progress.py @@ -155,10 +155,10 @@ def test_a_pr_merged_before_the_order_is_not_its_merge(checkout): assert read(checkout, pulls) == NoCommitProgress() -def test_a_merged_order_still_awaits_its_record(checkout): +def test_a_merge_not_yet_installed_waits_only_on_install(checkout): progress = read(checkout, Listed(merged=(PullRequest(8, 'fix/p1', 'b' * 40, 'd' * 40, 500.0, 500.0, orders=('order-P1',)),))) - assert progress == MergedProgress('d' * 9) - assert progress.line == 'merged as ddddddddd, awaiting record' + assert progress == MergedProgress('d' * 40) + assert progress.line == 'merged as ddddddddd, waiting on install' def test_unread_github_is_explicit(checkout): @@ -168,7 +168,7 @@ def test_unread_github_is_explicit(checkout): assert progress == UnreadProgress('gh: not logged in') -def test_board_and_overdue_name_a_merge_awaiting_its_record(commander, commander_script, monkeypatch, tmp_path): +def test_board_and_overdue_name_a_merge_waiting_on_install(commander, commander_script, monkeypatch, tmp_path): script, _ = commander_script # worker's P1 checkout, on the branch its merged PR came from. monkeypatch.setattr(script.COMMS.registry, 'storage_rule', StorageRule(checkout_base=tmp_path / 'wt')) @@ -178,9 +178,9 @@ def test_board_and_overdue_name_a_merge_awaiting_its_record(commander, commander create(commander, 'P1', deadline=1) merged = (PullRequest(8, 'fix/p1', 'b' * 40, 'd' * 40, 500.0, 500.0, orders=('order-P1',)),) monkeypatch.setattr(PullRequests, 'read', lambda self, repo, state, order: merged if state == 'merged' else ()) - assert "awaiting record" in deadline_lines(script)[0] + assert "waiting on install" in deadline_lines(script)[0] script.board() - assert '| merged as ddddddddd, awaiting record |' in (tmp_path / 'board.md').read_text() + assert '| merged as ddddddddd, waiting on install |' in (tmp_path / 'board.md').read_text() def test_order_due_moves_the_deadline_in_core_and_on_the_record(commander, commander_script, monkeypatch, tmp_path): @@ -229,7 +229,7 @@ def requests(self, state): (CommittedProgress, f'committed aaaaaaaaa at {MOVED}, not pushed', 1900, None), (PushedProgress, f'pushed eeeeeeeee at {MOVED}, no PR', 1900, None), (OpenPrProgress, 'PR #7 open at bbbbbbbbb', 1900, 1300), - (MergedProgress, 'merged as ccccccccc, awaiting record', None, None), + (MergedProgress, 'merged as ccccccccc, waiting on install', None, None), ]) def test_each_found_member_says_from_its_class_whether_it_stalls_and_is_untouched( member, line, stalls_at, untouched_at): @@ -333,7 +333,7 @@ def test_merged_and_delivered_orders_wait_on_the_one_install_deadline(commander, assert script.install_lines(held, read, 2001.0) == [ f'INSTALL LATE (due {due}), not the orders; waiting: P1 worker, P2 worker'] assert script.install_lines(held[2:], read, 2001.0) == [] - assert "awaiting record" in deadline_lines(script)[0] + assert "waiting on install" in deadline_lines(script)[0] @pytest.mark.parametrize("pid,branch,number,subject", [ @@ -368,9 +368,9 @@ def forge(command, **arguments): (foreign, merged) if state == "merged" else ()) assert not storage.order_checkout("worker", pid).exists() work = OrderProgress.read(todo, storage, PullRequests()) - assert work == MergedProgress("d" * 9) + assert work == MergedProgress("d" * 40) script.board() - assert "merged as ddddddddd, awaiting record" in (tmp_path / "board.md").read_text() + assert "merged as ddddddddd, waiting on install" in (tmp_path / "board.md").read_text() for state in (work, PushedProgress("b" * 9, at=500.0), NoCommitProgress()): alarms = script.deadline_lines([todo], {pid: state}, time.time() + 100) assert alarms[0].startswith("OVERDUE:") and state.line in alarms[0]