asyncio concurrency, cancellation, and cleanup

Corresponding source

Subject	Source
Queue consumption and task creation	nanobot/agent/loop.py::AgentLoop.run, _dispatch
Session serialization and task cancellation	nanobot/agent/loop.py::_cancel_active_tasks
Concurrent Tool batches	nanobot/agent/runner.py::_partition_tool_batches, _execute_tools
External processes and Shell	nanobot/agent/tools/shell.py::ExecTool
MCP cleanup	nanobot/agent/loop.py::close_mcp
Gateway shutdown	nanobot/cli/commands.py::_run_gateway

Learning objectives

After this lesson, you should be able to:

• explain the implementation of “serialized within one Session, concurrent across Sessions”;
• state when multiple Tool Calls in one Turn can run concurrently;
• trace /stop to Task.cancel() and checkpoint restoration;
• identify cleanup order during graceful Gateway shutdown.

Prerequisites

• Know that a coroutine switches only when it yields control at await or an equivalent boundary.
• Know that asyncio.Lock protects a critical section and Semaphore limits concurrency.
• Know that cancellation raises asyncio.CancelledError inside the canceled coroutine, so cleanup code normally re-raises it.

1. Separate four kinds of concurrency

nanobot has at least four concurrency levels:

Level	Concurrent?	Control point
Turns from different Sessions	Yes	AgentLoop.run() and _dispatch()
Multiple Turns in one Session	Serialized; new messages may become mid-turn injections	_session_locks, _pending_queues
Multiple Tool Calls from one model response	Only safe Tools can form a concurrent batch	AgentRunner._partition_tool_batches()
Gateway Agent, Channels, WebUI, health check	Run concurrently	Tasks inside nanobot/cli/commands.py::_run_gateway()

Collapsing these into “nanobot is asynchronous” makes debugging almost directionless.

2. From the global queue to each Session

AgentLoop.run() continuously consumes MessageBus.consume_inbound(). A normal message creates a _dispatch(msg) Task, allowing the main loop to keep receiving /stop or messages for other Sessions.

sequenceDiagram
    participant Bus as MessageBus
    participant AL as AgentLoop.run
    participant A as Session A Task
    participant B as Session B Task

    Bus->>AL: Message for A
    AL->>A: create_task(_dispatch)
    Bus->>AL: Message for B
    AL->>B: create_task(_dispatch)
    Note over A,B: The two Sessions can progress concurrently

_dispatch() then acquires two limiters:

1. _session_locks[session_key]: ensures one owner for a Session;
2. _concurrency_gate: limits requests running across the entire AgentLoop.

The global limit comes from NANOBOT_MAX_CONCURRENT_REQUESTS and currently defaults to 3. A value of 0 or lower creates no Semaphore, disabling this limit.

The two limiters solve different problems

The Session Lock protects conversation ordering and state. The global Semaphore prevents unbounded load on the Provider, CPU, memory, and downstream Tools.

3. Follow-up messages in one Session

While a Session is running, AgentLoop exposes _pending_queues[session_key] with a maximum length of 20. Later normal messages enter this queue for injection at an AgentRunner iteration boundary instead of starting a competing Turn.

Special cases:

• a priority command is handled inline;
• a dispatchable command is not injected as natural language;
• when the queue is full, source falls back to a new Task waiting for the Session Lock;
• after the Turn, an unconsumed message is republished to the inbound bus instead of being lost silently.

_pending_queues is therefore not a general backlog. It is an injectable mailbox for the currently running Turn.

4. Tool Call concurrency is not globally enabled

The main loop creates AgentRunSpec with concurrent_tools=True, but AgentRunner does not blindly call gather() for every Tool.

_partition_tool_batches() checks concurrency_safe:

• adjacent safe calls can form one batch and run with asyncio.gather();
• a non-concurrency-safe Tool forms its own batch;
• batches still run in order.

The base default is:

• nanobot/agent/tools/base.py::Tool.concurrency_safe

A read may be concurrency-safe, while an operation mutating shared files, processes, or Session state commonly must be serialized. A new Tool is not concurrency-safe merely because its implementation uses async def.

5. How /stop becomes cancellation

The control command eventually calls AgentLoop._cancel_active_tasks(key), retrieves the unfinished Task for that Session from _active_tasks, and calls cancel().

Cancellation propagates through the call stack:

flowchart TD
    A["/stop"] --> B["_cancel_active_tasks"]
    B --> C["Task.cancel()"]
    C --> D["AgentRunner / Tool / Provider receives CancelledError"]
    D --> E["ExecTool kills and reaps subprocess"]
    D --> F["AgentLoop attempts runtime-checkpoint restoration"]
    E --> G["Re-raise CancelledError"]
    F --> G

Key implementation:

• AgentLoop._cancel_active_tasks();
• the except asyncio.CancelledError branch in AgentLoop._dispatch();
• cancellation branches in AgentRunner.run() and _run_tool();
• ExecTool.execute() and _kill_process().

On timeout or cancellation, ExecTool terminates the subprocess and waits for it to exit, avoiding an orphaned zombie process. Cleanup then re-raises cancellation so upper layers know the Turn did not complete normally.

6. Why cancellation can preserve partial progress

During the tool loop, Runner writes checkpoints into Session metadata. When AgentLoop._dispatch() catches cancellation, _restore_runtime_checkpoint() attempts to restore already-produced assistant/tool messages to Session history before saving the Session.

Cancellation therefore does not mean “roll back the entire Turn.” It is closer to:

Stop further execution while preserving intermediate results that already happened and remain useful.

This is an important difference between Agent systems and database transactions. Real-world Tool side effects usually cannot be undone automatically.

7. Graceful Gateway shutdown

Gateway runs AgentLoop, ChannelManager, and configured WebUI/health services concurrently. After a shutdown signal, the finally path in nanobot/cli/commands.py performs cleanup. Its core order includes:

1. cancel remaining runtime Tasks;
2. wait for them with gather(..., return_exceptions=True);
3. stop Channels;
4. close MCP connections and background work;
5. call SessionManager.flush_all() to rewrite cached Sessions with fsync=True;
6. restore signal handlers.

AgentLoop.close_mcp() calls out one important constraint: an stdio MCP transport uses an AnyIO cancel scope and should close in the Task that opened it.

8. Debugging methods

Observe cross-Session concurrency

Send two requests that each wait several seconds under different Sessions. Set breakpoints at:

• Task creation in AgentLoop.run();
• async with lock, gate in AgentLoop._dispatch();
• AgentRunner._request_model().

Record each Task’s session_key and entry time. Expected result: different keys can overlap; the same key cannot hold the Lock concurrently.

Observe cancellation

Send /stop to a Turn running a long command and inspect:

• cancellation count returned by _cancel_active_tasks();
• ExecTool._kill_process();
• _restore_runtime_checkpoint();
• whether Session JSONL retains a pre-cancellation tool result.

Do not use a destructive command for a cancellation experiment

Cancellation can stop an unfinished process but cannot reliably undo data already written to an external system.

9. Source and tests

Source:

• nanobot/agent/loop.py::run
• nanobot/agent/loop.py::_dispatch
• nanobot/agent/loop.py::_cancel_active_tasks
• nanobot/agent/loop.py::close_mcp
• nanobot/agent/runner.py::_execute_tools
• nanobot/agent/runner.py::_partition_tool_batches
• nanobot/agent/tools/shell.py::ExecTool
• nanobot/cli/commands.py::_run_gateway

Tests:

• tests/agent/test_unified_session.py
• tests/agent/test_runner_injections.py
• tests/agent/test_stop_preserves_context.py
• tests/agent/test_runner_tool_execution.py
• tests/agent/test_loop_tool_context.py
• tests/session/test_session_fsync.py

Self-check

1. Why is _session_locks alone insufficient?
2. Is a read-only Tool automatically safe to mark concurrency_safe=True?
3. Why is canceling a Turn not the same as rollback?
4. Why must unconsumed _pending_queues messages be republished?

Next: Errors, retries, and data consistency