Make turn recovery progress-aware

README.md

@@ -365,9 +365,9 @@ The controller owns cadence, durable events, conversation selection, verified Gi
 - The advisor keeps one conversation UUID under the pod-local `/var/lib/senpai/<tag>/advisor/openhands_state`; it survives controller and container restarts within that pod.
 - A student uses one UUID per assignment revision; feedback, monitor events, and child-task results resume that exact conversation.
 - Still-actionable GitHub state is re-delivered on the configured reminder cadence, which defaults to at least ten minutes even when GitHub is polled more frequently. Immediate post-turn polls deliver changed state but not timed reminders, so a successful research-only turn cannot enter a no-sleep reminder loop. `research_base_changed` is keyed by assignment, revision, PR head, and the exact required/current base pair; each identity or base movement requires a new decision. Merge repeats the live-base check immediately before its mutation, while external base writers still require strict up-to-date branch protection or a merge queue for an atomic guarantee.
-- Each model request gets one bounded 15-minute attempt. Foreground terminal calls return control within ten minutes for explicit continuation, the whole turn retains its one-hour hard lease, and two consecutive failed turns exit to the supervisor for a clean worker restart. Restart backoff grows across failed workers to a five-minute ceiling; only a successfully acknowledged turn resets that streak, not process uptime or idle sleep.
+- Each model request gets one bounded 15-minute attempt. Foreground terminal calls return control within ten minutes for explicit continuation, and the whole turn retains its one-hour hard lease. Reaching that lease is a resumable `paused_timeout`: Senpai preserves the delivered turn, polls other ready work, and continues without counting a failure. Two genuine consecutive failures still exit to the supervisor for a clean worker restart. Restart backoff grows across failed workers to a five-minute ceiling; only a successfully acknowledged turn resets that supervisor streak, not process uptime or idle sleep.
 - Every controller prompt, GitHub event, monitor signal, and child result follows one durable `pending -> delivered -> processed` inbox. A provider failure resumes the already-delivered turn without resending it, and a crash after inference performs mailbox acknowledgement without another model call. New events wait behind an unresolved turn in the same conversation; normal drains are FIFO and bounded to 16 events or 64 KiB, while ready conversations take fair turns.
-- A newly completed tool observation renews the consecutive retry budget; timeout, error, interruption, state, and delivery events do not. A persisted final response is reconciled even if cancellation left the SDK status paused. After three no-progress attempts or three total hours, Senpai preserves the raw trace and retries one canonical copy on a fresh branch with the complete research brief. If that recovery exhausts the same budget, the turn is durably quarantined, reported as `SENPAI_TURN_QUARANTINED` on every controller start, and excluded from scheduling rather than entering a restart loop. `SENPAI_INBOX_MAX_STALLED_ATTEMPTS`, `SENPAI_INBOX_MAX_TURN_AGE_SECONDS`, and `SENPAI_INBOX_MAX_RECOVERY_GENERATIONS` configure these positive attempt/age limits and the non-negative number of fresh branches.
+- A newly completed successful tool observation resets the consecutive retry and recovery budgets and renews the stall clock at the observation's persisted timestamp; failed tool calls, timeouts, errors, interruptions, state changes, and delivery events do not. A persisted final response is reconciled even if cancellation left the SDK status paused. After three no-progress attempts or three hours since the last model-visible progress, Senpai preserves the raw trace and retries one canonical copy on a fresh branch with the complete research brief. If that recovery exhausts the same no-progress budget, only that turn is durably quarantined as an auditable dead letter. Later student and GitHub events continue on one preserved fresh model branch, preceded by a typed quarantine notice, rather than being starved or re-entering the poisoned context. `SENPAI_TURN_QUARANTINED` reports the dead letter on controller start, while a quarantine with queued later events is published in the controller lease as degraded-but-live health and logged once as `SENPAI_HEALTH_DEGRADED`. `SENPAI_INBOX_MAX_STALLED_ATTEMPTS`, `SENPAI_INBOX_MAX_TURN_AGE_SECONDS`, and `SENPAI_INBOX_MAX_RECOVERY_GENERATIONS` configure these positive attempt/age limits and the non-negative number of consecutive no-progress fresh branches.
 - A typed context-window or malformed-history failure uses the same bounded fresh-branch recovery. The reset and its canonical recovery copy are durable across crashes; transient failures remain unacknowledged and retry after at least ten minutes.
 - On restart, an incomplete persisted tool action is rejected rather than replayed implicitly. A checked-out assignment branch that was deliberately rebased or extended locally is preserved and surfaced to its existing student conversation for explicit reconciliation.
 - The complete OpenHands event log remains locally searchable. Senpai does not prune conversation directories; operators own retention.

SPEC.md

@@ -105,20 +105,22 @@ batches; immediate post-turn polls drain later batches without dropping them.
 While an OpenHands turn is running, `ActiveGitHubWatcher` polls the same GitHub
 state. It enqueues all newly visible advisor events, and only PR feedback bound
 to the currently running student UUID, in the role's local event store.
-OpenHands 1.40 supports concurrent `send_message`; `AdvisorEventPump` injects at
-its state lock boundary without cancelling unrelated work. Successfully
-injected student feedback is acknowledged in `github-feedback.json` only when
-the enclosing student turn succeeds.
+`AdvisorEventPump` transfers those events into the persistent delivery inbox;
+it does not change the immutable membership of an unresolved turn. The events
+therefore become model-visible, in order and exactly once, on the next turn
+boundary. Successfully processed student feedback is acknowledged in
+`github-feedback.json` only when its durable turn succeeds.
 
 Generic child results use a local SQLite WAL event store because parent and
 child run on the same advisor or student instance. That is not an inter-node
 protocol.
 
-The only SQLite databases are `advisor-events.sqlite3`, for unacknowledged
-advisor watcher/child events; `student-events.sqlite3`, for unacknowledged
-student feedback/child events; and `training/monitors.sqlite3`, for student
-monitor policy, samples, and deduplicated actionable signals. OpenHands
-conversation history is a separate file-backed per-UUID event log.
+The SQLite databases are `advisor-events.sqlite3`, for unacknowledged advisor
+watcher/child events; `student-events.sqlite3`, for unacknowledged student
+feedback/child events; `delivery-inbox.sqlite3`, for stable turn membership and
+delivery receipts; and `training/monitors.sqlite3`, for student monitor policy,
+samples, and deduplicated actionable signals. OpenHands conversation history is
+a separate file-backed per-UUID event log.
 
 ## State and conversations
 
@@ -129,6 +131,7 @@ Advisor state:
 ├── advisor-conversation-id
 ├── controller-lease.json
 ├── advisor-events.sqlite3
+├── delivery-inbox.sqlite3
 ├── conversation-state.json
 ├── github/
 └── conversations managed by OpenHands
@@ -145,6 +148,7 @@ Student state:
 ├── github-feedback.json
 ├── student-conversations.json
 ├── student-events.sqlite3
+├── delivery-inbox.sqlite3
 ├── conversation-state.json
 ├── training/
 │   ├── <training-id>.json
@@ -174,9 +178,19 @@ the previous `started-conversations.json` and
 legacy files' two writes resumes without replaying its initial brief and
 receives the current system context once.
 
-OpenHands stores base state and individual events beneath that UUID. A killed
-worker resumes from the last persisted event. An in-flight response or tool
-call without a durable event is retried from the preceding event.
+OpenHands stores base state and individual events beneath that UUID. The
+delivery inbox advances each bounded turn monotonically through `pending`,
+`delivered`, and `processed`. A killed worker resumes an already-delivered turn
+without appending its messages again; a processed turn reconciles mailbox
+acknowledgements without another model call. New events remain queued behind an
+unresolved turn in the same conversation.
+
+Successful tool observations renew the turn's stall clock at their persisted
+event timestamp and reset its consecutive recovery allowance. When a turn
+exhausts that no-progress allowance, Senpai preserves and quarantines only that
+turn. Later events continue on a fresh active branch with one typed quarantine
+notice; the dead-letter trace remains available for audit. A quarantined turn
+with queued later events is published as degraded-but-live controller health.
 
 The controller marks a conversation's initial instructions delivered and
 records its current system-context digest in the same atomic update, only after
@@ -591,10 +605,12 @@ Denied patterns include raw GitHub mutations, raw `git push`, direct training
 launches, sleeps, polling loops, `watch`, and `tail -f`, including nested shell
 and `env` wrappers.
 
-Every OpenHands turn has a controller-configured hard deadline. The deadline
-interrupts the conversation, produces a non-success result, and leaves durable
-events unacknowledged. The controller then retries with bounded exponential
-backoff. Controller termination interrupts and closes the current conversation,
+Every OpenHands turn has a controller-configured hard deadline. Reaching that
+deadline interrupts the current run as a resumable `paused_timeout`, preserves
+the delivered turn, and lets the controller poll and serve other ready
+conversations before continuing it without replay. Provider errors and other
+genuine failures remain failures and use bounded exponential backoff.
+Controller termination interrupts and closes the current conversation,
 cancels active supervised training, closes local stores, and flushes Weave
 before the controller exits. Standalone and child runners flush Weave at runner
 exit.

senpai_agent/controller.py

@@ -50,6 +50,7 @@
     WorkspaceDivergenceLedger,
 )
 from senpai_agent.supervisor import LEASE_ENV, ProgressLease
+from senpai_agent.turns import TurnOutcome
 from senpai_agent.workspace import StudentWorkspaceReconciler, WorkspaceDivergence
 
 
@@ -75,6 +76,10 @@ class TurnResult:
     exit_code: int
     delivered_event_keys: frozenset[str] = frozenset()
 
+    @property
+    def outcome(self) -> TurnOutcome:
+        return TurnOutcome.from_exit_code(self.exit_code)
+
 
 class ConversationRecoveryExhausted(RuntimeError):
     def __init__(self, conversation_id: UUID, error: Exception):
@@ -354,6 +359,7 @@ def __init__(
         self._deferred_until: dict[str, float] = {}
         self._deferred_conversations: dict[UUID, float] = {}
         self._workspace_divergence: dict[UUID, str] = {}
+        self._reported_degraded_backlog: tuple[int, int] | None = None
 
     def run(self, *, max_cycles: int | None = None) -> None:
         self._wait_for_start_gates()
@@ -422,6 +428,7 @@ def run(self, *, max_cycles: int | None = None) -> None:
                         file=sys.stderr,
                         flush=True,
                     )
+                    self._publish_progress("quarantine")
                     continue
                 except Exception as error:  # noqa: BLE001
                     failures = turn_failures.get(conversation_id, 0) + 1
@@ -439,7 +446,19 @@ def run(self, *, max_cycles: int | None = None) -> None:
                     failed_conversations.add(conversation_id)
                     cycle_had_failure = True
                     continue
-                if result.exit_code != 0:
+                if result.outcome is TurnOutcome.PAUSED_TIMEOUT:
+                    turn_failures.pop(conversation_id, None)
+                    served_conversations.add(conversation_id)
+                    print(
+                        "SENPAI_TURN_PAUSED_TIMEOUT "
+                        f"conversation_id={conversation_id} "
+                        f"turn_id={turn.turn_id}",
+                        file=sys.stderr,
+                        flush=True,
+                    )
+                    self._poll_into_inbox(allow_reminders=False)
+                    continue
+                if result.outcome is TurnOutcome.FAILED:
                     failures = turn_failures.get(conversation_id, 0) + 1
                     turn_failures[conversation_id] = failures
                     print(
@@ -500,6 +519,7 @@ def _poll_into_inbox(self, *, allow_reminders: bool = True) -> None:
             allow_reminders=allow_reminders,
         )
         self._enqueue_events(events)
+        self._publish_progress("poll")
 
     def _enqueue_events(self, events: Sequence[ControllerEvent]) -> None:
         for batch in self._event_batches(events):
@@ -647,11 +667,27 @@ def _publish_progress(
         *,
         completed_turn: bool = False,
     ) -> None:
+        quarantined_turns, pending_events = self.inbox.degraded_backlog()
+        degraded_backlog = (quarantined_turns, pending_events)
+        if degraded_backlog != self._reported_degraded_backlog:
+            if quarantined_turns and pending_events:
+                print(
+                    "SENPAI_HEALTH_DEGRADED "
+                    f"quarantined_turns={quarantined_turns} "
+                    f"pending_events={pending_events}",
+                    file=sys.stderr,
+                    flush=True,
+                )
+            elif self._reported_degraded_backlog not in (None, (0, 0)):
+                print("SENPAI_HEALTH_RECOVERED", file=sys.stderr, flush=True)
+            self._reported_degraded_backlog = degraded_backlog
         if self.progress is not None:
             self.progress.update(
                 phase,
                 timeout_seconds or self.operation_timeout_seconds,
                 completed_turn=completed_turn,
+                quarantined_turns=quarantined_turns,
+                pending_events=pending_events,
             )
 
     def _sleep(self, phase: str, seconds: float) -> None: