feat(acp): stream an agent's reply into the channel as it is written

[noah-shipley-centric]

Problem

ACP delivers an answer as a run of agent_message_chunk updates. handle_session_update logged each one to acp::stream and returned:

"agent_message_chunk" => {
    if let Some(text) = update["content"]["text"].as_str() {
        tracing::info!(target: "acp::stream", "{text}");
    }
    false
}

So a channel sees nothing until the agent posts a finished message. Every token was already arriving — it just never left the log. From the room, the agent goes quiet for a minute and then a whole answer appears at once.

Approach

Post once on the first meaningful text, then edit that same message as more arrives. A stream is one message that grows, not a wall of fragments.

This works cleanly because kind 40003 edits are already excluded from CHANNEL_MESSAGE_EVENT_KINDS — the existing comment says edits "can land after the last human-visible message and would otherwise create phantom unreads". That is exactly the property streaming needs: a streamed answer does not become hundreds of badge pings.

The policy (stream_flush.rs, pure)

Relaying every chunk would be worse than not streaming, so publishing is coalesced:

• The first publish ignores the throttle. Delaying the first sign of life is backwards — that is when a reader most needs to know the agent is working.
• Mid-turn edits require both a minimum delta and a throttle interval.
• End-of-turn always flushes, so the tail is never dropped (no_tokens_are_lost_between_the_last_edit_and_turn_end).
• An empty turn publishes nothing, and never blanks an existing message.
• Bodies are clamped under build_edit's 64 KiB limit, truncating on a char boundary — a rejected edit would strand the message mid-answer, and a naive byte slice would panic on a multi-byte codepoint.

No clock, no I/O: the caller supplies now_ms and performs the publish.

Lifecycle

The sender is scope-local in the turn driver on purpose — dropping it is the end-of-turn signal, so success, cancel, idle timeout and agent exit all flush without any path having to remember to. This mirrors the existing ReactionGuard pattern.

The driver awaits the streamer before reporting the outcome, so the complete reply is on the relay before anything observes "turn complete" — otherwise a caller could see a finished turn against a message still missing its last edit.

Only channel turns stream; a heartbeat has no channel to write into.

Safety

Off by default (BUZZ_ACP_STREAM_REPLIES) — this changes what every channel sees, so it is opted into rather than inherited. BUZZ_ACP_STREAM_THROTTLE_MS (1000) and BUZZ_ACP_STREAM_MIN_DELTA (24) tune it. With it unset, behaviour is byte-for-byte what it was: chunks are logged and dropped.

Tests

18 tests — every policy branch, plus the post-once-then-edit contract driven through an injected StreamPublisher fake:

• exactly one post per turn; every edit targets that event id
• a dropped sender flushes the tail even mid-throttle-window
• an empty turn publishes nothing
• a failed post produces no orphan edits — it rolls the state back, otherwise the policy would believe a message existed and the whole reply would vanish into edits against nothing
• disabled streaming publishes nothing at all

cargo test -p buzz-acp --lib        # 634 passed
cargo clippy -p buzz-acp --all-targets -- -D warnings   # clean
cargo fmt -p buzz-acp -- --check                        # clean

Notes

Independent of #3192 and #3196 — different files, no shared commits.

Not yet exercised against a live relay; the publish path is covered by the injected-fake seam rather than an end-to-end run. Happy to add an env-gated integration test if you would rather see one before merging.