perf: benchmark inline versus worker placement for streaming ingestion

[franciscojavierarceo]

the current server-sent events (SSE) benchmark exercises the offloaded accumulator under concurrency, but it has no equivalent inline SSE control and uses an in-memory stream. that doesn't establish whether moving the live pipeline to a worker improves the workloads we care about. we should compare placement after the synchronous pipeline exists, then record whether offloading is worth implementing.

proposed third slice of #241, using the synchronous core from #243. relay extraction and client backpressure do not depend on a positive result here. this is a suggested scope; feel free to revise, combine, or remove it.

we should be able to verify that:

• inline and worker variants run equivalent normalize/validate/fold/translate work with the same inputs and correctness checks; include both isolated processing and end-to-end measurements.
• workloads cover realistic event sizes and arrival pacing, concurrent streams, long outputs, and slow consumers. report throughput, tail event latency, CPU use, memory, and worker/thread counts with reproducible commands and environment details.
• experiments sweep bounded queue capacities with byte budgets and occupancy measurements; 16 entries is a candidate to test, not a prescribed default.
• the experimental worker driver feeds input and drains output concurrently, passes tiny-capacity deadlock tests, and has explicit disconnect, cancellation, shutdown, and join/error handling. don't rely on aborting an already-running spawn_blocking task.
• a recorded placement recommendation explains the tradeoffs against agreed latency/resource goals. keeping the live path inline is a valid outcome; production offloading is a separate follow-up only if the evidence supports it.

the existing benchmark is the baseline to extend.

[maralbahari]

Sub-issue of #241. Depends on
#243. Decides #241 step 8.

Why this is only answerable after #243

The two placements cannot be compared today because they are not the same code. The live
path re-implements the ingest loop inline (upstream.rs:208-252); from_stream
(accumulator.rs:214-260) has the mpsc + spawn_blocking offload and no production
caller. Benchmarking them measures two implementations, not two placements.

#243 removes that confound: AgentPipeline is synchronous, owns no I/O and no threads, so
placement becomes the only variable. Both arms call identical push/finish.

This issue supersedes the "keep the offload, the benchmark justifies it" line in the #241
plan. The existing accumulator_concurrent_stream bench has no inline control, so it
cannot support that claim. Treat the placement as undecided until this lands.

Interfaces

// benches/placement.rs (new)

/// The only axis under test. Both arms drive the same AgentPipeline.
#[derive(Clone, Copy, Debug)]
pub enum Placement {
    /// Drive AgentPipeline on the caller's task.
    Inline,
    /// Drive it on spawn_blocking behind bounded channels.
    Worker { line_queue: usize, out_queue: usize },
}

pub struct PlacementRun {
    pub placement: Placement,
    pub workload: Workload,
    pub payload: ResponsePayload,   // correctness check, must match across arms
    pub metrics: Metrics,
}

pub async fn run(placement: Placement, workload: &Workload) -> ExecutorResult<PlacementRun>;

pub struct Workload {
    pub name: &'static str,
    pub lines: Vec<SseLine>,          // replayed from cassettes, not synthesised
    pub arrival: Arrival,             // burst | paced { every } | jittered
    pub concurrent_streams: usize,
    pub consumer: Consumer,           // immediate | slow { per_event }
}

pub struct Metrics {
    pub throughput_lines_per_sec: f64,
    pub latency_p50: Duration,
    pub latency_p99: Duration,
    pub latency_max: Duration,
    pub cpu_time: Duration,           // user + sys, not wall
    pub peak_rss: u64,
    pub threads_peak: usize,
    pub queue_occupancy_p99: usize,   // Worker only
}

The worker driver is the same one #241 step 8 would ship, exercised here first:

struct WorkerDriver {
    lines: mpsc::Sender<SseLine>,
    out: mpsc::Receiver<FunctionSseTranslation>,
    join: JoinHandle<ExecutorResult<ResponsePayload>>,
}

impl WorkerDriver {
    fn spawn(pipeline: AgentPipeline, line_queue: usize, out_queue: usize) -> Self;
    async fn push(&mut self, line: SseLine) -> ExecutorResult<()>;   // awaits on backpressure
    async fn finish(self) -> ExecutorResult<ResponsePayload>;
    async fn cancel(self);                                           // no leaked task, no hang
}

Logic

Both arms must produce byte-identical payloads. Assert equality of the two
ResponsePayloads per workload before recording any timing. A placement that is faster and
wrong is not a result.

Concurrent send and receive is mandatory in the worker arm. Feeding all lines before
draining the output channel deadlocks at any bound smaller than the stream. The driver must
pump both directions, and the deadlock test at capacity 1 is the proof.

Measure CPU time, not just wall clock. The offload's claim is that it frees the tokio
executor between chunk arrivals. Wall-clock throughput on a burst workload will not show
that; a paced workload with concurrent streams and executor-thread occupancy will.

Queue sizing is a swept parameter, not a constant. Sweep line_queue and out_queue
over 1, 16, 64, 256 and report occupancy at p99 alongside latency. Capacity 1 is the
deadlock canary; the useful range is wherever occupancy stops tracking capacity.

Slow consumers are a first-class workload, not an edge case. This is where backpressure
either works or turns into an upstream next_chunk timeout (inference.rs:29).

Workloads

Replay real cassettes so event sizes and arrival shapes are realistic:

Workload	Source	Exercises
text_short	text_only/	per-line overhead floor
reasoning_long	reasoning/	sustained deltas, large buffers
tool_calls	tool_calls/, parallel_tool_calls/	translator work in the fold
web_search_gateway	web_search/	deferral and gateway interleave
concurrent_n	any of the above, N streams	executor contention
slow_consumer	reasoning_long + paced drain	backpressure behaviour

Deliverable

A short decision record in the repo, not just numbers in a PR comment: the measured
tradeoff, the chosen placement, the queue bounds, and the goals it was judged against.

Keeping the inline architecture is a valid outcome. If it wins or ties, #241 step 8
reduces to deleting from_stream's offload rather than widening it, and the RFC's
"kept and widened" framing needs amending.

Tests

Beyond the benchmark itself, the worker driver needs real tests, since #241 step 8 would
ship it:

• capacity 1 on both channels, concurrent push and drain, terminates
• cancellation mid-stream: no leaked task, no hang, join observed
• consumer dropped mid-stream: driver returns, does not panic
• pipeline error propagates through join rather than being swallowed
• payload equality between placements on every workload

Out of scope

Ingestion semantics (#243) and emission or client-channel bounds (#244). This issue changes
no production behaviour; it produces a measurement and a decision.