feat(runtime): add user-controlled turn recovery

tests/agent/test_loop_save_turn.py

@@ -36,6 +36,7 @@ from nanobot.session.keys import (
     UNIFIED_SESSION_KEY,
 )
 from nanobot.session.manager import Session
+from nanobot.session.recovery import PENDING_FOLLOWUP_ID_KEY, PENDING_FOLLOWUPS_KEY
 from nanobot.session.turn_continuation import (
     INTERNAL_CONTINUATION_META,
     INTERNAL_CONTINUATION_RUN_STARTED_AT_META,
@@ -161,6 +162,35 @@ def test_persist_cron_turn_uses_distinct_history_marker(tmp_path: Path) -> None:
     assert message["cron_prompt_ref"] == prompt_ref
 
 
+def test_persist_user_message_acknowledges_durable_followup(tmp_path: Path) -> None:
+    loop = _make_full_loop(tmp_path)
+    session = loop.sessions.get_or_create("websocket:chat")
+    session.metadata[PENDING_FOLLOWUPS_KEY] = [
+        {
+            "id": "followup-1",
+            "sender_id": "user",
+            "chat_id": "chat",
+            "content": "queued while busy",
+            "media": [],
+            "metadata": {},
+        }
+    ]
+
+    persisted = loop._persist_user_message_early(
+        InboundMessage(
+            channel="websocket",
+            sender_id="user",
+            chat_id="chat",
+            content="queued while busy",
+            metadata={PENDING_FOLLOWUP_ID_KEY: "followup-1"},
+        ),
+        session,
+    )
+
+    assert persisted is True
+    assert PENDING_FOLLOWUPS_KEY not in session.metadata
+
+
 def test_persist_local_trigger_turn_uses_hidden_automation_marker(tmp_path: Path) -> None:
     loop = _make_full_loop(tmp_path)
     session = loop.sessions.get_or_create("websocket:auto")
@@ -381,6 +411,34 @@ def test_save_turn_keeps_multimodal_runtime_context_for_model_replay() -> None:
     assert public_history_message(session.messages[0])["content"] == []
 
 
+def test_save_turn_acknowledges_every_merged_recovery_followup() -> None:
+    """Persisting a merged injected row retires every durable follow-up ID."""
+    loop = _mk_loop()
+    session = Session(
+        key="test:recovery-followups",
+        metadata={
+            PENDING_FOLLOWUPS_KEY: [
+                {"id": "first"},
+                {"id": "second"},
+            ]
+        },
+    )
+
+    loop._save_turn(
+        session,
+        [
+            {
+                "role": "user",
+                "content": "first\n\nsecond",
+                PENDING_FOLLOWUP_ID_KEY: ["first", "second"],
+            }
+        ],
+        skip=0,
+    )
+
+    assert PENDING_FOLLOWUPS_KEY not in session.metadata
+
+
 def test_save_turn_keeps_image_placeholder_and_runtime_context() -> None:
     loop = _mk_loop()
     session = Session(key="test:image")

tests/agent/test_runner_injections.py

@@ -26,7 +26,7 @@ def _make_injection_callback(queue: asyncio.Queue):
     return inject_cb
 
 
-def _make_loop(tmp_path):
+def _make_loop(tmp_path, *, recovery_admission=None):
     from nanobot.agent.loop import AgentLoop
     from nanobot.bus.queue import MessageBus
 
@@ -39,7 +39,12 @@ def _make_loop(tmp_path):
          patch("nanobot.agent.loop.SubagentManager") as mock_sub_mgr:
         mock_sub_mgr.return_value.cancel_by_session = AsyncMock(return_value=0)
         mock_sub_mgr.return_value.close = AsyncMock()
-        loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path)
+        loop = AgentLoop(
+            bus=bus,
+            provider=provider,
+            workspace=tmp_path,
+            recovery_admission=recovery_admission,
+        )
     return loop
 
 @pytest.mark.asyncio
@@ -759,6 +764,20 @@ async def test_runner_merges_multiple_injected_user_messages_without_losing_medi
     )
 
 
+def test_runner_merge_keeps_all_recovery_followup_ids() -> None:
+    """Merged follow-ups stay acknowledged together after a later save."""
+    from nanobot.agent.runner import AgentRunner
+    from nanobot.session.recovery import PENDING_FOLLOWUP_ID_KEY
+
+    messages = [{"role": "user", "content": "first", PENDING_FOLLOWUP_ID_KEY: "one"}]
+    AgentRunner._append_injected_messages(
+        messages,
+        [{"role": "user", "content": "second", PENDING_FOLLOWUP_ID_KEY: "two"}],
+    )
+
+    assert messages[-1][PENDING_FOLLOWUP_ID_KEY] == ["one", "two"]
+
+
 def test_runner_merge_preserves_runtime_markers_with_media() -> None:
     from nanobot.agent.runner import AgentRunner
     from nanobot.runtime_context import (
@@ -967,6 +986,38 @@ async def test_followup_routed_to_pending_queue(tmp_path):
     assert queued_msg.session_key == UNIFIED_SESSION_KEY
 
 
+@pytest.mark.asyncio
+async def test_websocket_followup_is_admitted_before_recovery_queue(tmp_path):
+    """Recovery admission runs before a newer WebUI message is injected."""
+    from nanobot.bus.events import InboundMessage
+
+    admission = MagicMock()
+    admission.admit = AsyncMock(return_value=True)
+    loop = _make_loop(tmp_path, recovery_admission=admission)
+    loop._dispatch = AsyncMock()  # type: ignore[method-assign]
+
+    session_key = "websocket:chat"
+    pending = asyncio.Queue(maxsize=20)
+    loop._pending_queues[session_key] = pending
+
+    run_task = asyncio.create_task(loop.run())
+    msg = InboundMessage(
+        channel="websocket",
+        sender_id="u",
+        chat_id="chat",
+        content="new request",
+    )
+    await loop.bus.publish_inbound(msg)
+
+    queued_msg = await asyncio.wait_for(pending.get(), timeout=2)
+    admission.admit.assert_awaited_once_with(msg)
+    assert queued_msg.content == msg.content
+    assert queued_msg.metadata["_recovery_followup_id"]
+
+    loop.stop()
+    await asyncio.wait_for(run_task, timeout=2)
+
+
 @pytest.mark.asyncio
 async def test_mid_turn_subagent_result_does_not_resolve_a_new_turn_route(tmp_path):
     """Injected results stay inside the active turn instead of opening a side turn."""
@@ -1314,6 +1365,51 @@ async def test_pending_queue_full_falls_back_to_queued_task(tmp_path):
     assert pending.qsize() == 1
 
 
+@pytest.mark.asyncio
+async def test_pending_queue_overflow_keeps_websocket_followup_durable(tmp_path):
+    """Fallback dispatch must not acknowledge a WebUI message before it commits."""
+    from nanobot.bus.events import InboundMessage
+    from nanobot.session.manager import Session
+    from nanobot.session.recovery import pending_followups
+
+    loop = _make_loop(tmp_path)
+    dispatched = asyncio.Event()
+    release_dispatch = asyncio.Event()
+
+    async def _dispatch(_msg):
+        dispatched.set()
+        await release_dispatch.wait()
+
+    loop._dispatch = AsyncMock(side_effect=_dispatch)  # type: ignore[method-assign]
+    session = Session(key="websocket:c")
+    loop.sessions.get_or_create.return_value = session
+    pending = asyncio.Queue(maxsize=1)
+    pending.put_nowait(
+        InboundMessage(channel="websocket", sender_id="u", chat_id="c", content="already queued")
+    )
+    loop._pending_queues["websocket:c"] = pending
+
+    run_task = asyncio.create_task(loop.run())
+    await loop.bus.publish_inbound(
+        InboundMessage(
+            channel="websocket",
+            sender_id="u",
+            chat_id="c",
+            content="durable follow-up",
+            metadata={"webui": True},
+        )
+    )
+    await asyncio.wait_for(dispatched.wait(), timeout=2)
+
+    assert [message.content for message in pending_followups(session)] == ["durable follow-up"]
+    dispatched_msg = loop._dispatch.await_args.args[0]
+    assert dispatched_msg.metadata["_recovery_followup_id"]
+
+    release_dispatch.set()
+    loop.stop()
+    await asyncio.wait_for(run_task, timeout=2)
+
+
 @pytest.mark.asyncio
 async def test_dispatch_republishes_leftover_queue_messages(tmp_path):
     """Messages left in the pending queue after _dispatch are re-published to the bus.

tests/agent/test_stop_preserves_context.py

@@ -161,3 +161,25 @@ async def test_dispatch_cancellation_restores_checkpoint():
         "Checkpoint metadata should be cleared after restore"
     assert loop.sessions.save.called, \
         "Session should be persisted so the restored state survives process restart"
+
+
+@pytest.mark.asyncio
+async def test_dispatch_cancellation_keeps_checkpoint_for_managed_restart(tmp_path: Path) -> None:
+    """A restart preserves the checkpoint; an explicit stop still restores it."""
+    loop = _make_loop(tmp_path)
+    loop.preserve_inflight_turns_on_shutdown()
+    loop._restore_runtime_checkpoint = MagicMock()  # type: ignore[method-assign]
+
+    async def _cancel(*_args: object, **_kwargs: object) -> None:
+        raise asyncio.CancelledError()
+
+    loop._process_message = _cancel  # type: ignore[method-assign]
+
+    from nanobot.bus.events import InboundMessage
+
+    with pytest.raises(asyncio.CancelledError):
+        await loop._dispatch(
+            InboundMessage(channel="test", sender_id="u1", chat_id="c1", content="work")
+        )
+
+    loop._restore_runtime_checkpoint.assert_not_called()