fix(sessions): keep a running continuation out of the recoverable state

A transport failure that arrived after the runner had reported its outcome
demoted a running execution back to recoverable. The card then rendered
"Answer saved, retry needed" over a turn that was under way or already
finished. Two executions of the browser pass reached terminal/completed and
still carried continuation_delivery_failed.

_mark_continuation_recoverable now passes expected_states, so only an
execution that still waits for a runner can become recoverable. The caller
reports recoverable only when the write applied. A projection that raises
still counts, because the transport failure that brought us there is real.

Two smaller corrections travel with it. _deliver answers with an "exhausted"
receipt when the bounded delivery budget is spent, so the card says "Send your
next message to retry it" instead of a promise of a redelivery that no longer
happens. And resume_recoverable_continuation settles an exhausted command
through the existing exhaustion path before the reopen, so a Send that arrives
before the sweep retargets a fresh execution instead of doing nothing.

Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk

Author: mmabrouk

api/oss/src/core/sessions/commands/interfaces.py

@@ -35,9 +35,12 @@ class DeliveryReceipt(BaseModel):
       settlement sweep recovers it.
     * `not_held` — a reachable runner said it does not hold that session, which lets the
       service settle at once instead of waiting for the deadline.
+    * `exhausted` — the bounded delivery budget is spent, so the transport was never called.
+      Nothing retries the command on its own after this, which is why it is not `unreachable`:
+      the caller must tell the user the truth rather than promise a redelivery.
     """
 
-    status: str  # "accepted" | "unreachable" | "not_held"
+    status: str  # "accepted" | "unreachable" | "not_held" | "exhausted"
     detail: Optional[str] = None
     # Which runner process took it, when the transport learned that. The service uses it as the
     # claim owner, so the outcome route's guard reads the same way on every transport.

api/oss/src/core/sessions/commands/service.py

@@ -634,26 +634,62 @@ async def respond_interactions(
                 )
                 receipt = None
             if receipt is None or receipt.status != "accepted":
-                await self._mark_continuation_recoverable(admission)
-                admission.execution_state = SessionExecutionState.recoverable
+                if await self._mark_continuation_recoverable(admission, receipt):
+                    admission.execution_state = SessionExecutionState.recoverable
         return admission
 
     async def _mark_continuation_recoverable(
-        self, admission: InteractionContinuationAdmission
-    ) -> None:
+        self,
+        admission: InteractionContinuationAdmission,
+        receipt: Optional[DeliveryReceipt] = None,
+    ) -> bool:
+        """Project a failed delivery onto the execution the card reads.
+
+        Two guards, both learned from a browser pass where a delivered continuation was reported
+        `unreachable` anyway.
+
+        `expected_states` is the important one. A transport that fails AFTER the runner reported
+        its outcome would otherwise demote a `running` execution back to `recoverable`, telling
+        the user to retry a turn that is running underneath the card. Only an execution still
+        waiting for a runner may be turned recoverable.
+
+        The message is the other. It is what the card renders, so it must not promise a
+        redelivery that cannot happen: once the delivery budget is spent, nothing redelivers this
+        command on its own and only the user's next Send does (`resume_recoverable_continuation`
+        reopens the budget).
+
+        False means the DAO REFUSED, which happens only when the execution has moved on, so the
+        caller must not report `recoverable`. A projection that raises returns True: the write is
+        best effort, but the transport failure that brought us here is real and the user still
+        owns the retry.
+        """
         if self._executions is None or admission.command is None:
-            return
+            return False
+        exhausted = receipt is not None and receipt.status == "exhausted"
         try:
-            await self._executions.set_state(
+            applied = await self._executions.set_state(
                 project_id=admission.command.project_id,
                 session_id=admission.command.session_id,
                 execution_id=admission.execution_id,
                 state=SessionExecutionState.recoverable,
                 error={
-                    "code": "continuation_delivery_failed",
+                    "code": (
+                        "continuation_delivery_exhausted"
+                        if exhausted
+                        else "continuation_delivery_failed"
+                    ),
                     "retryable": True,
-                    "message": "The continuation is durable and awaiting redelivery.",
+                    "message": (
+                        "The continuation could not be delivered. Send your next message to "
+                        "retry it."
+                        if exhausted
+                        else "The continuation is durable and awaiting redelivery."
+                    ),
                 },
+                expected_states=[
+                    SessionExecutionState.pending_delivery,
+                    SessionExecutionState.recoverable,
+                ],
             )
         except Exception as error:  # noqa: BLE001 - recovery projection is best effort
             log.error(
@@ -662,6 +698,8 @@ async def _mark_continuation_recoverable(
                 admission.execution_id,
                 error,
             )
+            return True
+        return applied is not None
 
     async def resume_recoverable_continuation(
         self, *, project_id: UUID, session_id: str
@@ -690,6 +728,32 @@ async def resume_recoverable_continuation(
             # `recoverable`, after it has collapsed and tombstoned the old ownership. Until
             # then this durable continuation still owns Send, but it is never redelivered.
             return True
+        if (
+            command.state
+            in (
+                SessionCommandState.pending,
+                SessionCommandState.claimed,
+            )
+            and command.claim_count >= env.agenta.sessions.commands.max_deliveries
+        ):
+            # The budget bounds the AUTOMATIC retry loop, not the user. A command that spent it
+            # is undeliverable until the sweep settles it exhausted, so a Send arriving inside
+            # that window would deliver nothing and re-render a card asking for another Send.
+            # Settle it here instead. Redelivering it as it stands is not an option: the budget
+            # is spent precisely because this execution id keeps being refused, so the ending has
+            # to be recorded before the reopen below can retarget a fresh one.
+            if await self._settle_exhausted_continuation(command):
+                refreshed = await self._dao.fetch_command(command_id=command.id)
+                if refreshed is not None:
+                    command = refreshed
+                execution = (
+                    await self._executions.fetch_execution(
+                        project_id=project_id,
+                        session_id=session_id,
+                        execution_id=execution_id,
+                    )
+                    or execution
+                )
         if command.state not in (
             SessionCommandState.pending,
             SessionCommandState.claimed,
@@ -719,7 +783,7 @@ async def resume_recoverable_continuation(
             )
             receipt = None
         if receipt is None or receipt.status != "accepted":
-            await self._mark_continuation_recoverable(admission)
+            await self._mark_continuation_recoverable(admission, receipt)
         return True
 
     async def _reopen_continuation_attempt(
@@ -858,18 +922,36 @@ async def _deliver(self, command: SessionCommand) -> Optional[DeliveryReceipt]:
         """Hand the command to the transport, then record what the transport learned.
 
         Never raises. The user's request has already succeeded by the time this runs.
+
+        An `exhausted` receipt means the bounded delivery budget is spent. Nothing redelivers the
+        command after that, so the caller must say the true thing on the card rather than promise
+        a redelivery: only the user's next Send reopens the budget.
         """
+        maximum = env.agenta.sessions.commands.max_deliveries
+        requested = command
         try:
             command = await self._dao.record_delivery_attempt(
-                project_id=command.project_id,
-                command_id=command.id,
+                project_id=requested.project_id,
+                command_id=requested.id,
                 now=datetime.now(timezone.utc),
-                max_deliveries=env.agenta.sessions.commands.max_deliveries,
+                max_deliveries=maximum,
             )
         except Exception as error:  # noqa: BLE001 - delivery bookkeeping is post-commit
             log.warning("control delivery reservation failed: %s", error)
             return None
         if command is None:
+            if requested.claim_count >= maximum:
+                log.warning(
+                    "control delivery budget exhausted for command=%s session=%s after %s "
+                    "attempts",
+                    requested.id,
+                    requested.session_id,
+                    requested.claim_count,
+                )
+                return DeliveryReceipt(
+                    status="exhausted",
+                    detail=f"delivery budget of {maximum} attempts is spent",
+                )
             return None
 
         try:

api/oss/tests/pytest/unit/sessions/test_interaction_continuation_admission.py

@@ -1034,3 +1034,129 @@ async def test_recovery_hooks_are_disabled_with_durable_approvals(monkeypatch):
     )
     assert await service.settle_abandoned_commands(now=datetime.now(timezone.utc)) == 0
     assert delivery.delivered == []
+
+
+class _StartedThenUnreachable:
+    """The runner admits the continuation and reports `started`, then the transport fails.
+
+    The real shape of it (browser pass, 2026-09-04 17:28Z-17:35Z): the runner posted its
+    outcome, the API turned the execution `running`, and only afterwards did the detached-start
+    parser reject the stream. Both executions ran to completion carrying
+    `error.code = continuation_delivery_failed`, so the card offered a retry for work that was
+    already done.
+    """
+
+    def __init__(self, executions, execution_id):
+        self._executions = executions
+        self._execution_id = execution_id
+        self.delivered = []
+
+    async def deliver(self, **kwargs):
+        command = kwargs["command"]
+        self.delivered.append(command)
+        await self._executions.set_state(
+            project_id=command.project_id,
+            session_id=command.session_id,
+            execution_id=self._execution_id,
+            state=SessionExecutionState.running,
+            error=None,
+        )
+        return DeliveryReceipt(
+            status="unreachable", detail="parser rejected the stream"
+        )
+
+    async def acknowledge(self, **kwargs):
+        return None
+
+
+@pytest.mark.asyncio
+async def test_a_late_delivery_failure_does_not_demote_a_running_continuation():
+    project_id = uuid4()
+    interaction_id = uuid4()
+    interaction = SessionInteraction(
+        id=interaction_id,
+        project_id=project_id,
+        session_id="session-1",
+        turn_id="source-1",
+        token="approval-1",
+        kind=SessionInteractionKind.user_approval,
+        status=SessionInteractionStatus.responded,
+        data=SessionInteractionData(resolution={"approved": True}),
+    )
+    commands = _Commands()
+    commands.command = _continuation_command(project_id, interaction_id)
+    executions = _Executions(
+        project_id=project_id, session_id="session-1", source_id="source-1"
+    )
+    delivery = _StartedThenUnreachable(executions, "continuation-1")
+    service = SessionCommandsService(
+        commands_dao=commands,
+        streams_service=None,
+        interactions_service=_Interactions(interaction),
+        lock_engine=None,
+        delivery=delivery,
+        executions_dao=executions,
+    )
+
+    resumed = await service.resume_recoverable_continuation(
+        project_id=project_id, session_id="session-1"
+    )
+
+    assert resumed is True
+    assert delivery.delivered
+    # The turn the runner is already running keeps `running`. The recoverable projection is
+    # refused, so the card never asks the user to retry work that is under way.
+    assert executions.continuation.state == SessionExecutionState.running
+    assert executions.continuation.error is None
+
+
+@pytest.mark.asyncio
+async def test_a_send_after_the_budget_is_spent_reopens_the_continuation(monkeypatch):
+    """Command 01a06d7a of the same pass: three refusals, then a command nothing redelivers.
+
+    The budget bounds the automatic loop only. A Send arriving before the sweep settles the
+    command must not be swallowed: settle it exhausted, retarget a fresh execution, deliver.
+    """
+    maximum = 3
+    monkeypatch.setattr(env.agenta.sessions.commands, "max_deliveries", maximum)
+    project_id = uuid4()
+    interaction_id = uuid4()
+    interaction = SessionInteraction(
+        id=interaction_id,
+        project_id=project_id,
+        session_id="session-1",
+        turn_id="source-1",
+        token="approval-1",
+        kind=SessionInteractionKind.user_approval,
+        status=SessionInteractionStatus.responded,
+        data=SessionInteractionData(resolution={"approved": True}),
+    )
+    commands = _Commands()
+    commands.command = _continuation_command(
+        project_id, interaction_id, claim_count=maximum
+    )
+    spent = commands.command
+    executions = _Executions(
+        project_id=project_id, session_id="session-1", source_id="source-1"
+    )
+    delivery = _Unreachable()
+    service = SessionCommandsService(
+        commands_dao=commands,
+        streams_service=None,
+        interactions_service=_Interactions(interaction),
+        lock_engine=None,
+        delivery=delivery,
+        executions_dao=executions,
+    )
+
+    resumed = await service.resume_recoverable_continuation(
+        project_id=project_id, session_id="session-1"
+    )
+
+    assert resumed is True
+    # The exhausted attempt is recorded as ended and the command now targets a NEW execution:
+    # redelivering the old id is what spent the budget in the first place.
+    assert commands.command.target_turn_id != spent.target_turn_id
+    assert commands.command.claim_count == 0
+    assert delivery.delivered
+    assert delivery.delivered[0].target_turn_id == commands.command.target_turn_id