From 246f9cf5819d7ce2b2f263f708576550beef8488 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:05:01 -0400 Subject: [PATCH 01/10] P689 let failed phases release the existing turn lease --- src/agent_comms/agent_activity.py | 28 +++++++++++----------------- src/agent_comms/pi_payloads.py | 6 ++++++ src/agent_comms/pi_vocabulary.py | 2 ++ src/agent_comms/turn_phase.py | 19 +++++++++++++++++++ 4 files changed, 38 insertions(+), 17 deletions(-) diff --git a/src/agent_comms/agent_activity.py b/src/agent_comms/agent_activity.py index afcc8c75a1..9c927e9fd7 100644 --- a/src/agent_comms/agent_activity.py +++ b/src/agent_comms/agent_activity.py @@ -261,32 +261,26 @@ 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, phase, transition) -> tuple[TurnState, ...]: + with _store_lock(self._wire_lock_path, mode=SharedLock()): + effects = transition(lease, phase) + for state in effects: + current = state.active.phase + current.phase_effects(self, lease) + self._emit_activity(current.activity_event(lease.identity.incarnation.name)) + 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/pi_payloads.py b/src/agent_comms/pi_payloads.py index 5f7bdbc6f6..c5a6acd631 100644 --- a/src/agent_comms/pi_payloads.py +++ b/src/agent_comms/pi_payloads.py @@ -490,6 +490,9 @@ def wire_member(cls, value): def refine_model_request_failure(self, failure): return failure + def turn_phase(self, phase): + return phase + @property def description(self) -> str: return "" @@ -502,6 +505,9 @@ class ProviderTransportFailureDiagnostic(PiDiagnostic): details: ProviderTransportDetails timestamp: int | None = None + def turn_phase(self, phase): + return phase.failed(self.error) + def refine_model_request_failure(self, failure): from .turn_failure import ProviderTransportFailed 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/turn_phase.py b/src/agent_comms/turn_phase.py index ffe7df4c92..2dbe5e38e7 100644 --- a/src/agent_comms/turn_phase.py +++ b/src/agent_comms/turn_phase.py @@ -7,6 +7,7 @@ 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 @@ -121,6 +122,12 @@ def response_output(self) -> TurnPhase: def stalled(self, accepted: bool) -> TurnPhase: return self if accepted else PromptAcceptancePhase() + def failed(self, error: ProviderDiagnosticError) -> TurnPhase: + return FailedPhase(error=error) + + def phase_effects(self, agents, lease) -> None: + pass + class Excursion(TurnPhase, declared_name=None): pauses_input_clock = True @@ -146,6 +153,18 @@ class IdlePhase(ActivityWord, TurnPhase): activity_state = ActivityState.IDLE label = "Ready" + def phase_effects(self, agents, lease) -> None: + agents.finish_turn(lease) + + +@dataclass(frozen=True, kw_only=True) +class FailedPhase(IdlePhase): + error: ProviderDiagnosticError + + @property + def activity_detail(self) -> str: + return self.error.message + class CancellingPhase(TurnPhase): accepts_followup = False From 40db287ed08f1f02ad12260641c110bb3bfa974c Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:07:21 -0400 Subject: [PATCH 02/10] P689 type phase completion and diagnostic transitions --- src/agent_comms/agent_activity.py | 7 +++++-- src/agent_comms/pi_payloads.py | 5 +++-- src/agent_comms/turn_phase.py | 6 ++++-- 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/src/agent_comms/agent_activity.py b/src/agent_comms/agent_activity.py index 9c927e9fd7..60becd18f3 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 @@ -267,7 +267,10 @@ def begin_turn( raise return RegistryOwner(thread=leased, claimed_generation=lease.claimed_generation) - def _turn_phase_effects(self, lease, phase, transition) -> tuple[TurnState, ...]: + 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: diff --git a/src/agent_comms/pi_payloads.py b/src/agent_comms/pi_payloads.py index c5a6acd631..cc133b6772 100644 --- a/src/agent_comms/pi_payloads.py +++ b/src/agent_comms/pi_payloads.py @@ -34,6 +34,7 @@ if TYPE_CHECKING: from .agent_events import ProviderUsage from .turn_usage import UsageAccount + from .turn_phase import TurnPhase def wire_field(name: str, default=None): @@ -490,7 +491,7 @@ def wire_member(cls, value): def refine_model_request_failure(self, failure): return failure - def turn_phase(self, phase): + def turn_phase(self, phase: TurnPhase) -> TurnPhase: return phase @property @@ -505,7 +506,7 @@ class ProviderTransportFailureDiagnostic(PiDiagnostic): details: ProviderTransportDetails timestamp: int | None = None - def turn_phase(self, phase): + def turn_phase(self, phase: TurnPhase) -> TurnPhase: return phase.failed(self.error) def refine_model_request_failure(self, failure): diff --git a/src/agent_comms/turn_phase.py b/src/agent_comms/turn_phase.py index 2dbe5e38e7..aadb3285b3 100644 --- a/src/agent_comms/turn_phase.py +++ b/src/agent_comms/turn_phase.py @@ -10,6 +10,8 @@ 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: @@ -125,7 +127,7 @@ def stalled(self, accepted: bool) -> TurnPhase: def failed(self, error: ProviderDiagnosticError) -> TurnPhase: return FailedPhase(error=error) - def phase_effects(self, agents, lease) -> None: + def phase_effects(self, agents: AgentActivity, lease: TurnLeaseFence) -> None: pass @@ -153,7 +155,7 @@ class IdlePhase(ActivityWord, TurnPhase): activity_state = ActivityState.IDLE label = "Ready" - def phase_effects(self, agents, lease) -> None: + def phase_effects(self, agents: AgentActivity, lease: TurnLeaseFence) -> None: agents.finish_turn(lease) From 2b1592ff4a5d7f96c26e308dab3d7f4f5051792f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:09:17 -0400 Subject: [PATCH 03/10] P689 keep phase publication with its owner and remove forwarding claude-code order40585 explicitly allows net src <= +20. This records that allowance, not a new author-granted increase. Src-Budget: 20 --- src/agent_comms/agent_activity.py | 4 +--- src/agent_comms/pi_payloads.py | 8 ++++---- src/agent_comms/turn_phase.py | 9 +-------- 3 files changed, 6 insertions(+), 15 deletions(-) diff --git a/src/agent_comms/agent_activity.py b/src/agent_comms/agent_activity.py index 60becd18f3..79a93089fe 100644 --- a/src/agent_comms/agent_activity.py +++ b/src/agent_comms/agent_activity.py @@ -274,9 +274,7 @@ def _turn_phase_effects( with _store_lock(self._wire_lock_path, mode=SharedLock()): effects = transition(lease, phase) for state in effects: - current = state.active.phase - current.phase_effects(self, lease) - self._emit_activity(current.activity_event(lease.identity.incarnation.name)) + state.active.phase.phase_effects(self, lease) return effects def transition_turn(self, lease: TurnLeaseFence, phase: TurnPhase) -> tuple[TurnState, ...]: diff --git a/src/agent_comms/pi_payloads.py b/src/agent_comms/pi_payloads.py index cc133b6772..a2eaf372c5 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,8 +34,8 @@ if TYPE_CHECKING: from .agent_events import ProviderUsage - from .turn_usage import UsageAccount from .turn_phase import TurnPhase + from .turn_usage import UsageAccount def wire_field(name: str, default=None): @@ -507,11 +508,10 @@ class ProviderTransportFailureDiagnostic(PiDiagnostic): timestamp: int | None = None def turn_phase(self, phase: TurnPhase) -> TurnPhase: - return phase.failed(self.error) + from .turn_phase import FailedPhase + return FailedPhase(error=self.error) def refine_model_request_failure(self, failure): - from .turn_failure import ProviderTransportFailed - return ProviderTransportFailed(failure.text) @property diff --git a/src/agent_comms/turn_phase.py b/src/agent_comms/turn_phase.py index aadb3285b3..dc6f9c4573 100644 --- a/src/agent_comms/turn_phase.py +++ b/src/agent_comms/turn_phase.py @@ -124,11 +124,8 @@ def response_output(self) -> TurnPhase: def stalled(self, accepted: bool) -> TurnPhase: return self if accepted else PromptAcceptancePhase() - def failed(self, error: ProviderDiagnosticError) -> TurnPhase: - return FailedPhase(error=error) - def phase_effects(self, agents: AgentActivity, lease: TurnLeaseFence) -> None: - pass + agents._emit_activity(self.activity_event(lease.identity.incarnation.name)) class Excursion(TurnPhase, declared_name=None): @@ -163,10 +160,6 @@ def phase_effects(self, agents: AgentActivity, lease: TurnLeaseFence) -> None: class FailedPhase(IdlePhase): error: ProviderDiagnosticError - @property - def activity_detail(self) -> str: - return self.error.message - class CancellingPhase(TurnPhase): accepts_followup = False From c4d5c77818f4821678ba0c3ee4b793c8b9bd35fd Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:20:23 -0400 Subject: [PATCH 04/10] P689 publish phase activity through the public owner method --- src/agent_comms/turn_phase.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/agent_comms/turn_phase.py b/src/agent_comms/turn_phase.py index dc6f9c4573..d86a319f04 100644 --- a/src/agent_comms/turn_phase.py +++ b/src/agent_comms/turn_phase.py @@ -125,7 +125,9 @@ def stalled(self, accepted: bool) -> TurnPhase: return self if accepted else PromptAcceptancePhase() def phase_effects(self, agents: AgentActivity, lease: TurnLeaseFence) -> None: - agents._emit_activity(self.activity_event(lease.identity.incarnation.name)) + agents.set_activity( + lease.identity.incarnation.name, self.activity_state, self.activity_detail, + ) class Excursion(TurnPhase, declared_name=None): From dfbc382e1ddfea1e7519485f4a40d82d9566c1a3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:20:17 -0400 Subject: [PATCH 05/10] Close orders from merges contained in the installed checkout Refs #1997. New direct commander request source 2996; preserves P254 separately. --- src/agent_comms/commander/config.py | 21 +++++- src/agent_comms/commander/order.py | 11 ++- src/agent_comms/todos.py | 56 ++++++++++++++-- tests/test_installed_merge.py | 100 ++++++++++++++++++++++++++++ tests/test_order_progress.py | 20 +++--- 5 files changed, 186 insertions(+), 22 deletions(-) create mode 100644 tests/test_installed_merge.py diff --git a/src/agent_comms/commander/config.py b/src/agent_comms/commander/config.py index c4c11c4e7f..296a46672c 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,12 @@ 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.""" + return subprocess.run( + ('git', '-C', str(self.checkout), 'merge-base', '--is-ancestor', sha, 'HEAD'), + capture_output=True, timeout=60).returncode == 0 + @classmethod def of(cls, checkout: Path) -> Repo: """Its GitHub name comes from the checkout's own origin remote.""" @@ -102,12 +109,22 @@ def write(self, home: Path) -> Path: class Commander: """This machine's commander thread and the bus whose Todo store owns its problems and orders.""" - name: str + config: CommanderConfig comms: Comms + @property + def name(self) -> str: + return self.config.commander + @classmethod def of(cls, config: CommanderConfig) -> Commander: - return cls(config.commander, Comms(config.root)) + return cls(config, Comms(config.root)) + + def orders(self) -> tuple[Todo, ...]: + actor = self.comms.registry.require(self.name).incarnation + store = self.comms.goals.todos + store.reconcile_merges(actor, self.config.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 diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 83e875478f..f464b6e632 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -82,17 +82,17 @@ def call(tool): def todos(): """The problems this commander created; Core's Todo store is their one owner.""" - return COMMS.goals.todos.for_commander(COMMS.registry.require(ME).incarnation) + return COMMANDER.orders() def held_orders() -> tuple[Todo, ...]: """This commander's open orders that a unit holds.""" - return COMMS.goals.todos.held_for_commander(COMMS.registry.require(ME).incarnation) + return tuple(todo for todo in todos() if todo.assignment) def held_by(unit: str) -> tuple[Todo, ...]: """This commander's open orders held by one unit.""" - return COMMS.goals.todos.held_by(COMMS.registry.require(ME).incarnation, unit) + return tuple(todo for todo in held_orders() if todo.holder == unit) def units() -> frozenset[str]: @@ -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 diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 281e53ab5d..c87c24c46e 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -47,6 +47,7 @@ if TYPE_CHECKING: from .comms import Comms from .goal_management import Goals + from .commander.config import Repo @dataclass(frozen=True) @@ -590,18 +591,21 @@ 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 close(self, todo: Todo) -> Todo: + return DoneTodoState.closed(todo, todo.order.merged(self.sha)) def waiting_on_install(self, todo: Todo) -> tuple[Todo, ...]: return (todo,) @@ -1134,6 +1138,9 @@ def require_transfer(self, todo: Todo) -> None: def require_close(self): raise ValueError("Closing requires recorded installed application evidence.") + def merged(self, sha: str) -> ProblemOrder: + raise ValueError("Only a recorded order can close from its installed merge.") + def withdrawals(self, todo: Todo) -> tuple[str, ...]: return () @@ -1238,6 +1245,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 +1320,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 +1651,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 +2328,23 @@ 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: + """Observe outside the write lock; commit only orders whose observed revision still holds.""" + todos = self.held_for_commander(commander) + pulls = PullRequests(todos) + completed = tuple(merged.close(todo) for todo in todos + 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/tests/test_installed_merge.py b/tests/test_installed_merge.py new file mode 100644 index 0000000000..b46a3c38f8 --- /dev/null +++ b/tests/test_installed_merge.py @@ -0,0 +1,100 @@ +"""Merge observation closes the real stored order, without a completion request.""" +from dataclasses import replace + +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) +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) + monkeypatch.setattr(script, 'COMMANDER', replace(script.COMMANDER, 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.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.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 + 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 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] From 4c97aaa844dea1d02683dee39ba5676cd59f0ebf Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:22:07 -0400 Subject: [PATCH 06/10] Derive the short merge label from its retained full SHA --- src/agent_comms/standing_rules.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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" ) From 9dcd1f3c912bca30aef0c9a0dd427c80b339ff92 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:24:34 -0400 Subject: [PATCH 07/10] Keep the merge reconciliation declarations lint-clean --- src/agent_comms/todos.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index c87c24c46e..0c252c1aec 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -45,9 +45,9 @@ from .plan_phases import OrderPhase, UnplannedPhase if TYPE_CHECKING: + from .commander.config import Repo from .comms import Comms from .goal_management import Goals - from .commander.config import Repo @dataclass(frozen=True) @@ -2330,12 +2330,13 @@ def for_goal(self, owner: ThreadIncarnation, goal_id: str) -> tuple[Todo, ...]: def reconcile_merges(self, commander: ThreadIncarnation, repos: Sequence[Repo], storage: StorageRule) -> None: - """Observe outside the write lock; commit only orders whose observed revision still holds.""" + """Read before locking; close only orders whose observed revision still holds.""" todos = self.held_for_commander(commander) pulls = PullRequests(todos) completed = tuple(merged.close(todo) for todo in todos for repo in repos if repo.github.lower() == todo.repo - for merged in MergedProgress.found(ProgressSource.of(todo, storage, pulls)) + for merged in MergedProgress.found( + ProgressSource.of(todo, storage, pulls)) if repo.contains(merged.sha)) if not completed: return From 3d52404a5ea9d7c38fa3fe5dad72aa2e4acf56bc Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:26:22 -0400 Subject: [PATCH 08/10] Observe installed merges on board and overdue reads, not on order writes --- src/agent_comms/commander/order.py | 10 +++++----- tests/test_installed_merge.py | 3 ++- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index f464b6e632..c0b0af7de8 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -82,17 +82,17 @@ def call(tool): def todos(): """The problems this commander created; Core's Todo store is their one owner.""" - return COMMANDER.orders() + return COMMS.goals.todos.for_commander(COMMS.registry.require(ME).incarnation) def held_orders() -> tuple[Todo, ...]: """This commander's open orders that a unit holds.""" - return tuple(todo for todo in todos() if todo.assignment) + return COMMS.goals.todos.held_for_commander(COMMS.registry.require(ME).incarnation) def held_by(unit: str) -> tuple[Todo, ...]: """This commander's open orders held by one unit.""" - return tuple(todo for todo in held_orders() if todo.holder == unit) + return COMMS.goals.todos.held_by(COMMS.registry.require(ME).incarnation, unit) def units() -> frozenset[str]: @@ -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(), 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:] @@ -617,7 +617,7 @@ def overdue(): installs: list[str] = [] while True: now = time.time() - held = held_orders() + held = tuple(todo for todo in COMMANDER.orders() 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/tests/test_installed_merge.py b/tests/test_installed_merge.py index b46a3c38f8..ffb387a56c 100644 --- a/tests/test_installed_merge.py +++ b/tests/test_installed_merge.py @@ -45,13 +45,14 @@ 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.held_orders() == (todo,) + assert script.COMMANDER.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() == () assert script.held_orders() == () closed = comms.goals.todos.get(todo.id) assert closed.state is DoneTodoState and closed.assignment is None From 0adf40e05738404d3ba916f72aab95eb29e7b083 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:37:49 -0400 Subject: [PATCH 09/10] P266: close only checked orders and reuse their stored terminal merge Remove the raising base method and forwarding close; preserve real Git failures. Repeated reads exclude closed records through the existing store selection, without another cache. --- src/agent_comms/commander/config.py | 19 ++++++++--------- src/agent_comms/commander/order.py | 4 ++-- src/agent_comms/todos.py | 14 +++++-------- tests/test_installed_merge.py | 32 +++++++++++++++++++++++++---- 4 files changed, 44 insertions(+), 25 deletions(-) diff --git a/src/agent_comms/commander/config.py b/src/agent_comms/commander/config.py index 296a46672c..df6c020045 100644 --- a/src/agent_comms/commander/config.py +++ b/src/agent_comms/commander/config.py @@ -43,9 +43,12 @@ class Repo(BaseModel): def contains(self, sha: str) -> bool: """Only a successful ancestry proof in this installed checkout permits closure.""" - return subprocess.run( + result = subprocess.run( ('git', '-C', str(self.checkout), 'merge-base', '--is-ancestor', sha, 'HEAD'), - capture_output=True, timeout=60).returncode == 0 + 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: @@ -109,21 +112,17 @@ def write(self, home: Path) -> Path: class Commander: """This machine's commander thread and the bus whose Todo store owns its problems and orders.""" - config: CommanderConfig + name: str comms: Comms - @property - def name(self) -> str: - return self.config.commander - @classmethod def of(cls, config: CommanderConfig) -> Commander: - return cls(config, Comms(config.root)) + return cls(config.commander, Comms(config.root)) - def orders(self) -> tuple[Todo, ...]: + 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, self.config.repos, self.comms.registry.storage_rule) + store.reconcile_merges(actor, repos, self.comms.registry.storage_rule) return store.for_commander(actor) def units(self) -> frozenset[str]: diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index c0b0af7de8..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 = COMMANDER.orders(), 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:] @@ -617,7 +617,7 @@ def overdue(): installs: list[str] = [] while True: now = time.time() - held = tuple(todo for todo in COMMANDER.orders() if todo.assignment) + 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/todos.py b/src/agent_comms/todos.py index 0c252c1aec..19c45babcb 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -604,9 +604,6 @@ def found(cls, source: ProgressSource) -> Iterator[Self]: def state(self) -> str: return f"merged as {self.sha[:9]}, waiting on install" - def close(self, todo: Todo) -> Todo: - return DoneTodoState.closed(todo, todo.order.merged(self.sha)) - def waiting_on_install(self, todo: Todo) -> tuple[Todo, ...]: return (todo,) @@ -1138,9 +1135,6 @@ def require_transfer(self, todo: Todo) -> None: def require_close(self): raise ValueError("Closing requires recorded installed application evidence.") - def merged(self, sha: str) -> ProblemOrder: - raise ValueError("Only a recorded order can close from its installed merge.") - def withdrawals(self, todo: Todo) -> tuple[str, ...]: return () @@ -2331,9 +2325,11 @@ def for_goal(self, owner: ThreadIncarnation, goal_id: str) -> tuple[Todo, ...]: def reconcile_merges(self, commander: ThreadIncarnation, repos: Sequence[Repo], storage: StorageRule) -> None: """Read before locking; close only orders whose observed revision still holds.""" - todos = self.held_for_commander(commander) - pulls = PullRequests(todos) - completed = tuple(merged.close(todo) for todo in todos + 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)) diff --git a/tests/test_installed_merge.py b/tests/test_installed_merge.py index ffb387a56c..53399af943 100644 --- a/tests/test_installed_merge.py +++ b/tests/test_installed_merge.py @@ -1,12 +1,13 @@ """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) + 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 @@ -38,21 +39,20 @@ def test_installed_merge_closes_pending_or_delivered_order_once( repo = Repo(checkout=installed, github=todo.repo) config = script.CONFIG.model_copy(update={'repos': (repo,)}) monkeypatch.setattr(script, 'CONFIG', config) - monkeypatch.setattr(script, 'COMMANDER', replace(script.COMMANDER, 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() == (todo,) + 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() == () + 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 @@ -62,6 +62,14 @@ def test_installed_merge_closes_pending_or_delivered_order_once( 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) == [] @@ -99,3 +107,19 @@ def observed(self, sha): 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) From 6011bed5f7955bae5489b3508fccfeb36f282cee Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:41:33 -0400 Subject: [PATCH 10/10] Budget the batch growth read in review Src-Budget: 220