fix(acp): stream durable session updates that follow a message chunk

`DurableUpdateSink::handle_notification` buffered every non-message
`session/update` — `tool_call`, `tool_call_update`, `plan`, `mode`,
`available_commands_update` — whenever a message chunk had already opened
the completion buffer, and committed the buffer only at the next message
boundary.

Any agent that streams text before calling a tool therefore withholds its
whole durable stream until the *post-tool* message arrives. Pi prints a
banner chunk on some turns, so roughly a third of prompt turns delivered
`tool_call_update { in_progress }` for a `sleep 60` only when the command
finished, together with every other update of the turn. A caller that
waits for that boundary before cancelling never observes a live turn:
`session/cancel` then races prompt completion and reports
`no_active_prompt`.

Commit the in-progress text run first, then the update. The durable
sequence is unchanged — `coalesce_completed_message` already ends the
text run it is building at any non-text update, so the same events are
emitted in the same order, only in more (smaller) batches — and the
update now reaches the host the moment the adapter produces it.

The regression test drives an ACP adapter that emits one message chunk,
then the tool updates, then holds the turn open for two seconds. It
records the live event sink and fails if the `in_progress` update lands
at turn end: on the previous behavior it arrives at 2.06s of a 2.06s
turn.

Author: Yuriy Butenko

crates/agentos-sidecar/src/acp/turn.rs

@@ -1017,20 +1017,24 @@ impl DurableUpdateSink {
         }
 
         // Tool, plan, mode, and other durable updates may be interleaved with
-        // message deltas. Keep their native order inside the in-progress
-        // completion buffer instead of treating them as a message boundary.
-        if self.buffered_kind.is_some() {
-            self.buffered_bytes = checked_acp_bytes(
-                &self.user_session_id,
-                self.buffered_bytes,
-                update_bytes,
-                self.limits.max_completed_message_bytes,
-                "limits.acp.maxCompletedMessageBytes",
-            )?;
-            self.buffered.push(update);
-        } else {
-            self.persist(ctx, events, vec![update]).await?;
-        }
+        // message deltas. Commit the in-progress text run first, then this
+        // update, so it reaches the host the moment the adapter produces it.
+        //
+        // Holding it in the completion buffer instead would keep it invisible
+        // until the next message boundary: an agent that streams any text
+        // before calling a tool (Pi prints a banner chunk on some turns) arms
+        // the buffer, and every later `tool_call` / `tool_call_update` is then
+        // withheld until the *post-tool* message arrives — so a
+        // `tool_call_update { in_progress }` for a `sleep 60` reaches the host
+        // only when the command finishes, and a caller waiting on that
+        // boundary before cancelling never gets a live turn to cancel.
+        //
+        // The durable sequence is unchanged: `coalesce_completed_message`
+        // already ends the text run it is building at any non-text update, so
+        // flushing here emits the same events in the same order, just in more
+        // (smaller) batches.
+        self.flush(ctx, events).await?;
+        self.persist(ctx, events, vec![update]).await?;
         Ok(true)
     }