Channels: send autonomous turns to the channel that last messaged the session

[niklassemmler]

Problem to be solved:

On Telegram, nothing the agent says in an autonomous turn reaches the user. An autonomous turn is the full turn the CLI runs when a background task settles (run_in_background Bash or Agent, Monitor). Typical case: the user asks for something, the agent starts a background job and answers "on it". The job finishes, and the agent's follow-up (results, a PR link, the actual answer) appears only in the web UI. In one session this hid five consecutive messages, about an hour of output, until the user asked whether the agent was still working.

The cause is in how output is routed. ChannelRouter._run_single registers a StreamAdapter as a broadcaster listener for the run of the inbound message it answers, and _teardown_streaming unregisters it when that run ends. Autonomous turns arrive later, through the idle-stream watcher (_drain_pending_messages). They are broadcast normally, so the web UI's WebSocket listener sees them, but by then no channel listener is registered. telegram.py has no other path for them.

Changes:

• ChannelRouter records the (channel, target) that last messaged each session.
• New open_autonomous_stream(session_id) / close_autonomous_stream(session_id):
  • Open creates a StreamAdapter for that target under its own listener id (<channel>:<target>:autonomous), so closing it never unregisters a user run's adapter.
  • Open does nothing when the session was never messaged through a router channel (web UI, cron, workflow runs), or when a user run is already streaming to the same target. That second case covers a turn drained inside run(), which reaches the channel through the run's own adapter.
  • Close is idempotent. If the turn ended without done (cancelled, failed), close sends whatever text arrived.
• _drain_pending_messages opens the stream when an autonomous turn opens and closes it after the turn is finalized, once per turn. It also closes it in finally for turns cut short. Errors from the hooks are logged and never break the drain. Engines without a router skip the hooks.
• StreamAdapter gets two changes:
  • send_empty=False, used for autonomous turns: an empty turn deletes its placeholder instead of sending (no response).
  • A finished guard, so a backstop done after the real one can't send the message twice.

Testing:

• New tests/test_router_autonomous.py runs the router against a real StreamBroadcaster and a fake streaming channel. It covers delivery to the last inbound target, sessions never messaged through a channel, empty turns, turns cut short, duplicate done, an active user-run adapter, idempotency, and listener-id isolation.
• New drain tests in tests/test_autonomous_turns.py cover open, then finalize, then close order; one stream per turn when several turns drain in one call; empty turns; timeouts; cancellation mid-turn; and engines without a router.
• I broke the code on purpose nine ways, and each was caught by at least one test: no open, no close per turn, no close in finally, no inbound record, no finished guard, send_empty=True, a shared listener id, no user-run check, no flush on close.
• Full suite: run after opening; results are in the comments below.

Not in scope:

• _last_inbound is in memory only. That's fine, because a restart also kills the CLI subprocess and any background tasks that could produce an autonomous turn.
• Turns drained inside run() still share the user run's adapter, exactly as before this PR.

🤖 Generated with Claude Code

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.

---

MAGI (agent of Niklas Semmler) seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[niklassemmler]

Full suite on this branch: 3943 passed.

[niklassemmler]

Follow-up gap I found while deploying this, left out of this PR on purpose:

1. Resume turns don't reach Telegram. nerve restart --resume <id> starts an internal "continue" turn through engine.run() (resume_enrolled_sessions), not through ChannelRouter.handle_message. No channel adapter is registered for it, so a Telegram session sees nothing it says.
2. _last_inbound is empty after a restart. It lives in memory, so autonomous turns started from a resumed session can't be routed until the user sends a new message.

Both could be fixed together by seeding the router's session-to-target mapping from the session's persisted channel_key / source, rather than only from inbound messages. I'll do that in a separate PR if it's wanted.