fix(telegram): concurrent background-task resumes open overlapping flow blocks on one session

[adolfousier]

When several detached background commands finish close together, each one spawns its own full streaming resume turn on the SAME session. Every turn opens and edits its own flow block in the same chat, so the blocks duplicate and overlap instead of one block updating in place.

Evidence

Five resume_session calls on one session inside 850ms, on four different threads:

16:07:52.088  ThreadId(03)  resume_session a1a11b73... with full streaming pipeline
16:07:52.282  ThreadId(03)  resume_session a1a11b73... with full streaming pipeline
16:07:52.437  ThreadId(05)  resume_session a1a11b73... with full streaming pipeline
16:07:52.606  ThreadId(11)  resume_session a1a11b73... with full streaming pipeline
16:07:52.764  ThreadId(11)  resume_session a1a11b73... with full streaming pipeline

Fifteen more on the same session over the following two minutes.

Mechanism

build_enqueue_callback in src/channels/telegram/resume.rs does a bare tokio::spawn per finished task:

Arc::new(move |session_id, msg| {
    tokio::spawn(async move {
        ...
        resume_session(bot, chat_id, thread_id, session_id, msg.context_text, agent, state).await
    });
})

There is no per-session gate. resume.rs never calls try_begin_turn, so the background path bypasses the concurrency gate that both user messages (handler.rs:3237) and reactions (handler.rs:4244) go through. Nothing serializes the resumes and nothing tells a second one that a turn is already streaming.

Fix

Claim the turn before resuming, and enqueue rather than fork when the session is busy:

• try_begin_turn(session_id) returns a guard: run the resume holding it, so a concurrent completion sees the session as busy.
• It returns None: push the result onto the existing per-session queue via enqueue_reaction. The tool loop drains that queue between rounds through reaction_queue_callback, which is registered on the AgentService at manager.rs:146 and therefore applies to any in-flight turn on the session, resume turns included.

The result lands inside the block that is already open instead of opening a second one, which is the requested behaviour: one block per session, edited in place.

Scope this to build_enqueue_callback rather than to resume_session itself, so startup crash-recovery resume keeps its current behaviour.

Impact

• Duplicate, overlapping flow blocks whenever more than one background task finishes at once
• Concurrent turns on a single session racing to edit the same chat
• Background results interleaved across blocks rather than reported in one place

[adolfousier]

Fixed in edd4123b.

Cause

build_enqueue_callback did a bare tokio::spawn per finished task with no per-session gate. resume.rs never called try_begin_turn, so the background path was the only one bypassing the concurrency gate that user messages (handler.rs:3237) and reactions (handler.rs:4244) have gone through since #501. Nothing serialized the resumes and nothing told a second one a turn was already streaming.

Fix

The producer claims the turn before resuming. A completion that loses the claim pushes its result onto the per-session queue instead of forking a second turn:

let Some(_turn_guard) = state.try_begin_turn(session_id) else {
    state.enqueue_reaction(session_id, msg);
    return;
};

The tool loop drains that queue between rounds through reaction_queue_callback, registered on the AgentService at manager.rs:146, so it applies to any in-flight turn on the session including a resume turn. The result lands in the block that is already open, which is the requested behaviour: one block per session, edited in place.

Scoped to the background producer rather than to resume_session itself, so startup crash-recovery resume is untouched.

Tests

src/tests/telegram_bg_resume_gate_test.rs pins both halves and the failure modes a gate like this invites:

• only the first of five simultaneous completions claims the turn
• a loser's result is recoverable rather than dropped, and multiple losers keep FIFO order
• the session does not stay wedged once the turn ends
• unrelated sessions do not gate each other
• the enqueue rail is the same one the tool loop's callback drains, so a queued result cannot sit undelivered

cargo clippy --all-features --lib --bins --tests -- -D warnings clean, 5669 tests pass.

Worth noting the limit of this: the gate guarantees one turn per session, but the queued result is delivered by the running turn only when that turn reaches its next round. A background task finishing just as a turn is wrapping up will surface in the following turn instead.