diff --git a/src/agent_comms/goal_presentation.py b/src/agent_comms/goal_presentation.py index c1744e4e0f..20584dac9e 100644 --- a/src/agent_comms/goal_presentation.py +++ b/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() diff --git a/src/agent_comms/owner_lifecycle.py b/src/agent_comms/owner_lifecycle.py index 2d4e616f55..252b528e87 100644 --- a/src/agent_comms/owner_lifecycle.py +++ b/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() ): diff --git a/tests/conftest.py b/tests/conftest.py index 940d72178d..a788b8645a 100644 --- a/tests/conftest.py +++ b/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.""" diff --git a/tests/goal_owner_fixture.py b/tests/goal_owner_fixture.py index 2aa982d2d2..33e0220004 100644 --- a/tests/goal_owner_fixture.py +++ b/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) + + diff --git a/tests/test_goal_answers.py b/tests/test_goal_answers.py index 005712c7f5..907fea00dd 100644 --- a/tests/test_goal_answers.py +++ b/tests/test_goal_answers.py @@ -1,7 +1,6 @@ """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 @@ -9,7 +8,6 @@ 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( diff --git a/tests/test_goal_execution.py b/tests/test_goal_execution.py index c58926e091..5e9fe3b4d7 100644 --- a/tests/test_goal_execution.py +++ b/tests/test_goal_execution.py @@ -1,7 +1,8 @@ -"""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 @@ -9,6 +10,7 @@ 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. - diff --git a/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py index ada2c2f221..5884abafcd 100644 --- a/tests/test_holder_standing_rules.py +++ b/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: diff --git a/tests/test_native_unknown_recovery.py b/tests/test_native_unknown_recovery.py index 99dc4ee6e1..dc988ab9e6 100644 --- a/tests/test_native_unknown_recovery.py +++ b/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())