Keep failed goal blocks with their turn and settle verified successors

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,25 +272,26 @@ 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):
         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)
 

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):

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,
@@ -40,6 +39,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,
@@ -433,27 +433,22 @@ 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,
         *,
         started_goal: Goal,
         expected_worktree: str,
         diagnostic: str,
+        turn: TurnId,
     ) -> GoalDeclaration:
         """Make an unresolved same-ID attempt visible without losing newer progress."""
         with _store_lock(self._wire_lock_path):
             thread = self.registry.require(name)
             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))

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(

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) -> 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) -> GoalState:
         from .goal_actions import required_block_reason
 
-        return BlockedGoal(required_block_reason(diagnostic))
+        return FailedTurnBlockedGoal(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) -> 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()
@@ -260,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."""

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:
@@ -56,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: ...
@@ -79,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)

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) -> 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) -> 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) -> 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) -> 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) -> 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)

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)

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}"
@@ -67,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
@@ -159,18 +160,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 +168,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)

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)