fix: preserve unaccepted work while keeping active-turn controls reachable

docs/acp.md

@@ -60,9 +60,18 @@ assert adapter.config.command == ("codex-acp",)
   MCP servers remain outside SDK registration control.
 - **Approved permissions are silent.** Only a denied request posts a synthetic
   `tool_call`/`tool_result` pair, and only when `Emit.TOOL_CALLS` is on.
-- **Replay happens once**, only for a freshly minted session (a failed `session/load`
-  counts), under a nonce'd boundary marker so a replayed message cannot spoof it. History
-  stops strictly before the triggering message (`messages_before`).
+- **Definitively rejected prompts stay pending.** The exact OMP JSON-RPC busy
+  response (`-32003` with `data.reason == "session_busy"`) raises
+  `InputNotAccepted` after one attempt. It preserves the room's runtime and
+  session without cancellation or a failure event; the shared runtime schedules
+  later attempts. Other errors retain normal failure cleanup. Session-scoped
+  external notifications cannot complete the rejected request: their successful
+  effects commit only after a prompt closes successfully.
+- **Bootstrap commits after successful prompt completion.** A rejection keeps
+  room context owed. Fresh sessions also keep transcript replay owed; restored
+  sessions already hold their history and receive only Band context. Replay uses
+  a nonce'd boundary marker so history cannot spoof the live message, and stops
+  strictly before the triggering message (`messages_before`).
 
 ## Upgrading from SDK 4.x to 5.x
 

docs/adapters/claude_sdk.md

@@ -111,3 +111,11 @@ with pytest.raises(ValueError, match="wait_timeout_s must be less than"):
 
 A config is plain data, so a host can load it from YAML or JSON with
 `ClaudeSDKAdapterConfig.model_validate(data)`.
+
+An ordinary input arriving while accepted work owns the session remains
+outstanding. Approval and status commands stay reachable through the shared
+control hook under the existing sender and token checks. The active turn keeps
+its original tools; handling a control never rebinds them to the new message.
+Only a local rejection before `query()` qualifies as `InputNotAccepted`:
+transport errors and timeouts do not establish that Claude rejected the input.
+The shared room scheduler owns later attempts.

docs/adapters/codex.md

@@ -49,3 +49,9 @@ For existing integrations, follow the [SDK 4.x to 5.x migration](../acp.md#upgra
   with `skills/extraRoots/set` before the room's thread starts. An older CLI
   rejects the request, and the room fails to start rather than running without
   the skills. Roots must be absolute paths; Codex accepts ones that don't exist.
+
+While an accepted turn awaits approval, newer ordinary room work remains
+outstanding. The runtime can deliver approval and status controls ahead of that
+pending work through Codex's existing authorization rules. The active turn keeps
+its original tools; an ordinary input rejected before provider invocation is
+retried by the shared room scheduler after that turn releases the session.

docs/adapters/opencode.md

@@ -38,6 +38,13 @@ while the human decides), so the adapter judges that turn itself when OpenCode
 finishes it: a turn that ends without a reply then posts one missing-reply
 failure. A turn cancelled by cleanup or an interrupt posts nothing.
 
+New ordinary work remains outstanding while the session is occupied. Approval,
+status and question replies remain reachable through the shared control hook;
+free text while a question is pending keeps its existing question-answer
+meaning. Controls use their own delivery tools, while the active OpenCode turn
+retains its original tools. The room scheduler retries rejected ordinary work
+in order after the session becomes available.
+
 `variant` is an opaque, provider- and model-specific name the running server
 owns: it may select reasoning effort or be a custom provider setting, so read
 the server's model catalog rather than assuming `"high"` exists. The model

docs/turn-outcome.md

@@ -15,7 +15,7 @@ A turn is **complete** when it:
 - declined on purpose (`band_no_reply`)
 - did real work (any Band tool whose effect is `act`, such as
   `band_add_participant`)
-- was settled by the adapter itself (a control reply or a busy notice)
+- was settled by the adapter itself (such as a control reply)
 - already reported a failure (`send_failure`)
 
 Anything else is a **missing reply**. Fetching state or narrating through
@@ -115,6 +115,29 @@ async def main() -> None:
 asyncio.run(main())
 ```
 
+## Input that was not accepted
+
+`InputNotAccepted` means this particular attempt provably started no provider
+execution. A local occupancy check before sending or a validated provider busy
+rejection can establish that; timeouts and transport errors cannot. The SDK does
+not judge, fail or complete the rejected delivery, and does not spend its
+execution-failure retry budget. It remains outstanding for the room scheduler.
+One-shot invocation instead returns `OneShotStatus.NOT_ACCEPTED`; its caller
+owns later invocation, with no acknowledgement or history drain.
+
+Ordinary pending work keeps its order. Adapter-owned approval, status and
+question controls remain reachable behind it, using the existing authorization
+rules and the control message's own tools. A consumed control completes only
+itself. Already accepted detached work retains its original tools until its
+actual end and follows the existing acknowledgement and judgment rules.
+
+The default preprocessor prepares bootstrap, participant and system context
+without consuming it. A rejected attempt leaves that context owed; every other
+outcome commits it conservatively, preserving updates that arrived during the
+attempt. Custom preprocessors that consume context must defer that consumption
+through `AgentInput.commit_context` to support this contract. This waiting is
+process-local recovery, not an exactly-once guarantee across crashes.
+
 ## Declining with `band_no_reply`
 
 `band_no_reply` is in every adapter's tool set, and `BASE_INSTRUCTIONS` tells

docs/websocket-events.md

@@ -7,13 +7,15 @@ and `src/band/platform/event.py`. What the code cannot say for itself:
   rule-free typed projections: `WirePayload.from_wire` has core validate and normalize the
   payload (alias sync, defaulting, coercion), then hydrates the model without
   re-validating.
-- **Delivery-lifecycle decisions are core's too.** `OneShotInvoker`
-  (`src/band/runtime/oneshot.py`) is a thin caller-owns-the-loop wrapper over core's
-  `evaluate_delivery_event`, `evaluate_next_message`, `evaluate_drain_candidate` and
-  `evaluate_adapter_result`: event routing, drain-candidate classification (including
-  self-echo) and the ack decision are core's, not SDK logic.
-- **`ExecutionContext` is a separate machine.** It calls none of those functions. It dedups
-  on `metadata.delivery_status` and calls core's `is_self_echo` directly.
+- **Terminal delivery rules live in core.** `OneShotInvoker`
+  (`src/band/runtime/oneshot.py`) uses `evaluate_delivery_event`,
+  `evaluate_drain_candidate` and `evaluate_adapter_result` for event routing,
+  history drain classification and terminal acknowledgement. Python owns
+  ordinary-work ordering and adapter-control reachability. Definitively rejected
+  input returns before the terminal acknowledgement evaluator.
+- **`ExecutionContext` owns room scheduling.** It reuses core's claims, retry
+  tracking, participant roster and self-echo rules, with delivery metadata for
+  deduplication.
 
 ## Room reconciliation
 
@@ -41,3 +43,23 @@ retry cadence. A completed or skipped head that cannot advance ends the pass
 without spinning; retryable handler failures retain their normal attempt budget.
 These intervals bound scheduling, subject to active handlers, HTTP operations
 and recovery retries.
+
+## Pending work and control recovery
+
+A provider attempt that raises `InputNotAccepted` leaves its delivery
+outstanding. The next scheduler pass retries the oldest work once at the base
+reconciliation cadence. It releases local ownership and returns that attempt's
+failure-budget charge; adapters do not wait or retry internally. Durable work
+stays in platform rows, while synthetic messages remain in the local room queue.
+No scopes or tools survive merely to retry an input.
+
+`/next` exposes only the oldest outstanding delivery. While that work is blocked,
+the runtime also scans one chronological page of actionable `/messages` per
+pass to find missed controls, retaining the cursor across ordinary traffic.
+A changed pending head or the end of the scan starts a new scan. PLAY and
+reconnect bring a pass forward without rewinding its cursor.
+
+Controls use the same duplicate checks and accepted processing gate before an
+adapter handles them. A control consumed behind pending work acknowledges only
+its own delivery; an ordinary candidate is left outstanding. Runtime recovery
+never interprets approval syntax or drains model context for a control.

src/band/adapters/claude_sdk.py

@@ -82,6 +82,7 @@
 )
 from band.core.adapterconfig import BaseAdapterConfig
 from band.core.defaultmodels import ANTHROPIC_MODEL
+from band.core.exceptions import InputNotAccepted
 from band.core.protocols import (
     FAILURE_CODE_TIMEOUT,
     GENERIC_PROVIDER_FAILURE_MESSAGE,
@@ -761,6 +762,29 @@ def _room_mcp_servers(self, room_id: str) -> dict[str, McpServerConfig]:
             )
         }
 
+    async def try_handle_control(
+        self, msg: PlatformMessage, tools: AgentToolsProtocol
+    ) -> bool:
+        if self.config.approvals is None:
+            return False
+        command = self._extract_command(msg.content)
+        if command is None:
+            return False
+        cmd, args = command
+        sender = {"id": msg.sender_id, "name": msg.sender_name or msg.sender_type}
+        if cmd in _APPROVAL_CMDS:
+            await self._handle_approval_command(
+                tools=tools, room_id=msg.room_id, command=cmd, args=args, sender=sender
+            )
+        elif cmd == ClaudeSDKCommand.STATUS:
+            await self._handle_status_command(
+                tools=tools, room_id=msg.room_id, sender=sender
+            )
+        else:
+            return False
+        tools.turn.settle()
+        return True
+
     # --- Adapted from BandClaudeSDKAgent._handle_message ---
     async def on_message(
         self,
@@ -787,52 +811,16 @@ async def on_message(
                 "ClaudeSDKAdapter session manager not initialized — was on_started() called?"
             )
 
-        # Approval flow: track notify target and intercept local commands
+        running_turn = self._turn_tasks.get(room_id)
+        if running_turn is not None and not running_turn.done():
+            raise InputNotAccepted("The room already has an active Claude turn")
         if self.config.approvals is not None:
             self._room_last_sender[room_id] = {
                 "id": msg.sender_id,
                 "name": msg.sender_name or msg.sender_type,
             }
-            command = self._extract_command(msg.content)
-            if command is not None:
-                cmd, args = command
-                sender = self._room_last_sender[room_id]
-                if cmd in _APPROVAL_CMDS:
-                    await self._handle_approval_command(
-                        tools=tools,
-                        room_id=room_id,
-                        command=cmd,
-                        args=args,
-                        sender=sender,
-                    )
-                    tools.turn.settle()
-                    return
-                elif cmd == ClaudeSDKCommand.STATUS:
-                    await self._handle_status_command(
-                        tools=tools,
-                        room_id=room_id,
-                        sender=sender,
-                    )
-                    tools.turn.settle()
-                    return
-
-        # A prior turn's background task (see _run_turn) may still be running
-        # this room's ClaudeSDKClient -- most commonly while it's parked on a
-        # manual approval. A second concurrent client.query() on the same
-        # session is unsafe, so refuse rather than race it (mirrors
-        # OpencodeAdapter's own "still processing" guard).
-        running_turn = self._turn_tasks.get(room_id)
-        if running_turn is not None and not running_turn.done():
-            await tools.send_notice(
-                "Still processing the previous request in this room.",
-                mentions=[msg.sender_id] if msg.sender_id else None,
-            )
-            tools.turn.settle()
-            return
 
-        # Bound only once this message runs a turn: a command or busy message
-        # must not take over the in-flight turn's tools, which its approval
-        # prompts and MCP tool calls still post through.
+        # The running turn's approvals and MCP calls keep their original tools.
         self._room_tools[room_id] = tools
         await self._bind_mcp_tools(room_id, tools)
         await self._mcp.ensure()

src/band/adapters/codex.py

@@ -41,6 +41,7 @@
     deliver_notice,
     reraise_delivery_cause,
 )
+from band.core.exceptions import InputNotAccepted
 from band.core.protocols import (
     FAILURE_CODE_TIMEOUT,
     GENERIC_PROVIDER_FAILURE_MESSAGE,
@@ -169,9 +170,6 @@ class CodexCommand(StrEnum):
 APPROVAL_RESOLVED_TEMPLATE = "Approval `{token}` resolved as `{decision}`."
 APPROVAL_TIMED_OUT_TEMPLATE = "Approval `{token}` timed out. Applied `{decision}`."
 NO_APPROVALS_TO_RESOLVE_MESSAGE = "No pending approvals to resolve."
-TURN_IN_PROGRESS_MESSAGE = (
-    "Codex is still processing the previous request in this room."
-)
 
 
 class CodexSubcommand(StrEnum):
@@ -855,20 +853,17 @@ async def on_event(self, inp: AgentInput) -> None:
             self._raw_history_by_room[inp.room_id] = inp.history.raw
         await super().on_event(inp)
 
-    async def on_message(
-        self,
-        msg: PlatformMessage,
-        tools: AgentToolsProtocol,
-        history: CodexSessionState,
-        participants_msg: str | None,
-        contacts_msg: str | None,
-        *,
-        is_session_bootstrap: bool,
-        room_id: str,
-    ) -> None:
+    async def try_handle_control(
+        self, msg: PlatformMessage, tools: AgentToolsProtocol
+    ) -> bool:
+        command = self._extract_local_command(msg.content)
+        if command is None or (
+            command[0] not in _APPROVAL_COMMANDS and command[0] != CodexCommand.STATUS
+        ):
+            return False
+        room_id = msg.room_id
         self._room_client(room_id)
         self._active_room.set(room_id)
-        command = self._extract_local_command(msg.content)
         if command is not None and command[0] in _APPROVAL_COMMANDS:
             try:
                 handled = await self._handle_approval_command(
@@ -882,15 +877,38 @@ async def on_message(
                 reraise_delivery_cause(e)
             if handled:
                 tools.turn.settle()
-                return
+                return True
 
-        running = self._turn_tasks.get(room_id)
-        if running is not None and not running.done():
-            await tools.send_notice(
-                TURN_IN_PROGRESS_MESSAGE, mentions=self._sender_mention(msg)
+        if command is not None and command[0] == CodexCommand.STATUS:
+            await self._handle_local_command(
+                tools=tools,
+                msg=msg,
+                history=CodexSessionState(),
+                room_id=room_id,
+                command=command[0],
+                args=command[1],
             )
             tools.turn.settle()
-            return
+            return True
+        return False
+
+    async def on_message(
+        self,
+        msg: PlatformMessage,
+        tools: AgentToolsProtocol,
+        history: CodexSessionState,
+        participants_msg: str | None,
+        contacts_msg: str | None,
+        *,
+        is_session_bootstrap: bool,
+        room_id: str,
+    ) -> None:
+        self._room_client(room_id)
+        self._active_room.set(room_id)
+        running = self._turn_tasks.get(room_id)
+        if running is not None and not running.done():
+            raise InputNotAccepted("The room already has an active Codex turn")
+        command = self._extract_local_command(msg.content)
 
         release: asyncio.Future[None] = asyncio.get_running_loop().create_future()
         self._turn_release[room_id] = release

src/band/adapters/opencode/adapter.py

@@ -20,7 +20,7 @@
 from band.adapters.opencode.config import OpencodeAdapterConfig
 from band.converters.opencode import OpencodeHistoryConverter
 from band.core.delivery import relay_reply
-from band.core.exceptions import BandConnectionError
+from band.core.exceptions import BandConnectionError, InputNotAccepted
 from band.core.protocols import (
     FAILURE_CODE_TIMEOUT,
     GENERIC_PROVIDER_FAILURE_MESSAGE,
@@ -415,6 +415,15 @@ def _log_startup_config(self, agent_name: str) -> None:
             len(self._custom_tools),
         )
 
+    async def try_handle_control(
+        self, msg: PlatformMessage, tools: AgentToolsProtocol
+    ) -> bool:
+        room_state = await self._get_or_create_room_state(msg.room_id)
+        if not await room_state.approvals.try_handle_reply(msg.content, msg.sender_id):
+            return False
+        tools.turn.settle()
+        return True
+
     async def on_message(
         self,
         msg: PlatformMessage,
@@ -434,24 +443,13 @@ async def on_message(
                 agent_id or self.agent_name
             )
 
-        if await room_state.approvals.try_handle_reply(msg.content, msg.sender_id):
-            tools.turn.settle()
-            return
-
         if room_state.turn and (
             not room_state.turn.turn_future.done()
             or room_state.approvals.awaiting_human()
         ):
-            await tools.send_event(
-                "OpenCode is still processing the previous request in this room.",
-                "error",
-            )
-            tools.turn.settle()
-            return
+            raise InputNotAccepted("The room already has an active OpenCode turn")
 
-        # Only a message that starts a turn rebinds the room's tools: the MCP
-        # backend resolves band tool calls through them, so a message consumed
-        # above must not take over the in-flight turn's tools.
+        # MCP calls keep the active turn's tools until that turn finishes.
         room_state.tools = tools
 
         await self._ensure_client_started()