diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index cd7d376377..34234da01a 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -3,43 +3,45 @@ from __future__ import annotations import time -from collections.abc import Iterable, Sequence from abc import abstractmethod +from collections.abc import Iterable, Sequence from contextlib import closing from dataclasses import dataclass, field from pathlib import Path from typing import ClassVar, Self from agent_comms.build_identity import RecordedBuildIdentity +from agent_comms.bus_page_index import BusPageIndex +from agent_comms.channel_targets import ThreadTarget +from agent_comms.commander.config import CommanderConfig +from agent_comms.commander_heartbeat import CommanderHeartbeat from agent_comms.comms import Comms from agent_comms.goal_actions import ReissuedOrderGoalAction -from agent_comms.goals import Goal from agent_comms.goal_presentation import BlockedExecution -from agent_comms.message_page import MessageReadRequest +from agent_comms.goals import Goal +from agent_comms.message_bus import WatchPass +from agent_comms.message_page import MessageReadRequest, PageBounds from agent_comms.message_reference import MessageReference from agent_comms.messages import Message -from agent_comms.read_basis import RepliesDisplayScope -from agent_comms.bus_page_index import BusPageIndex -from agent_comms.wire_log import WireSnapshot -from agent_comms.message_bus import WatchPass -from agent_comms.channel_targets import ThreadTarget -from agent_comms.commander_heartbeat import CommanderHeartbeat -from agent_comms.message_page import PageBounds -from agent_comms.read_basis import NamedViewer +from agent_comms.phase_evidence import PlanReading +from agent_comms.plan_phases import PlanPhase +from agent_comms.read_basis import NamedViewer, RepliesDisplayScope from agent_comms.recorded_time import RecordedEventTime from agent_comms.standing_rules import ( - DatedStandingRule, OwedAnswer, ReasonedStandingRule, StandingRule, UndatedStandingRule, + DatedStandingRule, + OwedAnswer, + ReasonedStandingRule, + StandingRule, + UndatedStandingRule, ) from agent_comms.storage_rules import OrderCheckoutWork from agent_comms.thread_identity import CoreWatcherIncarnation, ThreadIncarnation from agent_comms.threads import Thread from agent_comms.todos import Todo from agent_comms.turn_digests import OrderEditCountRead, TurnDigestRead, TurnsSince +from agent_comms.wire_log import WireSnapshot from agent_comms.work_branches import WorkBranches from agent_comms.work_dispatch import ReusedWorkDispatch, WorkDispatch -from agent_comms.commander.config import CommanderConfig -from agent_comms.phase_evidence import PlanReading -from agent_comms.plan_phases import PlanPhase @dataclass(frozen=True) @@ -131,25 +133,21 @@ def line(self) -> str: @dataclass(frozen=True) class GoalCannotRun(UndatedStandingRule): - """An ordered unit whose goal execution supplies the block and its action.""" + """Report blocked work without rewriting the unit's goal or orders.""" unit: str orders: tuple[str, ...] - goal: Goal execution: BlockedExecution @classmethod def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: return tuple( - cls(owner.name, tuple(todo.id for todo in todos), view.goal.require(), block) + cls(owner.name, tuple(todo.id for todo in todos), block) for owner, todos in cls.holders(comms, commander).items() for view in (comms.goals.goal_snapshot(owner.name),) for block in view.execution.unit_blocks() ) - def enforce(self, comms: Comms, commander: str) -> None: - self.execution.enforce_unit_block(comms, commander, self.unit, self.orders, self.goal) - def line(self) -> str: return self.execution.unit_block_line(self.unit, self.orders) diff --git a/src/agent_comms/goal_attempts.py b/src/agent_comms/goal_attempts.py index d5fd121a66..42c598c46f 100644 --- a/src/agent_comms/goal_attempts.py +++ b/src/agent_comms/goal_attempts.py @@ -11,8 +11,6 @@ from __future__ import annotations -from .lifecycle_capability import Failed - import hashlib import hmac import json @@ -27,9 +25,11 @@ from dataclasses import dataclass, field from decimal import Decimal from pathlib import Path -from typing import TYPE_CHECKING, ClassVar, TypeVar, Annotated +from typing import TYPE_CHECKING, Annotated, ClassVar, TypeVar from uuid import uuid4 +from agent_comms.goal_turn_permit import GoalTurnPermit + from .coordination_database import CoordinationStore from .declared_family import DeclaredFamily from .field_codec import MalformedRecord @@ -59,8 +59,8 @@ ReadyGeneration, ReservedGeneration, ) +from .lifecycle_capability import Failed from .locked_store import DurableFile, SQLiteFile -from agent_comms.goal_turn_permit import GoalTurnPermit from .pi_counts import PiCount, UnreportedCount from .pi_payloads import PiUsage from .typed_table import Column, Generated, SQLiteJournalMode, TypedTable @@ -135,6 +135,14 @@ def failure_identity(self) -> GoalAttemptIdentity: self.lifecycle.require(Failed, FailureNotObserved()) return self.attempt_identity + def failure_evidence(self, store: GoalAttemptStore) -> GoalFailureEvidence: + identity = self.failure_identity() + with closing(store._connect()) as conn: + return next(FailedTurnEvidence.stream( + conn, attempt_id=identity.attempt_id, + goal_id=identity.goal_id, generation=identity.generation, + ), NoFailedTurnEvidence()) + def has_ready_grant(self, number: int, digest: str) -> bool: return self.lifecycle.ready and self.number == number and self.ready_digest == digest @@ -807,12 +815,6 @@ def read_attempt(self, attempt_id: str) -> LedgerAttempt: with closing(self._connect()) as conn: return AttemptRecord.read_in(conn, attempt_id) - def failure_evidence(self, generation: Generation) -> GoalFailureEvidence: - with closing(self._connect()) as conn: - return next(FailedTurnEvidence.stream( - conn, attempt_id=generation.attempt_identity.attempt_id, - ), NoFailedTurnEvidence()) - def stopped_permit(self, goal_id: str) -> GoalTurnPermit: generation = self.require_generation(goal_id) return generation.lifecycle.stopped_permit(self, generation) diff --git a/src/agent_comms/goal_generation.py b/src/agent_comms/goal_generation.py index 0ceb2fce1a..e4ea68cb42 100644 --- a/src/agent_comms/goal_generation.py +++ b/src/agent_comms/goal_generation.py @@ -159,7 +159,7 @@ def resume_status(self, generation: Generation, goal: Goal, ctx: GoalActionConte super().resume_status(generation, goal, ctx) def evidence(self, store: GoalAttemptStore, generation: Generation) -> GoalFailureEvidence: - return store.failure_evidence(generation) + return generation.failure_evidence(store) def idle_execution(self, source: GoalExecutionSource) -> GoalExecution: return source.evidence.execution(source.goal.id) diff --git a/src/agent_comms/goal_presentation.py b/src/agent_comms/goal_presentation.py index b06de85097..c1744e4e0f 100644 --- a/src/agent_comms/goal_presentation.py +++ b/src/agent_comms/goal_presentation.py @@ -4,22 +4,21 @@ from abc import ABC, abstractmethod from collections.abc import Mapping -from dataclasses import asdict, dataclass, fields, replace +from dataclasses import asdict, dataclass, fields from functools import cached_property from typing import TYPE_CHECKING, ClassVar, Self from .field_codec import NullMember, RecordMember, SubclassUnion, WireValue from .goal_failure_observation import FailedTurnEvidence from .goal_states import PauseSource -from .goals import Goal, GoalChangedError, GoalCheckpoint, GoalDeclaration +from .goals import Goal, GoalCheckpoint, GoalDeclaration from .lifecycle_capability import Active -from .thread_identity import CoreWatcherIncarnation, RecordedThreadIncarnation +from .thread_identity import RecordedThreadIncarnation from .thread_presentation import ThreadPresentation if TYPE_CHECKING: from .agent_activity import AgentActivity from .bus_publication import CommittedDelivery - from .comms import Comms from .goal_actions import GoalAction from .goal_attempts import Generation from .goal_failure_observation import GoalFailureEvidence @@ -226,27 +225,7 @@ def unit_blocks(self) -> tuple[BlockedExecution, ...]: def unit_block_line(self, unit: str, orders: tuple[str, ...]) -> str: return (f"BLOCKED {unit}: orders {', '.join(orders)}; goal {self.goal_id}; " - f"{self.reason} -> Core reissues the orders") - - def enforce_unit_block( - self, comms: Comms, commander: str, unit: str, orders: tuple[str, ...], goal: Goal, - ) -> None: - from .goal_actions import GoalPrecondition, ReissuedOrderGoalAction - - if comms.goals.goal_execution(unit) != self: - raise GoalChangedError(comms.registry.require(unit), goal.captured) - observed = goal - for problem in orders: - todo = comms.goals.todos.get(problem) - todo.require_assignee(comms.registry.require(unit).incarnation, observed.id) - action = ReissuedOrderGoalAction.for_todo( - todo, commander, CoreWatcherIncarnation.watcher()) - observed = comms.goals.update_goal(unit, replace( - action, expect=GoalPrecondition( - goal_id=observed.id, expected_state=observed.state, - expected_checkpoint=observed.captured, - ), - )) + f"{self.reason} -> inspect the failed work; never replay") @dataclass(frozen=True) @@ -275,12 +254,6 @@ def unit_block_line(self, unit: str, orders: tuple[str, ...]) -> str: return (f"UNRESOLVED BLOCK {unit}: orders {', '.join(orders)}; " f"goal {self.goal_id} -> {self.reason}") - def enforce_unit_block( - self, comms: Comms, commander: str, unit: str, orders: tuple[str, ...], goal: Goal, - ) -> None: - pass - - @dataclass(frozen=True) class AttemptFailedExecution(BlockedExecution): evidence: FailedTurnEvidence diff --git a/tests/test_goal_execution.py b/tests/test_goal_execution.py index ab93bedadb..b7e97e0a1c 100644 --- a/tests/test_goal_execution.py +++ b/tests/test_goal_execution.py @@ -5,20 +5,27 @@ from dataclasses import replace import pytest +from cutover.carry_goal_execution import FAILED_BACKEND_REASON, BackendGoalBlocks 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.goal_actions import ( - BlockedGoalAction, PausedGoalAction, RetryGoalAction, SetGoalAction, + BlockedGoalAction, + PausedGoalAction, + RetryGoalAction, + SetGoalAction, ) from agent_comms.goal_failure_observation import FailedTurnEvidence -from agent_comms.goal_presentation import AttemptFailedExecution, RunnableExecution, RunningExecution +from agent_comms.goal_presentation import ( + AttemptFailedExecution, + RunnableExecution, + RunningExecution, +) from agent_comms.goal_states import ActiveGoal, BlockedGoal from agent_comms.owned_turn import OwnedTurn from agent_comms.turn_terminal import OwnerStop -from cutover.carry_goal_execution import BackendGoalBlocks, FAILED_BACKEND_REASON from test_goal_answers import copied_goal_bus @@ -45,7 +52,6 @@ async def turn_on(comms, *, original_owner_input=False): ) async with AsyncExitStack() as resources, AsyncExitStack() as permits: assert await turn.acquire(resources, permits) - turn.open_stream(resources, permits) yield turn diff --git a/tests/test_goal_failure_observation.py b/tests/test_goal_failure_observation.py index 030dee5af4..b5c90ec931 100644 --- a/tests/test_goal_failure_observation.py +++ b/tests/test_goal_failure_observation.py @@ -9,7 +9,12 @@ from agent_comms.child_process import ProcessIdentity from agent_comms.diagnostics import FailureReason -from agent_comms.goal_attempts import AttemptRecord, GoalAttemptStore, StaleAttemptError, UnresolvedAttemptError +from agent_comms.goal_attempts import ( + AttemptRecord, + GoalAttemptStore, + StaleAttemptError, + UnresolvedAttemptError, +) from agent_comms.goal_failure_observation import FailedTurnObservation, NoFailedTurnEvidence from agent_comms.goal_generation import BlockedGeneration from agent_comms.goals import Goal @@ -58,7 +63,7 @@ def test_duplicate_callback_and_restart_preserve_failure_bytes(bound): observation.reservation.fail(current, "duplicate", observation=observation) with pytest.raises(UnresolvedAttemptError): current.resume(owner.goal.id, 1) - assert current.failure_evidence(current.require_generation(owner.goal.id)) == observation.evidence + assert current.require_generation(owner.goal.id).failure_evidence(current) == observation.evidence assert store.path.read_bytes() == before assert len(rows(store, "failed_turn_evidence")) == 1 encoded = json.dumps(rows(store, "failed_turn_evidence")) @@ -81,7 +86,7 @@ def test_mismatched_turn_lease_is_not_bound(bound, mutation): assert rejected is None observation.reservation.fail(store, "failed anyway", observation=rejected) assert store.snapshot("goal").lifecycle == BlockedGeneration() - assert store.failure_evidence(store.require_generation("goal")) == NoFailedTurnEvidence() + assert store.require_generation("goal").failure_evidence(store) == NoFailedTurnEvidence() @pytest.mark.parametrize("binding", ["token", "evidence_identity"]) @@ -95,7 +100,7 @@ def test_mismatched_permit_cannot_attach_observation_or_weaken_failure(bound, bi observation.reservation.fail(store, "failed", observation=forged) assert rows(store, "failed_turn_evidence") == [] assert store.snapshot("goal").lifecycle == BlockedGeneration() - assert store.failure_evidence(store.require_generation("goal")) == NoFailedTurnEvidence() + assert store.require_generation("goal").failure_evidence(store) == NoFailedTurnEvidence() def test_unclaimed_attempt_failure_has_no_backend_incident(bound): diff --git a/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py index 7c5813c933..7474b7a7ed 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -1,62 +1,66 @@ """Held-order recovery and unanswered DMs on an independently copied bus.""" -from contextlib import closing -from pathlib import Path -from dataclasses import replace -import os +import asyncio import json -import sqlite3 +import os import subprocess import sys import time +from contextlib import closing +from dataclasses import replace +from functools import partial +from pathlib import Path from uuid import uuid4 import pytest +from cutover.carry_goal_work_decision import RETIRED_TABLE, GoalWorkDecisions -from source_revision_fixture import SourceRepository -from conftest import declare_test_bus - -from agent_comms.bus_publication import AgentOrigin -from agent_comms.comms import Comms -from agent_comms.coordination_database import CoordinationStore from agent_comms import agent_events as events -import asyncio -from test_goal_execution import turn_on -from agent_comms.errors import RelationViolationError -from agent_comms.installer.carry import CarryChain, CarryLedger -from agent_comms.locked_store import SQLiteFile +from agent_comms.bus_publication import AgentOrigin +from agent_comms.child_process import ProcessIdentity from agent_comms.commander.config import CommanderConfig from agent_comms.commander.order import Commander, OrderFile, SayVerb, SendVerb from agent_comms.commander.standing import ( + Detached, + GoalCannotRun, HeartbeatSilent, - GoalCannotRun, Detached, HolderCannotRun, UnitAskUnanswered, UnownedWheelSpin, + HolderCannotRun, + UnitAskUnanswered, + UnownedWheelSpin, ) from agent_comms.commander_heartbeat import CommanderHeartbeat +from agent_comms.comms import Comms +from agent_comms.coordination_database import CoordinationStore +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, SetGoalAction -from agent_comms.goal_attempts import AttemptRecord, GoalAttemptStore, GoalWorkDecision -from agent_comms.goal_attempt_phase import ResolvedAttempt -from agent_comms.goal_generation import ReadyGeneration +from agent_comms.goal_attempts import GoalAttemptStore, GoalWorkDecision +from agent_comms.goal_presentation import RunnableExecution from agent_comms.goal_states import ActiveGoal from agent_comms.goals import CountCheckResult, Goal, OrderCheck, RequestGoalOrder from agent_comms.input_disposition import InputDispositions -from agent_comms.messages import InfoKind, Message +from agent_comms.installer.backup import StoreBackup +from agent_comms.installer.carry import CarryChain, CarryLedger +from agent_comms.launch_root import LaunchRoot +from agent_comms.locked_store import SQLiteFile from agent_comms.message_page import MessagePageRequest -from agent_comms.read_basis import DMDisplayScope -from agent_comms.field_codec import FieldCodec +from agent_comms.messages import InfoKind, Message +from agent_comms.native_arguments import GivenBinary from agent_comms.order_runs import MechanicalOrderRun +from agent_comms.read_basis import DMDisplayScope from agent_comms.recorded_time import RecordedEventTime from agent_comms.repository_name import RepositoryName from agent_comms.standing_rules import StandingRule +from agent_comms.thread_identity import CoreWatcherIncarnation +from agent_comms.thread_model import DeclaredModel from agent_comms.thread_status import StoppedThreadStatus -from agent_comms.installer.backup import StoreBackup -from agent_comms.child_process import ProcessIdentity from agent_comms.todos import ( - DeliverTodo, OperationDelivery, OrderMessage, - PendingProblemOrder, TodoConflict, + PendingProblemOrder, + TodoConflict, ) -from agent_comms.tool_caller import ThreadCaller -from agent_comms.tools import CommsOrderDeliverTool -from agent_comms.thread_identity import CoreWatcherIncarnation -from cutover.carry_goal_work_decision import GoalWorkDecisions, RETIRED_TABLE +from conftest import declare_test_bus +from native_backend_fixture import native_backend_fixture, wait_for +from test_goal_execution import turn_on @pytest.fixture @@ -134,79 +138,22 @@ def publish(comms, sender, target, timestamp, *, answers=()): ) -def test_blocked_holder_republishes_every_order_without_replaying_unknown(held_bus, comms): - reservation = blocked_holder(held_bus) - original = comms.registry.require("worker").goal.require() +def test_failed_holder_is_reported_without_reissuing_orders_or_replaying_unknown(held_bus): + blocked_holder(held_bus) + goal = held_bus.registry.require("worker").goal.require() + store = held_bus.goals.open_goal_store() + generation = store.require_generation(goal.id) + orders = held_bus.goals.todos.held_orders() inputs = InputDispositions(held_bus.root) unresolved = inputs.lookup("unresolved-input") violation, = GoalCannotRun.violations(held_bus, "lead") - previous_orders = {todo.id: todo.order.checked().id - for todo in held_bus.goals.todos.held_orders()} + assert violation.orders == ("P1", "P2") violation.enforce(held_bus, "lead") - current = held_bus.registry.require("worker").goal.require() - assert current.id == original.id and isinstance(current.state, ActiveGoal) - assigned = held_bus.goals.todos.for_goal(held_bus.registry.require("worker").incarnation, current.id) - assert {todo.id for todo in assigned} == {"P1", "P2"} - assert all(todo.order.checked().id != previous_orders[todo.id] for todo in assigned) - assert isinstance(held_bus.goals.open_goal_store().require_generation(current.id).lifecycle, - ReadyGeneration) - with closing(sqlite3.connect(held_bus.goals.open_goal_store().path)) as connection: - assert AttemptRecord.read_in(connection, reservation.attempt_id).holds(reservation, ResolvedAttempt()) + assert held_bus.registry.require("worker").goal == goal + assert held_bus.goals.todos.held_orders() == orders + assert store.require_generation(goal.id) == generation assert inputs.lookup("unresolved-input") == unresolved - assert comms.registry.require("worker").goal.require() == original - assert GoalCannotRun.violations(held_bus, "lead") == () - - -def test_blocked_holder_reissues_only_undelivered_orders(held_bus): - delivered = held_bus.goals.todos.get("P2") - delivered = held_bus.goals.todos.apply("P2", DeliverTodo( - delivered.revision, held_bus.registry.require("lead").incarnation, - OperationDelivery(at=RecordedEventTime(time.time())), delivered.order.id)) - blocked_holder(held_bus) - violation, = GoalCannotRun.violations(held_bus, "lead") - assert violation.orders == ("P1",) - violation.enforce(held_bus, "lead") - assert held_bus.goals.todos.get("P2") == delivered - - -def test_blocked_holder_reissues_pending_sibling_but_never_the_claimed_order(held_bus): - todo = held_bus.goals.todos.get("P2") - sent = held_bus.messaging.send("lead", "worker", todo.order.instructions(todo)) - held_bus.goals.todos.apply("P2", OrderMessage( - todo.revision, held_bus.registry.require("lead").incarnation, sent.message.message_id, - )) - ThreadCaller("worker").invoke(CommsOrderDeliverTool( - problem="P2", pull_request=1, head="a" * 40, - report="The source is published for the commander to inspect before accepting delivery.", - ), held_bus) - claimed = held_bus.goals.todos.get("P2") - pending = held_bus.goals.todos.get("P1") - blocked_holder(held_bus) - violation, = GoalCannotRun.violations(held_bus, "lead") - assert violation.orders == ("P1",) - violation.enforce(held_bus, "lead") - assert held_bus.goals.todos.get("P2") == claimed - assert held_bus.goals.todos.get("P1").order.id != pending.order.id - - -def test_blocked_holder_is_reissued_its_accepted_order_after_editing_its_scope(held_bus, tmp_path): - original = held_bus.goals.todos.get('P1').order.checked() - owner = Path(original.check.directory) / original.owner.file - owner.write_text(SourceRepository.main_source) - now = time.time() - held_bus.goals.update_goal("worker", CreateProblemGoalAction( - problem="P3", commander="lead", request="request", - repo=RepositoryName.from_wire("OpenHCSDev/agent-comms"), text="Repair P3", - review_at=RecordedEventTime(now + 3600), - order=replace(original, id='order-P3'), - )) - accepted = held_bus.goals.todos.get("P3").order.checked().id - owner.write_text('"""The unit moved its class while working the order."""\n\n\nclass Owner:\n pass\n') - blocked_holder(held_bus) - violation, = GoalCannotRun.violations(held_bus, "lead") - violation.enforce(held_bus, "lead") - assert held_bus.goals.todos.get("P3").order.checked().id != accepted - assert GoalCannotRun.violations(held_bus, "lead") == () + assert GoalCannotRun.violations(held_bus, "lead") == (violation,) def test_an_order_to_a_blocked_goal_activates_it_and_keeps_other_orders(held_bus): @@ -223,6 +170,38 @@ def test_an_order_to_a_blocked_goal_activates_it_and_keeps_other_orders(held_bus )} == {"P1", "P2"} +async def test_sigterm_group_preserves_active_goal_and_never_reissues_orders( + held_bus, native_runtime_package, source_programs, 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: + fixture.provider.status = 0 + fixture.provider.response_timeout = 60 + worker = held_bus.registry.require("worker") + held_bus.registry.register(replace(worker, model=DeclaredModel("response-local/fixture"))) + goal = held_bus.registry.require("worker").goal.require() + orders = held_bus.goals.todos.held_orders() + unresolved = InputDispositions(held_bus.root).lookup("unresolved-input") + with LaunchRoot.from_path(held_bus.root).selected(): + await Coordination.run_worker(partial( + held_bus.owners.start, "worker", agent_bin=GivenBinary("pi"))) + try: + await wait_for(lambda: fixture.provider.posts == 1) + assert held_bus.registry.require("worker").executing + await Coordination.run_worker(partial(held_bus.owners.stop, "worker")) + assert held_bus.registry.require("worker").goal == goal + assert isinstance(held_bus.goals.goal_execution("worker"), RunnableExecution) + for violation in GoalCannotRun.violations(held_bus, "lead"): + violation.enforce(held_bus, "lead") + assert held_bus.goals.todos.held_orders() == orders + assert InputDispositions(held_bus.root).lookup("unresolved-input") == unresolved + finally: + if held_bus.registry.require("worker").process_alive: + await Coordination.run_worker(partial(held_bus.owners.stop, "worker")) + + def test_goal_generation_cannot_transfer_to_a_different_owner(held_bus): todo = held_bus.goals.todos.get("P1") with pytest.raises(TodoConflict):