P448: installed stop proof and completed carry removal

src/agent_comms/goal_presentation.py

@@ -3,12 +3,12 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from collections.abc import Mapping
-from dataclasses import asdict, dataclass, fields
+from dataclasses import dataclass, fields
 from functools import cached_property
 from typing import TYPE_CHECKING, ClassVar, Self
 
-from .field_codec import NullMember, RecordMember, SubclassUnion, WireValue
+from .declared_family import DeclaredFamily
+from .field_codec import NullMember, WireValue
 from .goal_failure_observation import FailedTurnEvidence
 from .goal_states import PauseSource
 from .goals import Goal, GoalCheckpoint, GoalDeclaration
@@ -39,7 +39,7 @@ def to_wire(self) -> dict[str, str | float]:
         return self.incarnation.to_wire()
 
     @classmethod
-    def from_wire(cls, data: Mapping) -> GoalWaitTarget:
+    def from_wire(cls, data: dict[str, object]) -> GoalWaitTarget:
         return cls(RecordedThreadIncarnation.from_wire(data))
 
     @classmethod
@@ -71,7 +71,7 @@ def execution(self) -> GoalExecution:
         return self.goal.state.execution(self)
 
 
-class GoalExecution(SubclassUnion, affix="Execution"):
+class GoalExecution(DeclaredFamily, affix="Execution"):
     """The complete execution family consumed by commands, views and the commander."""
 
     def details_heading(self, heading: str) -> str:
@@ -97,7 +97,7 @@ def heading(self, revision: int) -> str: ...
 
 
 @dataclass(frozen=True)
-class IdleGoalExecution(NullMember, GoalExecution):
+class IdleGoalExecution(GoalExecution):
     def validate_goal(self, goal: GoalCheckpoint) -> None:
         pass
 
@@ -106,7 +106,7 @@ def heading(self, revision: int) -> str:
 
 
 @dataclass(frozen=True)
-class BoundGoalExecution(RecordMember, GoalExecution, declared_name=None):
+class BoundGoalExecution(GoalExecution, declared_name=None):
     goal_id: str
     glyph: ClassVar[str] = "✓"
     @property
@@ -149,6 +149,15 @@ def toggle(self) -> type[GoalAction]:
         return PausedGoalAction
 
 
+class RetryableExecution(ControlledExecution, declared_name=None):
+    toggle_label = "Retry"
+
+    @property
+    def toggle(self) -> type[GoalAction]:
+        from .goal_actions import RetryGoalAction
+        return RetryGoalAction
+
+
 @dataclass(frozen=True)
 class RunnableExecution(ActiveExecution):
     label = "Active"
@@ -200,15 +209,9 @@ def waiting_for(self, goal: Goal) -> bool:
         return goal.id == self.goal_id and bool(goal.state.items_with(Active, goal))
 
 
-class BlockedExecution(WaitingExecution, ControlledExecution, declared_name=None):
+class BlockedExecution(WaitingExecution, declared_name=None):
     glyph = "!"
     label = "Blocked"
-    toggle_label = "Retry"
-
-    @property
-    def toggle(self) -> type[GoalAction]:
-        from .goal_actions import RetryGoalAction
-        return RetryGoalAction
 
     @property
     @abstractmethod
@@ -229,7 +232,7 @@ def unit_block_line(self, unit: str, orders: tuple[str, ...]) -> str:
 
 
 @dataclass(frozen=True)
-class DecidedBlockExecution(BlockedExecution):
+class DecidedBlockExecution(BlockedExecution, RetryableExecution):
     block_reason: str
 
     @property
@@ -238,7 +241,7 @@ def reason(self) -> str:
 
 
 @dataclass(frozen=True)
-class UnrecordedBlockExecution(BlockedExecution):
+class UnrecordedBlockExecution(BlockedExecution, RetryableExecution):
     @property
     def reason(self) -> str:
         return "reason unavailable"
@@ -255,7 +258,7 @@ def unit_block_line(self, unit: str, orders: tuple[str, ...]) -> str:
                 f"goal {self.goal_id} -> {self.reason}")
 
 @dataclass(frozen=True)
-class AttemptFailedExecution(BlockedExecution):
+class AttemptFailedExecution(BlockedExecution, RetryableExecution):
     evidence: FailedTurnEvidence
 
     @property
@@ -285,7 +288,7 @@ class GoalSnapshot(GoalPublicationSnapshot):
     execution: GoalExecution
 
     def changed_by(self, current: GoalSnapshot) -> bool:
-        return asdict(self) != asdict(current)
+        return self != current
 
     def standby(self) -> bool:
         return self.execution.standby()

src/agent_comms/owner_lifecycle.py

@@ -444,6 +444,9 @@ def _stop_selection(self, name: str) -> tuple[Thread, int] | None:
         with _store_lock(self._wire_lock_path):
             snapshot = self.registry.snapshot()
             thread = self.registry.require(name)
+            thread.role.require_executable(
+                RelationViolationError("Only agent threads can be stopped.")
+            )
             if not snapshot.status(thread.name).active and (
                 thread.process_identity.allows_inactive_stop()
             ):

tests/conftest.py

@@ -12,6 +12,7 @@
 from pathlib import Path
 
 import pytest
+import goal_owner_fixture
 from xdist import is_xdist_worker
 from source_programs import SourcePrograms
 from source_revision_fixture import SourceRepository
@@ -27,6 +28,8 @@
 from agent_comms.store_files import _store_lock
 from agent_comms.thread_text import ForkedParent, GivenThreadText
 
+pytest_plugins = (goal_owner_fixture.__name__,)
+
 
 class TranscriptRecords:
     """Fixed test time, projected through each record's declared constructor."""

tests/goal_owner_fixture.py

@@ -1,6 +1,10 @@
 """Canonical bus setup for owner lifecycle tests with no automatic model work."""
 
 
+import shutil
+
+import pytest
+
 from agent_comms.comms import Comms
 
 
@@ -11,3 +15,11 @@ def canonical_goal_wire(root):
     return comms
 
 
+@pytest.fixture
+def copied_goal_bus(tmp_path):
+    source = canonical_goal_wire(tmp_path / "source")
+    target = tmp_path / "copy"
+    shutil.copytree(source.root, target)
+    return Comms(target)
+
+

tests/test_goal_answers.py

@@ -1,15 +1,13 @@
 """Goal absence is dispatched before work; a clear never revives a turn's goal."""
 
 import os
-import shutil
 from contextlib import ExitStack, contextmanager
 from uuid import uuid4
 
 import pytest
 
 from agent_comms import agent_events as events
 from agent_comms.child_process import ProcessIdentity
-from agent_comms.comms import Comms
 from agent_comms.goal_actions import ClearGoalAction, SetGoalAction
 from agent_comms.goal_attempts import ProviderUsageTotal
 from agent_comms.goal_turn_permit import NoLaunchPermit
@@ -20,15 +18,6 @@
 from agent_comms.turn_terminal import EndedWithoutResult, OperatorStopped
 
 
-@pytest.fixture
-def copied_goal_bus(tmp_path):
-    source = Comms(tmp_path / "source")
-    source.messaging.initialize_private_initial_protocol()
-    target = tmp_path / "copy"
-    shutil.copytree(source.root, target)
-    return Comms(target)
-
-
 @contextmanager
 def goal_turn(comms, original_goal):
     comms.registry.declare(comms.registry.new_thread(

tests/test_goal_execution.py

@@ -1,14 +1,16 @@
-"""Real turn acquisition, terminal handling and cutover share the execution owner."""
+"""Real turn acquisition and terminal handling share the execution owner."""
 import os
 import sqlite3
 from contextlib import AsyncExitStack, asynccontextmanager, closing
+from dataclasses import fields
 
 import pytest
 
 from agent_comms import agent_events as events
 from agent_comms.acp import CommsAgent
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.comms import Comms
+from agent_comms.field_codec import FieldCodec
 from agent_comms.goal_actions import (
     PausedGoalAction,
     RetryGoalAction,
@@ -17,6 +19,8 @@
 from agent_comms.goal_failure_observation import FailedTurnEvidence
 from agent_comms.goal_presentation import (
     AttemptFailedExecution,
+    GoalExecution,
+    GoalSnapshot,
     RunnableExecution,
     RunningExecution,
 )
@@ -36,6 +40,18 @@ def execution_bus(copied_goal_bus):
     return comms
 
 
+@pytest.mark.parametrize("execution_type", tuple(
+    member for member in GoalExecution.members_with(GoalExecution)
+    if fields(member) == fields(RunnableExecution)
+))
+def test_execution_identity_survives_wire_and_change_detection(execution_bus, execution_type):
+    goal = execution_bus.registry.require("worker").goal
+    execution = execution_type(goal.id)
+    assert FieldCodec.decode(GoalExecution, FieldCodec.encode(execution)) == execution
+    current = GoalSnapshot(goal, RunnableExecution(goal.id))
+    assert current.changed_by(GoalSnapshot(goal, execution)) == (current.execution != execution)
+
+
 @asynccontextmanager
 async def turn_on(comms, *, original_owner_input=False):
     agent = CommsAgent(comms, auto_wake=False, owner_stop=OwnerStop())
@@ -123,4 +139,3 @@ async def test_verified_productive_turn_recovers_blocked_generation(execution_bu
     assert isinstance(comms.goals.goal_execution("worker"), RunnableExecution)
     assert comms.registry.require("worker").goal.state == ActiveGoal()
     assert failure_count(comms) == 1  # The old failure remains evidence, never replayed.
-

tests/test_holder_standing_rules.py

@@ -26,6 +26,7 @@
 from agent_comms.commander_heartbeat import CommanderHeartbeat
 from agent_comms.comms import Comms
 from agent_comms.coordinator import Coordination
+from agent_comms.errors import RelationViolationError
 from agent_comms.field_codec import FieldCodec
 from agent_comms.goal_actions import (
     CreateProblemGoalAction, DirectiveGoalAction, ReissuedOrderGoalAction, SetGoalAction,
@@ -61,6 +62,7 @@ def held_bus(comms, tmp_path, source_repository, configured_repository):
     """One setup owner: real declarations, checked orders, UNKNOWN row, then a copy."""
     source = source_repository
     scope = source.scope()
+    comms.messaging.initialize_private_initial_protocol()
     declare_test_bus(comms, channels=('comms',), threads=tuple(
         comms.registry.new_thread(name=name, tags=frozenset({'comms'}), worktree=str(source.checkout))
         for name in ('lead', 'worker', 'outsider')
@@ -96,6 +98,14 @@ def held_bus(comms, tmp_path, source_repository, configured_repository):
     return copied_bus
 
 
+def test_stop_rejects_a_human_before_any_registry_change(comms, tmp_path):
+    reader = comms.messaging.user_identity(str(tmp_path))
+    before = comms.registry.store.path.read_bytes()
+    with pytest.raises(RelationViolationError, match="Only agent threads can be stopped"):
+        comms.owners.stop(reader.name)
+    assert comms.registry.store.path.read_bytes() == before
+
+
 def blocked_holder(comms):
     """Settle a real failed turn; never manufacture a stored backend block."""
     owner = comms.registry.require("worker")
@@ -163,9 +173,8 @@ def test_an_order_to_a_blocked_goal_activates_it_and_keeps_other_orders(held_bus
 
 
 async def test_sigterm_group_preserves_active_goal_and_never_reissues_orders(
-    held_bus, native_runtime_package, source_programs, tmp_path,
+    held_bus, native_runtime_package, tmp_path,
 ):
-    source_programs.bind(held_bus.root.parent)
     provider_root = tmp_path / "stopping-provider"
     provider_root.mkdir(mode=0o700)
     async with native_backend_fixture(provider_root, package=native_runtime_package) as fixture:

tests/test_native_unknown_recovery.py

@@ -132,7 +132,7 @@ def running_successor(comms):
         new_owner=True,
     )
     successor, _ = comms.registry.lease_local_turn("beta", "productive-successor")
-    assert successor.process_alive and successor.turn_record.read().executing
+    assert successor.executing
     return successor
 
 
@@ -141,7 +141,7 @@ def test_undispatched_original_recovers_beside_a_running_successor(released_star
     successor = running_successor(comms)
     recover_starting_owner(released_starting)
     assert comms.registry.require("beta") == successor
-    assert successor.process_alive and successor.turn_record.read().executing
+    assert successor.executing
     comms.registry.release_turn(successor.require_turn_lease())
 
 
@@ -387,7 +387,7 @@ def test_dead_original_settles_unknown_beside_a_live_successor(released_unknown)
         assert input_evidence(store) == before_input
     assert (root / "bus.jsonl").read_bytes() == before_bus
     assert comms.registry.require("beta") == successor
-    assert successor.process_alive and successor.turn_record.read().executing
+    assert successor.executing
     comms.registry.release_turn(successor.require_turn_lease())