P448: generation-owned failure evidence and report-only blocked work

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)
 

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)

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)

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

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
 
 

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