fix: robust notification loop — per-connection locking, alive check, drain window, and startup cleanup

[ruan330]

Problem

The notification loop in stream_prompt is built on three assumptions that don't hold in production with long-running ACP backends like Claude Code (#76):

1. ACP events arrive in order — end_turn sometimes arrives before the final agent_message_chunk, causing empty responses
2. Prompts always complete — long tool calls (build commands, test suites) block the loop forever, and with the global pool write lock, freeze the entire broker
3. Session lifecycle is self-managing — after restart, stale Discord threads create zombie sessions that consume pool slots

These are three symptoms of one root cause: the notification loop has no resilience against real-world conditions.

Solution

A single cohesive fix addressing all three, not three separate patches:

Per-connection locking (pool.rs)

with_connection() holds a global write lock for the entire streaming duration. One busy session blocks all others (#58).

Replace HashMap<String, AcpConnection> with HashMap<String, Arc<Mutex<AcpConnection>>>. Pool RwLock is now only held for brief lookups. Each connection has its own Mutex.

// Before: global write lock held for minutes
pool.with_connection(thread_key, |conn| { /* streaming */ })

// After: brief read lock for lookup, per-connection lock for use
let conn_arc = pool.get_connection(thread_key).await?;
let mut conn = conn_arc.lock().await;  // only locks THIS connection

Alive check + hard timeout (discord.rs)

Replace bare while let rx.recv() with tokio::select!:

• Every 30s: check if agent process is alive. Dead → break immediately
• After 30 minutes: hard timeout safety net. Alive but stuck → break

Correctly handles long tool calls (process alive → wait) without false timeouts, while preventing infinite blocking.

Drain window (discord.rs)

After receiving end_turn, drain the notification channel for 200ms to capture late-arriving text chunks:

if notification.id.is_some() {
    let drain_until = Instant::now() + Duration::from_millis(200);
    while let Ok(Some(n)) = timeout_at(drain_until, rx.recv()).await {
        if let Some(AcpEvent::Text(t)) = classify_notification(&n) {
            text_buf.push_str(&t);
        }
    }
    break;
}

Empty response fallback (discord.rs)

If text_buf is empty after draining but tool activity was recorded, compose a fallback from tool lines instead of showing "(no response)".

Startup thread cleanup (discord.rs, ready handler)

On startup, archive all active threads in allowed channels created by this bot. Prevents stale threads from spawning zombie sessions after restart.

Tradeoffs

Decision	Cost	Why
200ms drain window	Adds 200ms to every prompt completion	Small; avoids losing entire responses
30-min hard timeout	Very long tasks get interrupted	Safety net; should not trigger in normal use
Auto-archive on startup	Can't resume old threads	Old threads have no session context anyway
Arc<Mutex> vs checkout pattern	Slightly more memory per connection	Correct abstraction; enables future prompt queueing via try_lock()

Testing

Tested with claude-agent-acp backend, concurrent sessions:

• Two threads running simultaneously — no cross-session blocking ✅
• Long tool call (flutter build, 5+ min) — other sessions respond normally ✅
• Process crash — detected within 30s, session released ✅
• Stale threads after restart — auto-archived on startup ✅
• ACP event ordering violation — text captured via drain window ✅

Changes

File	Lines	What
pool.rs	+54/-43	Arc<Mutex<AcpConnection>>, get_connection(), remove with_connection()
discord.rs	+145/-90	Per-connection lock, alive check, hard timeout, drain window, fallback, startup cleanup

Supersedes #59. Closes #58. Ref #76.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: f991c783d6

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Don't hold pool write lock while waiting on connection mutex

cleanup_idle now grabs self.connections.write() and then awaits each conn_arc.lock(). If any session is actively streaming (its connection mutex is held in stream_prompt), the cleanup task blocks on that await while still holding the pool write lock, which prevents get_or_create/get_connection for unrelated threads. Because main.rs runs this cleanup every 60s, one long prompt can periodically stall the whole broker again.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Treat empty allowed_channels as wildcard in startup cleanup

The message path treats an empty allowlist as "all channels allowed", but the startup cleanup path requires self.allowed_channels.contains(parent_id), so when allowed_channels is empty (the default), is_ours is always false and no stale threads are archived. That makes the new zombie-thread cleanup feature silently ineffective in default deployments.

Useful? React with 👍 / 👎.

[ruan330]

Closing this in favor of 3 smaller, focused PRs — on reflection this one bundled too many concerns for a clean review. Sorry for the churn.

The split plan:

1. Per-connection Arc<Mutex<AcpConnection>> — just the pool.rs architecture change + discord.rs call sites. This implements the locking approach discussed in RFC RFC: Session Management — Design Proposal #78 §2b, and closes fix: pool write lock held during entire prompt streaming causes cross-session deadlock #58 / supersedes fix: per-connection locking to prevent cross-session deadlock during long prompts #59.
2. Notification loop resilience — drain window + empty-response fallback (end_turn can arrive before the last agent_message_chunk via tokio::select!). Fixes fix: notification loop assumes ordered events, bounded prompts, and managed session lifecycle — none hold in production #76, which we have a production repro for. ~20-30 lines.
3. Alive check + hard timeout — defensive safety net. 30s alive check + 30min hard timeout via tokio::select!.

Dropped from original scope: startup thread cleanup. We reversed that decision in our fork — archiving active threads on restart loses user context, and with per-thread state persistence (related to RFC #78 Phase 2) there's no longer a correctness reason to do it.

I'll send them in order, starting with #1 soon. Happy to discuss any of these beforehand if you'd prefer a different approach.