Preserve expected Retry refusals across owner RPC

src/agent_comms/errors.py

@@ -1,3 +1,3 @@
 """Errors: declaration and persistence owners."""
 
 from __future__ import annotations
@@ -123,12 +123,16 @@
         return self.diagnostic_text
 
 
-class UnresolvedAttemptError(GoalAttemptError):
+class GoalAttemptRefused(GoalRefused, GoalAttemptError):
+    """An expected attempt refusal keeps its ledger identity and RPC refusal contract."""
+
+
+class UnresolvedAttemptError(GoalAttemptRefused):
     """A blocked or in-flight attempt needs an explicit resolution."""
 
     diagnostic_text = "Goal attempt unresolved; inspect diagnostics before Retry."
 
 
-class StaleAttemptError(GoalAttemptError):
+class StaleAttemptError(GoalAttemptRefused):
     """A late outcome cannot modify the current goal generation."""
 

src/agent_comms/goal_attempts.py

@@ -104,11 +104,11 @@ class StorageUncertainError(errors.GoalAttemptError):
     """A write or durability/readback check failed; never launch from it."""
 
 
-class ReservationConflictError(errors.GoalAttemptError):
+class ReservationConflictError(errors.GoalAttemptRefused):
     """This goal/generation is already reserved or is not ready."""
 
 
-class MissingGoalGenerationError(errors.GoalAttemptError, ValueError):
+class MissingGoalGenerationError(errors.GoalAttemptRefused):
     """Historical goal has no launch record; Retry must not invent authority."""
 
     def __init__(self, goal_id: GoalId):

src/agent_comms/goal_generation.py

@@ -11,7 +11,7 @@
 from agent_comms.goal_turn_permit import GoalTurnPermit, NoLaunchPermit
 
 from .nominal.declared_family import DeclaredFamily
-from .errors import UnresolvedAttemptError
+from .errors import GoalRefused, UnresolvedAttemptError
 from .nominal.field_codec import AttemptId, NoAttemptId
 from .goal_attempt_phase import (
     FailedAttempt,
@@ -43,7 +43,7 @@ class GenerationState(DeclaredFamily, LifecycleState, affix="Generation"):
 
     def resume_status(self, generation: Generation, goal: Goal, ctx: GoalActionContext) -> None:
         if not self.allows_resume(ctx.thread.executing):
-            raise ValueError("The goal attempt is unresolved; inspect it before Retry.")
+            raise GoalRefused("The goal attempt is unresolved; inspect it before Retry.")
 
     def execution(self, source: GoalExecutionSource) -> GoalExecution:
         return source.turn.execution(source)
@@ -144,7 +144,7 @@ class ReservedGeneration(GenerationState):
     def authorize_retry(
         self, store: GoalAttemptStore, generation: Generation, decision: WorkDecision
     ) -> None:
-        raise ValueError(
+        raise GoalRefused(
             f"The goal generation is reserved for attempt {generation.attempt_identity.attempt_id}; "
             "Retry cannot replace it. Inspect that attempt before deciding on new work."
         )
@@ -167,7 +167,7 @@ def allows_resume(self, active_turn: bool) -> bool:
 class BlockedGeneration(Failed, GenerationState):
     def resume_status(self, generation: Generation, goal: Goal, ctx: GoalActionContext) -> None:
         """A status-only Resume cannot authorize an uncertain attempt."""
-        raise ValueError(
+        raise GoalRefused(
             "The interrupted goal attempt is unresolved. Inspect it, then use "
             "Retry to authorize a new attempt. Your messages can still be sent."
         )
@@ -209,7 +209,7 @@ class CompletedGeneration(Complete, Terminal, GenerationState):
     def authorize_retry(
         self, store: GoalAttemptStore, generation: Generation, decision: WorkDecision
     ) -> None:
-        raise ValueError("The goal generation is completed; Retry cannot restart completed work.")
+        raise GoalRefused("The goal generation is completed; Retry cannot restart completed work.")
 
     def execution(self, source: GoalExecutionSource) -> GoalExecution:
         return CompletedExecution(source.goal.id)
@@ -220,7 +220,7 @@ class CancelledGeneration(Terminal, GenerationState):
     def authorize_retry(
         self, store: GoalAttemptStore, generation: Generation, decision: WorkDecision
     ) -> None:
-        raise ValueError("The goal generation is cancelled; Retry cannot restart cancelled work.")
+        raise GoalRefused("The goal generation is cancelled; Retry cannot restart cancelled work.")
 
     def validate_attempt(self, attempt_id: AttemptId | NoAttemptId) -> None:
         """Retirement preserves either the recorded attempt or its declared absence."""

src/agent_comms/goal_scheduler.py

@@ -6,7 +6,7 @@
 
 from __future__ import annotations
 
-from .errors import GoalAttemptError
+from .errors import GoalAttemptError, GoalRefused
 import asyncio
 import os
 from collections.abc import Iterable
@@ -223,7 +223,7 @@ async def retry_goal(self, session_id: AcpSessionId, goal_id: GoalId, expected_r
         thread = await Coordination.run_worker(partial(self.comms.registry.require, name))
         goal = thread.goal.require_checkpoint(GoalRevision(goal_id, expected_revision))
         if self.pending_goal_origins.get(name) == goal_id:
-            raise ValueError("Wait for the goal origin turn to finish.")
+            raise GoalRefused("Wait for the goal origin turn to finish.")
         resumed = await Coordination.run_worker(lambda: self.comms.goals.update_goal(
             name,
             RetryGoalAction(

tests/test_goal_attempts.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from agent_comms.repo import Checkout
-from agent_comms.errors import StaleAttemptError
+from agent_comms.errors import GoalRefused, StaleAttemptError
 import multiprocessing
 import os
 import sqlite3
@@ -302,6 +302,27 @@ def test_verified_completion_is_terminal_without_a_new_ready_grant(store):
         store.ready_grant(goal_id, completed.number + 1)
 
 
+@pytest.mark.parametrize("prepare", (
+    pytest.param(lambda store, reservation: None, id="reserved-unclaimed"),
+    pytest.param(lambda store, reservation: store.claim_launch(reservation), id="reserved-claimed"),
+    pytest.param(lambda store, reservation: store.claim_launch(reservation).record_verified_completion(
+        store, "completion-witness"), id="completed"),
+    pytest.param(lambda store, reservation: store.retire_goal(
+        reservation.goal_id, expected_generation=reservation.generation,
+        attempt_id=reservation.attempt_id), id="cancelled"),
+))
+def test_retry_refusal_preserves_the_generation_and_owner_rpc_contract(store, prepare):
+    goal_id = GoalId.mint()
+    store.create_goal(goal_id)
+    reservation = store.reserve(goal_id, 1)
+    prepare(store, reservation)
+    before = store.snapshot(goal_id)
+    with pytest.raises(GoalRefused):
+        store.authorize_current_retry(goal_id, RetryDecision(RetryDecisionId.mint()))
+    assert store.snapshot(goal_id) == before
+    assert GoalAttemptStore.at(store.root).snapshot(goal_id) == before
+
+
 def test_provider_reported_responses_are_attributed_once_and_survive_reopen(store):
     goal_id = GoalId.mint()
     store.create_goal(goal_id)

tests/test_owner_lines.py

@@ -1,6 +1,12 @@
 """Both directions of the owner socket are declared families, decoded once."""
 
 
+import pytest
+
+from agent_comms.coordination_errors import Refused
+from agent_comms.errors import GoalAttemptError, StaleAttemptError, UnresolvedAttemptError
+from agent_comms.goal_attempts import MissingGoalGenerationError, ReservationConflictError, StorageUncertainError
+from agent_comms.nominal.field_codec import GoalId
 from agent_comms.runtime import ClientPermissionResponse
 
 from agent_comms.owner_protocol import (
@@ -36,3 +42,27 @@ def test_every_declared_line_round_trips_through_its_family():
         assert ClientLine.read(line.wire()) == line
 
 
+@pytest.mark.parametrize("refusal", (
+    UnresolvedAttemptError("The attempt is unresolved."),
+    StaleAttemptError("The attempt changed."),
+    ReservationConflictError("The generation is reserved."),
+    MissingGoalGenerationError(GoalId.mint()),
+))
+def test_expected_attempt_refusals_keep_their_rpc_contract(refusal):
+    assert isinstance(refusal, GoalAttemptError)
+    decoded = OwnerLine.read(OwnerFailure.refused(refusal).wire())
+    with pytest.raises(Refused) as raised:
+        decoded.reply()
+    assert raised.value.code == refusal.code
+    assert raised.value.message == refusal.message
+
+
+def test_storage_uncertainty_is_not_an_invalid_goal_request():
+    failure = StorageUncertainError("Write outcome is uncertain.")
+    assert isinstance(failure, GoalAttemptError)
+    assert not isinstance(failure, Refused)
+    decoded = OwnerLine.read(OwnerFailure.of(failure).wire())
+    with pytest.raises(RuntimeError, match="Write outcome is uncertain"):
+        decoded.reply()
+
+