From 6564006049bc7787b48a1c130f36915b4d6d50ce Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 21 Oct 2026 13:01:17 -0400 Subject: [PATCH 1/4] Fix goal failure blocks to retain and settle their failed turn --- src/agent_comms/agent_events.py | 15 +++--- src/agent_comms/goal_management.py | 4 +- src/agent_comms/goal_states.py | 33 ++++++++++-- src/agent_comms/goal_turn_permit.py | 4 -- src/agent_comms/goals.py | 24 ++++++--- src/agent_comms/turn_goal_account.py | 27 +++++----- src/agent_comms/turn_terminal.py | 11 ++-- tests/test_goal_answers.py | 76 ++++++++++++++++++++++------ 8 files changed, 136 insertions(+), 58 deletions(-) diff --git a/src/agent_comms/agent_events.py b/src/agent_comms/agent_events.py index 56de62ee7c..8896110cec 100644 --- a/src/agent_comms/agent_events.py +++ b/src/agent_comms/agent_events.py @@ -278,19 +278,20 @@ def failure_reason(self) -> str: def failure_observation(self, account, permit, captured): return None if self.ok else super().failure_observation(account, permit, captured) - def observe_goal(self, account) -> None: - if not self.ok: - account.permit.block(account, "Backend turn failed; inspect local diagnostics before resuming.") - else: - account.require_output(self.text) + def verified(self, account) -> bool: + return self.ok and bool(self.text.strip() or account.has_productive_tool) + + def settle_current(self, account) -> None: + if self.verified(account): + VerifiedGoalSettlement.settle_current(account) def settle_launched(self, account, permit, store, captured, current) -> None: - if self.ok: + if self.verified(account): current.settle(VerifiedGoalSettlement, store, permit, account.turn) super().settle_launched(account, permit, store, captured, current) def settle_origin(self, account, permit, store, goal_id, current) -> None: - if self.ok: + if self.verified(account): permit.settle_goal(OriginGoalSettlement, store, current, account.turn) super().settle_origin(account, permit, store, goal_id, current) diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index b4ee37d15b..275f634c52 100644 --- a/src/agent_comms/goal_management.py +++ b/src/agent_comms/goal_management.py @@ -40,6 +40,7 @@ from .message_reference import MessageReference from .native_input_owner import RegistryOwner from .registry_document import RegistrySnapshot + from .thread_identity import TurnId from .goal_presentation import ( GoalPublicationSnapshot, GoalSnapshot, @@ -446,6 +447,7 @@ def block_goal_after_failed_turn( started_goal: Goal, expected_worktree: str, diagnostic: str, + turn: TurnId | None = None, ) -> GoalDeclaration: """Make an unresolved same-ID attempt visible without losing newer progress.""" with _store_lock(self._wire_lock_path): @@ -453,7 +455,7 @@ def block_goal_after_failed_turn( current = thread.goal.matching(started_goal.id) if thread.worktree != expected_worktree: return thread.goal - blocked = current.after_failed_turn(diagnostic) + blocked = current.after_failed_turn(diagnostic, turn) if blocked is current: return thread.goal self.registry.register(replace(thread, goal=blocked), self.registry.status(thread.name)) diff --git a/src/agent_comms/goal_states.py b/src/agent_comms/goal_states.py index 15a5ec54e3..0371f56e3b 100644 --- a/src/agent_comms/goal_states.py +++ b/src/agent_comms/goal_states.py @@ -9,6 +9,7 @@ from .declared_family import DeclaredFamily from .goal_attempt_identity import FailureNotObserved from .lifecycle import LifecycleState +from .thread_identity import TurnId if TYPE_CHECKING: from .goal_actions import GoalAction, GoalActionContext, ProblemOrderGoalAction @@ -52,7 +53,10 @@ class GoalState(DeclaredFamily, LifecycleState, affix="Goal"): active: ClassVar[bool] = False terminal: ClassVar[bool] = False - def after_failed_turn(self, diagnostic: str) -> GoalState: + def after_failed_turn(self, diagnostic: str, turn: TurnId | None = None) -> GoalState: + return self + + def after_verified_turn(self, captured: GoalState, turn: TurnId) -> GoalState: return self def accept_order(self, action: ProblemOrderGoalAction, goal: Goal, ctx: GoalActionContext) -> Goal: @@ -125,10 +129,10 @@ class FromOpenGoal: class BlockingOnFailure: """An eligible state owns the failed-turn transition and reason validation.""" - def after_failed_turn(self, diagnostic: str) -> GoalState: + def after_failed_turn(self, diagnostic: str, turn: TurnId | None = None) -> GoalState: from .goal_actions import required_block_reason - return BlockedGoal(required_block_reason(diagnostic)) + return BlockedGoal(required_block_reason(diagnostic), turn=turn) class OpenGoal(BlockingOnFailure, GoalState): @@ -156,10 +160,10 @@ def toggle(self) -> type[GoalAction]: class PausedGoal(OpenGoal, FromOpenGoal): source: PauseSource = field(default_factory=OwnerPause) - def after_failed_turn(self, diagnostic: str) -> GoalState: + def after_failed_turn(self, diagnostic: str, turn: TurnId | None = None) -> GoalState: if self.protected: return self - return super().after_failed_turn(diagnostic) + return super().after_failed_turn(diagnostic, turn) def failure_projection(self, reason: str) -> tuple[str, str]: return self.source.failure_projection() @@ -235,6 +239,25 @@ def transition_refusal(self) -> str: @dataclass(frozen=True) class BlockedGoal(BlockedState): block_reason: str + turn: TurnId | None = None + + @classmethod + def successors(cls) -> tuple[type[GoalState], ...]: + return (*super().successors(), ActiveGoal) + + def check_transition(self, nxt: GoalState, *, owner: bool) -> None: + super().check_transition(nxt, owner=owner) + if nxt.active: + raise ValueError(self.transition_refusal()) + + def after_verified_turn(self, captured: GoalState, turn: TurnId) -> GoalState: + # The captured state was read before this turn was allocated. Equality + # excludes blocks written during it; UUIDs themselves have no ordering. + if self.turn is not None and self == captured and self.turn != turn: + active = ActiveGoal() + if self.may_become(active): + return active + return self def __post_init__(self) -> None: reason = self.block_reason diff --git a/src/agent_comms/goal_turn_permit.py b/src/agent_comms/goal_turn_permit.py index 77802a382b..c03ddf9bb9 100644 --- a/src/agent_comms/goal_turn_permit.py +++ b/src/agent_comms/goal_turn_permit.py @@ -44,10 +44,6 @@ def stop(self, store: GoalAttemptStore) -> None: for permit in self.claims: permit.stop(store) - def block(self, account: TurnGoalAccount, diagnostic: str) -> None: - for permit in self.claims: - account.block_active_goal(diagnostic) - def settle_goal(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptStore, goal: GoalDeclaration, turn: TurnId) -> None: for permit in self.claims: diff --git a/src/agent_comms/goals.py b/src/agent_comms/goals.py index 806e3e049b..e68bdd625f 100644 --- a/src/agent_comms/goals.py +++ b/src/agent_comms/goals.py @@ -698,12 +698,15 @@ def settle(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptSto @abstractmethod def block_after_failed_turn(self, goals: Goals, name: str, worktree: str, - diagnostic: str) -> GoalDeclaration: ... + diagnostic: str, turn: TurnId | None = None) -> GoalDeclaration: ... @abstractmethod def instructions_for(self, goals: Goals, owner: ThreadIncarnation) -> tuple[str, ...]: ... - def after_failed_turn(self, diagnostic: str) -> GoalDeclaration: + def after_failed_turn(self, diagnostic: str, turn: TurnId | None = None) -> GoalDeclaration: + return self + + def after_verified_turn(self, captured: GoalDeclaration, turn: TurnId) -> GoalDeclaration: return self def check_expected(self, current: GoalDeclaration) -> None: @@ -795,7 +798,7 @@ def settle(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptSto pass def block_after_failed_turn(self, goals: Goals, name: str, worktree: str, - diagnostic: str) -> GoalDeclaration: + diagnostic: str, turn: TurnId | None = None) -> GoalDeclaration: return self def instructions_for(self, goals: Goals, owner: ThreadIncarnation) -> tuple[str, ...]: @@ -906,9 +909,9 @@ def settle(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptSto settlement(store, permit, self, turn).dispatch_sync(self.state) def block_after_failed_turn(self, goals: Goals, name: str, worktree: str, - diagnostic: str) -> GoalDeclaration: + diagnostic: str, turn: TurnId | None = None) -> GoalDeclaration: return goals.block_goal_after_failed_turn( - name, started_goal=self, expected_worktree=worktree, diagnostic=diagnostic, + name, started_goal=self, expected_worktree=worktree, diagnostic=diagnostic, turn=turn, ) def instructions_for(self, goals: Goals, owner: ThreadIncarnation) -> tuple[str, ...]: @@ -1018,9 +1021,16 @@ def captured(self) -> GoalCheckpoint: def render_instructions(self, orders_text: str) -> str: return "\n\n".join(part for part in (self.order.goal_text(self), orders_text) if part) - def after_failed_turn(self, diagnostic: str) -> Goal: + def after_verified_turn(self, captured: GoalDeclaration, turn: TurnId) -> Goal: + for original in captured.matching(self.id): + state = self.state.after_verified_turn(original.state, turn) + if state is not self.state: + return replace(self, state=state, revision=self.revision + 1) + return self + + def after_failed_turn(self, diagnostic: str, turn: TurnId | None = None) -> Goal: progress = self.order.failed_turn_progress(self, diagnostic) - state = self.state.after_failed_turn(diagnostic) + state = self.state.after_failed_turn(diagnostic, turn) if state is self.state: return self return replace(self, state=state, progress=progress, revision=self.revision + 1) diff --git a/src/agent_comms/turn_goal_account.py b/src/agent_comms/turn_goal_account.py index 97ecd83894..ab7a0ee6be 100644 --- a/src/agent_comms/turn_goal_account.py +++ b/src/agent_comms/turn_goal_account.py @@ -7,7 +7,7 @@ from __future__ import annotations from contextlib import ExitStack -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from typing import TYPE_CHECKING from acp import RequestError @@ -38,6 +38,18 @@ class VerifiedGoalSettlement(MroDispatch): goal: Goal turn: TurnId + @classmethod + def settle_current(cls, account: TurnGoalAccount) -> None: + """Publish the state's verified successor, including a turn without a grant.""" + comms, owner = account.comms, account.owner + with _store_lock(comms._wire_lock_path): + current = comms.registry.require(account.thread_name) + if current.incarnation != owner.incarnation or current.worktree != owner.worktree: + return + settled = current.goal.after_verified_turn(owner.goal, account.turn) + if settled is not current.goal: + comms.registry.register(replace(current, goal=settled), comms.registry.status(current.name)) + @property def progress_witness(self) -> str: return f"native-terminal:{self.turn.value}" @@ -159,18 +171,6 @@ def _record_origin_usage(self, permit: LaunchPermit) -> None: self.unattributed_usage.clear() self.pending_origins[self.thread_name] = permit.reservation.goal_id - def block_active_goal(self, diagnostic: str) -> None: - current = self.comms.registry.require(self.thread_name).continuation_goal(self.owner) - current.block_after_failed_turn( - self.comms.goals, self.thread_name, self.owner.worktree, diagnostic, - ) - - def require_output(self, text: str) -> None: - if not (self.has_productive_tool or text.strip()): - self.permit.block(self, - "Backend reported success without assistant output " - "or tool activity; inspect the session before resuming.") - def settle_launched(self, permit: LaunchPermit, terminal: TurnTerminal) -> None: store = self.comms.goals.open_goal_store() captured = self.owner.goal.require().require_id(permit.reservation.goal_id) @@ -179,6 +179,7 @@ def settle_launched(self, permit: LaunchPermit, terminal: TurnTerminal) -> None: def finish(self, terminal: TurnTerminal) -> None: """Settle all original grants once from the actual producer terminal event.""" + terminal.settle_current(self) self.permit.settle(self, terminal) for goal_id, permit in self.originated.items(): current = self.comms.registry.require(self.thread_name).goal.matching(goal_id) diff --git a/src/agent_comms/turn_terminal.py b/src/agent_comms/turn_terminal.py index 2a683b54e6..184c901f08 100644 --- a/src/agent_comms/turn_terminal.py +++ b/src/agent_comms/turn_terminal.py @@ -51,6 +51,9 @@ async def report_failure(self, progress: TurnProgress, error: Exception): detail = f"Native turn failed: {type(cause).__name__}: {cause}" return (await progress.record_failure(detail)).outcome + def settle_current(self, account: TurnGoalAccount) -> None: + pass + @abstractmethod def settle_launched(self, account, permit, store, captured, current) -> None: ... @@ -88,12 +91,12 @@ def settle_launched(self, account, permit, store, captured, current) -> None: observation = self.failure_observation(account, permit, captured) with suppress(StaleAttemptError): permit.reservation.fail( - store, "Goal turn ended without verified terminal progress.", + store, self.failure_reason, observation=observation, ) current.block_after_failed_turn( account.comms.goals, account.thread_name, account.owner.worktree, - "Goal turn ended without verified terminal progress.", + self.failure_reason, turn=account.turn, ) @@ -105,10 +108,6 @@ def failure_reason(self) -> FailureReason: def interrupted(self, terminal: TurnTerminal) -> TurnTerminal: return terminal - def settle_launched(self, account, permit, store, captured, current) -> None: - account.block_active_goal("Backend turn ended without a result; inspect local diagnostics.") - super().settle_launched(account, permit, store, captured, current) - class OperatorStopped(TurnTerminal): def settle_launched(self, account, permit, store, captured, current) -> None: diff --git a/tests/test_goal_answers.py b/tests/test_goal_answers.py index b4bff8ed40..3cfa1be31f 100644 --- a/tests/test_goal_answers.py +++ b/tests/test_goal_answers.py @@ -10,7 +10,7 @@ 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_actions import BlockedGoalAction, ClearGoalAction, SetGoalAction from agent_comms.goal_attempts import ProviderUsageTotal from agent_comms.goal_turn_permit import NoLaunchPermit from agent_comms.goal_states import BlockedGoal @@ -31,34 +31,80 @@ def copied_goal_bus(tmp_path): @contextmanager -def goal_turn(comms, original_goal): - comms.registry.declare(comms.registry.new_thread( - "worker", frozenset(), str(comms.root.parent), - process_identity=ProcessIdentity.capture(os.getpid()), - )) - if original_goal: - goal = comms.goals.update_goal("worker", SetGoalAction(text="Original work")) +def worker_turn(comms): turn = TurnId(uuid4().hex) owner = comms.agents.begin_turn("worker", turn.value) lease = owner.thread.require_turn_lease() try: with ExitStack() as claims: - permit = ( - TurnGoalAccount.reserve(owner=owner.thread, goal=goal, comms=comms, autonomous=False) - if original_goal else NoLaunchPermit() - ) + permit = NoLaunchPermit() + for goal in owner.thread.goal.active(): + permit = TurnGoalAccount.reserve( + owner=owner.thread, goal=goal, comms=comms, autonomous=False, + ) account = TurnGoalAccount( comms=comms, owner=owner.thread, turn=turn, lease=lease, permit=permit, pending_origins={}, claims=claims, ) - if not original_goal: - comms.goals.update_goal("worker", SetGoalAction(text="Work created in turn")) - account.tool_ended(events.ToolEnd("set", "comms_set_goal", True, "created")) yield account finally: comms.agents.finish_turn(lease) +@contextmanager +def goal_turn(comms, original_goal): + comms.registry.declare(comms.registry.new_thread( + "worker", frozenset(), str(comms.root.parent), + process_identity=ProcessIdentity.capture(os.getpid()), + )) + if original_goal: + comms.goals.update_goal("worker", SetGoalAction(text="Original work")) + with worker_turn(comms) as account: + if not original_goal: + comms.goals.update_goal("worker", SetGoalAction(text="Work created in turn")) + account.tool_ended(events.ToolEnd("set", "comms_set_goal", True, "created")) + yield account + + +def test_failed_goal_turn_writes_one_block_naming_turn(copied_goal_bus): + comms = copied_goal_bus + with goal_turn(comms, True) as account: + terminal = events.Done("backend failed", False) + terminal.observe_goal(account) + account.finish(terminal) + blocks = [row.after.require().state for row in comms.goals.goal_history("worker") + if isinstance(row.after.require().state, BlockedGoal)] + assert blocks == [BlockedGoal(terminal.failure_reason, turn=account.turn)] + + +def test_later_verified_turn_supersedes_failed_turn_block(copied_goal_bus): + comms = copied_goal_bus + with goal_turn(comms, True) as failed: + failed.finish(events.Done("backend failed", False)) + blocked = comms.registry.require("worker").goal.require() + with worker_turn(comms) as later: + assert later.turn != failed.turn + later.tool_ended(events.ToolEnd("work", "read", True, "read source")) + later.finish(events.Done("", True)) + active = Comms(comms.root).registry.require("worker").goal.require() + assert active.id == blocked.id + assert active.state.active + assert active.revision == blocked.revision + 1 + assert active.progress == blocked.progress + + +def test_goal_action_block_survives_later_verified_turn(copied_goal_bus): + comms = copied_goal_bus + with goal_turn(comms, True) as first: + first.finish(events.Done("first step", True)) + blocked = comms.goals.update_goal("worker", BlockedGoalAction(progress="Need owner decision")) + with worker_turn(comms) as later: + later.tool_ended(events.ToolEnd("work", "read", True, "read source")) + later.finish(events.Done("", True)) + assert Comms(comms.root).registry.require("worker").goal == blocked + assert blocked.state.turn is None + + @pytest.mark.parametrize("original_goal", [False, True]) @pytest.mark.parametrize("terminal", [ events.Done("finished", False), events.Done("finished", True), From ad5841ea28755ebff080e088971b4675fca71a49 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 21 Oct 2026 13:08:28 -0400 Subject: [PATCH 2/4] Store the terminal failure reason value on blocked goals --- src/agent_comms/agent_events.py | 4 ++-- src/agent_comms/turn_terminal.py | 5 +++-- tests/test_goal_answers.py | 2 +- 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/src/agent_comms/agent_events.py b/src/agent_comms/agent_events.py index 8896110cec..df2edc704b 100644 --- a/src/agent_comms/agent_events.py +++ b/src/agent_comms/agent_events.py @@ -22,7 +22,7 @@ ) from .turn_terminal import FailedGoalTurn from .turn_goal_account import OriginGoalSettlement, VerifiedGoalSettlement -from .diagnostics import terminal_failure_reason +from .diagnostics import FailureReason, terminal_failure_reason from .child_process import ProcessIdentity from .declared_family import DeclaredFamily from .field_codec import NullMember @@ -272,7 +272,7 @@ def resolve( return replace(self, outcome=outcome) @property - def failure_reason(self) -> str: + def failure_reason(self) -> FailureReason: return terminal_failure_reason(asdict(self)) def failure_observation(self, account, permit, captured): diff --git a/src/agent_comms/turn_terminal.py b/src/agent_comms/turn_terminal.py index 184c901f08..4373f41903 100644 --- a/src/agent_comms/turn_terminal.py +++ b/src/agent_comms/turn_terminal.py @@ -89,14 +89,15 @@ def settle_launched(self, account, permit, store, captured, current) -> None: if permit.has_verified_progress(store): return observation = self.failure_observation(account, permit, captured) + reason = self.failure_reason.value with suppress(StaleAttemptError): permit.reservation.fail( - store, self.failure_reason, + store, reason, observation=observation, ) current.block_after_failed_turn( account.comms.goals, account.thread_name, account.owner.worktree, - self.failure_reason, turn=account.turn, + reason, turn=account.turn, ) diff --git a/tests/test_goal_answers.py b/tests/test_goal_answers.py index 3cfa1be31f..cbeb0562da 100644 --- a/tests/test_goal_answers.py +++ b/tests/test_goal_answers.py @@ -74,7 +74,7 @@ def test_failed_goal_turn_writes_one_block_naming_turn(copied_goal_bus): account.finish(terminal) blocks = [row.after.require().state for row in comms.goals.goal_history("worker") if isinstance(row.after.require().state, BlockedGoal)] - assert blocks == [BlockedGoal(terminal.failure_reason, turn=account.turn)] + assert blocks == [BlockedGoal(terminal.failure_reason.value, turn=account.turn)] def test_later_verified_turn_supersedes_failed_turn_block(copied_goal_bus): From d7c23672ca86296515472d29508cb2c178a48395 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 21 Oct 2026 13:16:57 -0400 Subject: [PATCH 3/4] Settle original and in-turn goal failures through one terminal writer --- src/agent_comms/goal_attempts.py | 4 ---- src/agent_comms/goal_management.py | 7 ------- src/agent_comms/goal_turn_permit.py | 9 --------- src/agent_comms/turn_goal_account.py | 11 ----------- src/agent_comms/turn_progress.py | 1 - src/agent_comms/turn_terminal.py | 16 ++++++++-------- tests/test_goal_answers.py | 10 ++++------ 7 files changed, 12 insertions(+), 46 deletions(-) diff --git a/src/agent_comms/goal_attempts.py b/src/agent_comms/goal_attempts.py index d7685c6a3a..64ee364820 100644 --- a/src/agent_comms/goal_attempts.py +++ b/src/agent_comms/goal_attempts.py @@ -482,10 +482,6 @@ def charge_for(self, goal_id: str, goals: Goals, response: str, def add_context(self, context: TurnContext, owner: Thread, goals: Goals) -> TurnContext: return context.with_goal(owner, goals) - def fail_origin(self, store: GoalAttemptStore, goal_id: str) -> None: - with suppress(StaleAttemptError): - self.reservation.fail(store, "Goal origin turn did not finish successfully.") - def retire_unverified(self, store: GoalAttemptStore) -> None: """Release this acquired claim; settled or superseded attempts stay intact.""" with suppress(StaleAttemptError): diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index 275f634c52..4b710eb96f 100644 --- a/src/agent_comms/goal_management.py +++ b/src/agent_comms/goal_management.py @@ -15,7 +15,6 @@ from .coordination_errors import StaleFence from .errors import RelationViolationError from .goal_actions import ( - BlockedGoalAction, GoalAction, GoalActionContext, GoalPrecondition, @@ -434,12 +433,6 @@ def _updating_goal[Result: GoalDeclaration]( with _store_lock(self._wire_lock_path): yield action.apply(GoalActionContext(self, self.registry.require(name), actor)) - def block_failed_origin(self, name: str, goal: Goal) -> GoalDeclaration: - return self.update_goal(name, BlockedGoalAction( - expect=GoalPrecondition(goal_id=goal.id, expected_goal=goal), - progress="Goal origin turn did not finish successfully.", - )) - def block_goal_after_failed_turn( self, name: str, diff --git a/src/agent_comms/goal_turn_permit.py b/src/agent_comms/goal_turn_permit.py index c03ddf9bb9..4e12ef5355 100644 --- a/src/agent_comms/goal_turn_permit.py +++ b/src/agent_comms/goal_turn_permit.py @@ -52,9 +52,6 @@ def settle_goal(self, settlement: type[VerifiedGoalSettlement], store: GoalAttem @abstractmethod def has_verified_progress(self, store: GoalAttemptStore) -> bool: ... - @abstractmethod - def fail_origin(self, store: GoalAttemptStore, goal_id: str) -> None: ... - @abstractmethod def charge_for(self, goal_id: str, goals: Goals, response: str, usage: PiUsage) -> bool: ... @@ -75,9 +72,3 @@ def charge_for(self, goal_id: str, goals: Goals, response: str, def add_context(self, context: TurnContext, owner: Thread, goals: Goals) -> TurnContext: return context - - def fail_origin(self, store: GoalAttemptStore, goal_id: str) -> None: - generation = store.snapshot(goal_id) - if generation and generation.lifecycle.allows_resume(True): - store.retire_goal(goal_id, expected_generation=generation.number, - attempt_id=generation.attempt_id) diff --git a/src/agent_comms/turn_goal_account.py b/src/agent_comms/turn_goal_account.py index ab7a0ee6be..9a43662bf9 100644 --- a/src/agent_comms/turn_goal_account.py +++ b/src/agent_comms/turn_goal_account.py @@ -79,17 +79,6 @@ def completion_witness(self) -> str: return f"origin-completed:{self.goal.revision}" -@dataclass(frozen=True) -class FailedGoalOrigin(MroDispatch): - comms: Comms - thread_name: str - goal: Goal - - @handles(ActiveGoal, CompletedGoal) - def block(self, state) -> None: - self.comms.goals.block_failed_origin(self.thread_name, self.goal) - - @dataclass(kw_only=True) class TurnGoalAccount: comms: Comms diff --git a/src/agent_comms/turn_progress.py b/src/agent_comms/turn_progress.py index b84d0932ca..1fd7466b93 100644 --- a/src/agent_comms/turn_progress.py +++ b/src/agent_comms/turn_progress.py @@ -198,7 +198,6 @@ async def done(self, event: events.Done) -> events.Done: # Retain the resolved outcome before transport publication. If delivery # throws, failure handling must not classify or resolve it a second time. self.turn.terminal = event - await Coordination.run_worker(partial(event.observe_goal, self.goals)) return event @handles(events.InputStarted) diff --git a/src/agent_comms/turn_terminal.py b/src/agent_comms/turn_terminal.py index 4373f41903..870d2c0279 100644 --- a/src/agent_comms/turn_terminal.py +++ b/src/agent_comms/turn_terminal.py @@ -10,7 +10,6 @@ from .diagnostics import FailureReason, terminal_failure_reason from .goal_failure_observation import FailedTurnObservation from .goal_attempts import StaleAttemptError -from .turn_goal_account import FailedGoalOrigin if TYPE_CHECKING: from .agent_events import Done @@ -34,9 +33,6 @@ def from_cancellation(error: asyncio.CancelledError) -> TurnTerminal: def interrupted(self, terminal: TurnTerminal) -> TurnTerminal: return self - def observe_goal(self, account: TurnGoalAccount) -> None: - pass - def reported(self, event: Done) -> Done: return event @@ -58,15 +54,19 @@ def settle_current(self, account: TurnGoalAccount) -> None: def settle_launched(self, account, permit, store, captured, current) -> None: ... def settle_origin(self, account, permit, store, goal_id, current) -> None: - if not permit.has_verified_progress(store): - permit.fail_origin(store, goal_id) - for goal in current: - FailedGoalOrigin(account.comms, account.thread_name, goal).dispatch_sync(goal.state) + pass class FailedGoalTurn(TurnTerminal): """Reported failures and missing results share the original failed ledger path.""" + def settle_origin(self, account, permit, store, goal_id, current) -> None: + if permit.has_verified_progress(store): + return + for launch in permit.claims: + for goal in current.matching(goal_id): + self.settle_launched(account, launch, store, goal, current) + def failure_observation(self, account, permit, captured): snapshot = account.comms.registry.snapshot() return FailedTurnObservation.from_terminal( diff --git a/tests/test_goal_answers.py b/tests/test_goal_answers.py index cbeb0562da..be1911ac20 100644 --- a/tests/test_goal_answers.py +++ b/tests/test_goal_answers.py @@ -66,11 +66,11 @@ def goal_turn(comms, original_goal): yield account -def test_failed_goal_turn_writes_one_block_naming_turn(copied_goal_bus): +@pytest.mark.parametrize("original_goal", [False, True]) +@pytest.mark.parametrize("terminal", [events.Done("backend failed", False), EndedWithoutResult()]) +def test_failed_goal_turn_writes_one_block_naming_turn(copied_goal_bus, original_goal, terminal): comms = copied_goal_bus - with goal_turn(comms, True) as account: - terminal = events.Done("backend failed", False) - terminal.observe_goal(account) + with goal_turn(comms, original_goal) as account: account.finish(terminal) blocks = [row.after.require().state for row in comms.goals.goal_history("worker") if isinstance(row.after.require().state, BlockedGoal)] @@ -117,7 +117,6 @@ def test_clear_during_original_or_origin_turn_does_not_block_missing_goal( with goal_turn(comms, original_goal) as account: goal = comms.registry.require("worker").goal.require() comms.goals.update_goal("worker", ClearGoalAction()) - terminal.observe_goal(account) account.finish(terminal) assert comms.registry.require("worker").goal == NoGoal() assert comms.goals.open_goal_store().snapshot(goal.id).lifecycle.terminal @@ -130,7 +129,6 @@ def test_clear_during_original_or_origin_turn_does_not_block_missing_goal( def test_unsuccessful_terminal_still_blocks_goal_on_copied_bus(copied_goal_bus, terminal): comms = copied_goal_bus with goal_turn(comms, True) as account: - terminal.observe_goal(account) account.finish(terminal) state = comms.registry.require("worker").goal.require().state assert isinstance(state, BlockedGoal) From 000b10bf99100b9bcc65e3cae7972948b5598985 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 21 Oct 2026 15:13:59 -0400 Subject: [PATCH 4/4] Checkpoint superseded failed-turn subtype correction for P448 --- src/agent_comms/goal_management.py | 2 +- src/agent_comms/goal_presentation.py | 2 +- src/agent_comms/goal_states.py | 39 ++++++++++------------- src/agent_comms/goals.py | 10 +++--- src/agent_comms/schedule_rules.py | 14 ++++---- tests/conftest.py | 7 ++++ tests/test_backend_failure_owner_pause.py | 12 +++---- tests/test_goal_answers.py | 10 ++++-- tests/test_goal_block_reason.py | 3 +- 9 files changed, 53 insertions(+), 46 deletions(-) diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index 4b710eb96f..db28ee3fe8 100644 --- a/src/agent_comms/goal_management.py +++ b/src/agent_comms/goal_management.py @@ -440,7 +440,7 @@ def block_goal_after_failed_turn( started_goal: Goal, expected_worktree: str, diagnostic: str, - turn: TurnId | None = None, + turn: TurnId, ) -> GoalDeclaration: """Make an unresolved same-ID attempt visible without losing newer progress.""" with _store_lock(self._wire_lock_path): diff --git a/src/agent_comms/goal_presentation.py b/src/agent_comms/goal_presentation.py index 3cb58bd0db..73c42428b5 100644 --- a/src/agent_comms/goal_presentation.py +++ b/src/agent_comms/goal_presentation.py @@ -50,7 +50,7 @@ class StateExecutionPresentation(ExecutionPresentation): state: type[GoalState] def represents(self, state: GoalState) -> bool: - return state.__class__ is self.state + return isinstance(state, self.state) def render(self, execution: GoalExecution) -> tuple[str, str]: return FieldCodec.decode( diff --git a/src/agent_comms/goal_states.py b/src/agent_comms/goal_states.py index 0371f56e3b..bfb60ec6fc 100644 --- a/src/agent_comms/goal_states.py +++ b/src/agent_comms/goal_states.py @@ -53,7 +53,7 @@ class GoalState(DeclaredFamily, LifecycleState, affix="Goal"): active: ClassVar[bool] = False terminal: ClassVar[bool] = False - def after_failed_turn(self, diagnostic: str, turn: TurnId | None = None) -> GoalState: + def after_failed_turn(self, diagnostic: str, turn: TurnId) -> GoalState: return self def after_verified_turn(self, captured: GoalState, turn: TurnId) -> GoalState: @@ -129,10 +129,10 @@ class FromOpenGoal: class BlockingOnFailure: """An eligible state owns the failed-turn transition and reason validation.""" - def after_failed_turn(self, diagnostic: str, turn: TurnId | None = None) -> GoalState: + def after_failed_turn(self, diagnostic: str, turn: TurnId) -> GoalState: from .goal_actions import required_block_reason - return BlockedGoal(required_block_reason(diagnostic), turn=turn) + return FailedTurnBlockedGoal(required_block_reason(diagnostic), turn=turn) class OpenGoal(BlockingOnFailure, GoalState): @@ -160,7 +160,7 @@ def toggle(self) -> type[GoalAction]: class PausedGoal(OpenGoal, FromOpenGoal): source: PauseSource = field(default_factory=OwnerPause) - def after_failed_turn(self, diagnostic: str, turn: TurnId | None = None) -> GoalState: + def after_failed_turn(self, diagnostic: str, turn: TurnId) -> GoalState: if self.protected: return self return super().after_failed_turn(diagnostic, turn) @@ -239,25 +239,6 @@ def transition_refusal(self) -> str: @dataclass(frozen=True) class BlockedGoal(BlockedState): block_reason: str - turn: TurnId | None = None - - @classmethod - def successors(cls) -> tuple[type[GoalState], ...]: - return (*super().successors(), ActiveGoal) - - def check_transition(self, nxt: GoalState, *, owner: bool) -> None: - super().check_transition(nxt, owner=owner) - if nxt.active: - raise ValueError(self.transition_refusal()) - - def after_verified_turn(self, captured: GoalState, turn: TurnId) -> GoalState: - # The captured state was read before this turn was allocated. Equality - # excludes blocks written during it; UUIDs themselves have no ordering. - if self.turn is not None and self == captured and self.turn != turn: - active = ActiveGoal() - if self.may_become(active): - return active - return self def __post_init__(self) -> None: reason = self.block_reason @@ -283,6 +264,18 @@ def presentation(self) -> tuple[str, str]: return "!", f"Blocked ยท {summary}" +@dataclass(frozen=True) +class FailedTurnBlockedGoal(BlockedGoal): + turn: TurnId + + def after_verified_turn(self, captured: GoalState, turn: TurnId) -> GoalState: + # Captured before this turn was allocated: equality excludes blocks + # written during it, without inventing ordering for UUIDs. + if self == captured and self.turn != turn: + return ActiveGoal() + return self + + @dataclass(frozen=True) class UnrecordedBlockGoal(BlockedState): """A recorded blocked status whose reason was not recorded.""" diff --git a/src/agent_comms/goals.py b/src/agent_comms/goals.py index 8b57691c74..7d09fcd2a3 100644 --- a/src/agent_comms/goals.py +++ b/src/agent_comms/goals.py @@ -698,12 +698,12 @@ def settle(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptSto @abstractmethod def block_after_failed_turn(self, goals: Goals, name: str, worktree: str, - diagnostic: str, turn: TurnId | None = None) -> GoalDeclaration: ... + diagnostic: str, turn: TurnId) -> GoalDeclaration: ... @abstractmethod def instructions_for(self, goals: Goals, owner: ThreadIncarnation) -> tuple[str, ...]: ... - def after_failed_turn(self, diagnostic: str, turn: TurnId | None = None) -> GoalDeclaration: + def after_failed_turn(self, diagnostic: str, turn: TurnId) -> GoalDeclaration: return self def after_verified_turn(self, captured: GoalDeclaration, turn: TurnId) -> GoalDeclaration: @@ -798,7 +798,7 @@ def settle(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptSto pass def block_after_failed_turn(self, goals: Goals, name: str, worktree: str, - diagnostic: str, turn: TurnId | None = None) -> GoalDeclaration: + diagnostic: str, turn: TurnId) -> GoalDeclaration: return self def instructions_for(self, goals: Goals, owner: ThreadIncarnation) -> tuple[str, ...]: @@ -909,7 +909,7 @@ def settle(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptSto settlement(store, permit, self, turn).dispatch_sync(self.state) def block_after_failed_turn(self, goals: Goals, name: str, worktree: str, - diagnostic: str, turn: TurnId | None = None) -> GoalDeclaration: + diagnostic: str, turn: TurnId) -> GoalDeclaration: return goals.block_goal_after_failed_turn( name, started_goal=self, expected_worktree=worktree, diagnostic=diagnostic, turn=turn, ) @@ -1028,7 +1028,7 @@ def after_verified_turn(self, captured: GoalDeclaration, turn: TurnId) -> Goal: return replace(self, state=state, revision=self.revision + 1) return self - def after_failed_turn(self, diagnostic: str, turn: TurnId | None = None) -> Goal: + def after_failed_turn(self, diagnostic: str, turn: TurnId) -> Goal: progress = self.order.failed_turn_progress(self, diagnostic) state = self.state.after_failed_turn(diagnostic, turn) if state is self.state: diff --git a/src/agent_comms/schedule_rules.py b/src/agent_comms/schedule_rules.py index 59c966d05f..350a9a9acc 100644 --- a/src/agent_comms/schedule_rules.py +++ b/src/agent_comms/schedule_rules.py @@ -11,6 +11,7 @@ from .drain_failures import drain_failures from .activity import FailedWakeDiagnostic from .errors import RelationViolationError +from .goal_actions import BlockedGoalAction, GoalPrecondition from .reservation_rules import ReservationRule, RuleCheck, ReservationViolationError from .routing import ScheduledTurn @@ -116,13 +117,14 @@ async def run(self) -> None: except RelationViolationError as missing: raise error from missing pending[0].require_dependency_wake(error) - await Coordination.run_worker(partial(inputs.comms.goals.block_goal_after_failed_turn, + await Coordination.run_worker(partial(inputs.comms.goals.update_goal, owner.name, - started_goal=started_goal, - expected_worktree=owner.worktree, - diagnostic=( - "Standby wake launch authority unavailable; inspect the UNKNOWN " - "input before explicit Retry. No input was replayed." + BlockedGoalAction( + expect=GoalPrecondition(expected_goal=started_goal), + block_reason=( + "Standby wake launch authority unavailable; inspect the UNKNOWN " + "input before explicit Retry. No input was replayed." + ), ), )) await inputs.effects.turns.goals.sync_goal_execution(session_id, owner.name) diff --git a/tests/conftest.py b/tests/conftest.py index 6fec971325..f6be97978b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,6 +26,13 @@ from agent_comms.messages import ConversationRole, InfoKind, Message from agent_comms.store_files import _store_lock from agent_comms.thread_text import ForkedParent, GivenThreadText +from agent_comms.thread_identity import TurnId + + +@pytest.fixture +def failed_turn() -> TurnId: + """A failure identity for direct goal-state transition checks.""" + return TurnId(uuid.uuid4().hex) class TranscriptRecords: diff --git a/tests/test_backend_failure_owner_pause.py b/tests/test_backend_failure_owner_pause.py index b68756e349..497c8a73ae 100644 --- a/tests/test_backend_failure_owner_pause.py +++ b/tests/test_backend_failure_owner_pause.py @@ -14,7 +14,7 @@ @pytest.mark.parametrize("attribution", ["model", "runtime"]) -def test_nonowner_pause_does_not_bypass_failure_block(wired, tmp_path, attribution): +def test_nonowner_pause_does_not_bypass_failure_block(wired, tmp_path, attribution, failed_turn): wired.registry.declare(wired.registry.new_thread(name='project', tags=frozenset(), worktree=str(tmp_path))) initial = wired.goals.update_goal("project", SetGoalAction(text="Work")) if attribution == "model": @@ -30,14 +30,14 @@ def test_nonowner_pause_does_not_bypass_failure_block(wired, tmp_path, attributi "project", PausedGoalAction(expect=GoalPrecondition(goal_id=initial.id)) ) blocked = wired.goals.block_goal_after_failed_turn( - "project", started_goal=initial, expected_worktree=str(tmp_path), diagnostic="Failed" + "project", started_goal=initial, expected_worktree=str(tmp_path), diagnostic="Failed", turn=failed_turn, ) - assert blocked.state.declared_name == "blocked" + assert blocked.state.turn == failed_turn assert blocked.revision == paused.revision + 1 assert blocked.state.reason == "Failed" -def test_stale_owner_pause_does_not_bypass_failure_block(wired, tmp_path): +def test_stale_owner_pause_does_not_bypass_failure_block(wired, tmp_path, failed_turn): wired.registry.declare(wired.registry.new_thread(name='project', tags=frozenset(), worktree=str(tmp_path))) initial = wired.goals.update_goal("project", SetGoalAction(text="Work")) wired.goals.update_goal( @@ -54,8 +54,8 @@ def test_stale_owner_pause_does_not_bypass_failure_block(wired, tmp_path): "project", PausedGoalAction(expect=GoalPrecondition(goal_id=initial.id)) ) blocked = wired.goals.block_goal_after_failed_turn( - "project", started_goal=initial, expected_worktree=str(tmp_path), diagnostic="Failed" + "project", started_goal=initial, expected_worktree=str(tmp_path), diagnostic="Failed", turn=failed_turn, ) - assert blocked.state.declared_name == "blocked" + assert blocked.state.turn == failed_turn assert blocked.revision == paused.revision + 1 assert blocked.state.reason == "Failed" diff --git a/tests/test_goal_answers.py b/tests/test_goal_answers.py index be1911ac20..25aebf0743 100644 --- a/tests/test_goal_answers.py +++ b/tests/test_goal_answers.py @@ -13,7 +13,7 @@ from agent_comms.goal_actions import BlockedGoalAction, ClearGoalAction, SetGoalAction from agent_comms.goal_attempts import ProviderUsageTotal from agent_comms.goal_turn_permit import NoLaunchPermit -from agent_comms.goal_states import BlockedGoal +from agent_comms.goal_states import ActiveGoal, BlockedGoal, FailedTurnBlockedGoal from agent_comms.goals import NoGoal from agent_comms.pi_payloads import PiUsage from agent_comms.thread_identity import TurnId @@ -74,7 +74,8 @@ def test_failed_goal_turn_writes_one_block_naming_turn(copied_goal_bus, original account.finish(terminal) blocks = [row.after.require().state for row in comms.goals.goal_history("worker") if isinstance(row.after.require().state, BlockedGoal)] - assert blocks == [BlockedGoal(terminal.failure_reason.value, turn=account.turn)] + assert blocks == [FailedTurnBlockedGoal(terminal.failure_reason.value, turn=account.turn)] + assert comms.goals.goal_execution("worker").presentation("worker").summary.startswith("Blocked") def test_later_verified_turn_supersedes_failed_turn_block(copied_goal_bus): @@ -82,6 +83,9 @@ def test_later_verified_turn_supersedes_failed_turn_block(copied_goal_bus): with goal_turn(comms, True) as failed: failed.finish(events.Done("backend failed", False)) blocked = comms.registry.require("worker").goal.require() + assert not blocked.state.may_become(ActiveGoal()) + with pytest.raises(ValueError): + blocked.state.check_transition(ActiveGoal(), owner=True) with worker_turn(comms) as later: assert later.turn != failed.turn later.tool_ended(events.ToolEnd("work", "read", True, "read source")) @@ -102,7 +106,7 @@ def test_goal_action_block_survives_later_verified_turn(copied_goal_bus): later.tool_ended(events.ToolEnd("work", "read", True, "read source")) later.finish(events.Done("", True)) assert Comms(comms.root).registry.require("worker").goal == blocked - assert blocked.state.turn is None + assert blocked.state == BlockedGoal("Need owner decision") @pytest.mark.parametrize("original_goal", [False, True]) diff --git a/tests/test_goal_block_reason.py b/tests/test_goal_block_reason.py index 42ff88a8da..fbedb2a3c4 100644 --- a/tests/test_goal_block_reason.py +++ b/tests/test_goal_block_reason.py @@ -64,7 +64,7 @@ def test_missing_or_blank_block_reason_rejects_without_side_effects(tmp_path, mo assert (tmp_path / "registry.json").read_bytes() == before_registry -def test_automatic_blocks_record_diagnostic_not_prior_progress(tmp_path): +def test_automatic_blocks_record_diagnostic_not_prior_progress(tmp_path, failed_turn): comms, started = _owner(tmp_path) progressed = comms.goals.update_goal("worker", ActiveGoalAction(progress="Verified step")) assert progressed is not None @@ -73,6 +73,7 @@ def test_automatic_blocks_record_diagnostic_not_prior_progress(tmp_path): started_goal=started, expected_worktree=str(tmp_path), diagnostic="Attempt outcome uncertain; inspect before Retry.", + turn=failed_turn, ) assert blocked is not None assert blocked.progress == "Verified step\n\nAttempt outcome uncertain; inspect before Retry."