[RFC]: one linear ingest pipeline for the executor streaming path

Problem statement / motivation

The llm-d split stream relay landed in
#235 and
#236. Both were the right fix for
the problem in front of them, and together they pushed a second ingest leg and a second
validation regime into the core, landing the cost on three already-coupled files:
ResponseAccumulator, upstream.rs and inference.rs.

The strict validator carries no state of its own; it reuses the accumulator's in_flight
and completed maps, so validation and accumulation are now one object. That is why this
needs a redesign rather than a tidy-up: the three files can no longer be changed
independently.

Where that shows up today:

• Adding one output-item kind edits 8 sites in accumulator.rs. ARCHITECTURE.md:348
  calls the file "a stability contract" that "should not grow", but its documented extension
  point is "add a variant and the match arms".
• Four of its six public methods have no production caller: from_stream (bench only),
  from_sse_lines (tests only), mark_incomplete (nothing), process_stream_chunks (only
  reachable from from_stream). The spawn_blocking design in the module doc benchmarks
  well, and the live path re-implements the loop inline without it.
• Ingest and emission are interleaved in one 70-line loop (upstream.rs:208-252), so
  neither can be tested or changed alone.
• The same rules are written several times and disagree: five in-flight lookup rules;
  three enumerations of the terminal-event set, one of which silently drops error on
  response.completed; two deferred-frame drains that order index-less frames oppositely;
  two emit funnels with the same tail; byte-identical helpers in upstream.rs and
  events/validate.rs.
• The two ingress legs disagree on data: spacing. The live path requires the space
  (inference.rs:187), the relay path does not. The tolerance is tested but unreachable
  live, and data:[DONE] without a space fails to terminate the stream.

The entanglement has already produced a live bug. A repeated
response.output_item.done yields two output items instead of one on the lenient path.
The strict path rejects it, so only fetch_stream_payload is affected. It cannot be fixed
locally: the change that repairs the lenient path makes every strict decode fail with
upstream stream ended with unfinished output items.

Proposed solution

One ingest pipeline, with the non-linear part confined to one named component.

Two caveats on "linear", both forced by existing behaviour rather than chosen. Emission is
not linear: client-visible order depends on gateway tool results that do not exist while
the stream runs (engine.rs:360), and sequence numbers are stamped at emission
(gateway_accumulator.rs:43) where the tests demand exactly 0..N. The fold sits in the
middle, not the end: the translator runs after it, because it reads state the fold
produces.

  async task
    Transport ──▶ LineSplitter                              inference.rs
                       │ SseLine
  ═════════════════════╪═══════════════════════════ bounded channel ═════
  spawn_blocking worker (everything owned, 'static)
                       ▼
    Normalize ──▶ Validate ──▶ Fold ──▶ Translate
    normalize.rs  validate.rs  accumulator.rs  function_sse.rs
                       │ FunctionSseTranslation
  ═════════════════════╪═══════════════════════════ bounded channel ═════
  async task
                       ▼
    StreamRelay  (not linear: defer window, sequence, emit)  upstream.rs
         ├──▶ SSE to client
         └──▶ residual frames, drained after gateway tools run

    worker join ──▶ ResponsePayload

The core type: AgentPipeline

The whole ingest path becomes one owned, synchronous state machine. No I/O, no threads, no
type parameters, so it is directly unit-testable by feeding it lines.

enum Validation { Strict, Lenient }

struct AgentPipeline {
    validation: Validation,
    accumulator: ResponseAccumulator,   // the fold: slot map plus response-level fields
    translator: FunctionSseTranslator,  // runs after the fold, reads what the fold produced
}

impl AgentPipeline {
    fn new(
        response_id: String,
        conversation_id: Option<String>,
        validation: Validation,
        tool_types: HashMap<String, ToolType>,
    ) -> Self;

    /// One line in, zero or more client-bound frames out. This is the linear path:
    /// normalize -> validate -> fold -> translate, in that order, exactly once.
    fn push(&mut self, line: SseLine) -> ExecutorResult<FunctionSseTranslation>;

    /// Consuming, so "forgot to finalize" is unrepresentable. Replaces the four
    /// finalizers that today leave different state behind.
    fn finish(self) -> ExecutorResult<ResponsePayload>;
}

Everything that could drift between the live and relay legs (line filtering, normalization,
validation, folding, per-kind behaviour, finalization) lives inside push and finish, so
it cannot. The drivers around it are thin loops, and the only thing that varies between them
is where AgentPipeline runs and whether a relay consumes its output.

Six concrete changes

1. from_stream becomes the live path. The spawn_blocking offload is kept and widened
to cover normalize, validate, fold and translate. FunctionSseTranslator::new takes an owned
HashMap (function_sse.rs:57), so the translator is 'static and can sit on the worker.
&mut GatewayStreamAccumulator cannot cross, so sequence numbering stays async, which the
0..N contract requires anyway. Both channels become bounded, per AGENTS.md.

2. One core, two drivers. decode_upstream (upstream.rs:145) is sync and cannot await
a worker join, so the stages live in the sync AgentPipeline and the drivers stay thin.

Driver	Colour	Validation	Worker	Relay
from_stream (live)	async	Lenient	spawn_blocking	StreamRelay
from_stream (collect-only)	async	Lenient	spawn_blocking	null
decode_upstream (relay)	sync	Strict	none, inline	null

3. One slot map with a lifecycle, replacing the in_flight map plus the completed vec
plus the 8-variant InFlight shadow enum.

struct Slot { output_index: u32, state: SlotState }
enum SlotState {
    Active { kind: SSEItemType, item: Option<OutputItem>, buffer: String },
    Done(OutputItem),
}

  output_item.added      delta       output_item.done          finish
        │                  │                │                    │
        ▼                  ▼                ▼                    ▼
   Active{None} ──▶ Active{Some} ──▶    Done(item) ──▶  drained, sorted
        │                                  ▲              by output_index
        └──── web_search_call stays None ──┘
              until done supplies the action

   a repeated done lands on Done and is ignored

This fixes the duplicate-done bug as a state property rather than a special case. Done
slots stay in the map, so has_output_index, has_item_id and validate_terminal_output
keep working, which split_execution_integration.rs:274-278 requires. item stays Option
because there is no meaningful WebSearchCall at output_item.added time.

4. Per-kind behaviour moves to types/io/output.rs behind a static descriptor table,
next to the TryFrom<&EventPayload> and ApplyDone impls that already live there and where
ARCHITECTURE.md:360 already says item construction belongs.

pub struct ItemKindOps {
    pub from_added: fn(&EventPayload) -> Option<OutputItem>,
    pub apply_done: fn(&mut OutputItem, &EventPayload, &mut String),
    pub finalize:   fn(OutputItem, String) -> Option<OutputItem>,
}

Adding a kind then touches types/io/output.rs only. Two gaps to fill first:
OutputMessage has a TryFrom but no ApplyDone, and WebSearchCall has neither.

5. Strict versus lenient becomes one field, enum Validation { Strict, Lenient },
removing the parallel process_sse_line / process_strict_sse_line traversal and the five
competing lookup rules.

6. One emit funnel. There are exactly two production emit_sse_frame call sites
(upstream.rs:325, gateway.rs:577). StreamRelay owns both, via emit_local and
emit_upstream, so the frame's origin is in the method name rather than a magic offset
argument. The response-id rewrite and tool-name restoration live inside emit_upstream,
because applying them to a gateway-synthesized frame would rewrite values already correct.

Alternatives considered

No response

Additional context

Implementation order

Implementation order

One PR, ordered so the tree stays green at each step and the bugfix lands before the
structural change that would obscure it.

1. Slot lifecycle, which fixes the duplicate-done bug.
2. SseLine newtype and one filter stage, dropping the space-requiring gate at
   inference.rs:187. messages_stream.rs:304 moves with it or the Messages bridge starts
   dropping frames silently.
3. Translator decoupling: pass the frame's own delta on deltas, authoritative
   arguments only at done. Removes an O(n²) diff and the borrow that would fight the fold.
4. Validation field, merging process_sse_line and process_strict_sse_line behind
   one slot resolver. expected_item_type returns Option first, since unifying runs its
   unreachable!() against live upstream data.
5. ItemKindOps table, deleting the InFlight enum and its seven enumerations.
6. Consuming finalizer, four down to one.
7. StreamRelay extraction and the single emit funnel, reconciling the two
   deferred-frame drains and routing gateway.rs:571-580 through it.
8. AgentPipeline as the single entry point: widen the worker boundary, make
   from_stream the live path, bound both channels, re-run the bench.
9. Delete dead surface and update ARCHITECTURE.md.

Open question for reviewers. How large should the two channel bounds be? Bounding them
introduces backpressure onto the socket read for the first time. Too small stalls
next_chunk and risks tripping its timeout (inference.rs:29); too large reproduces
today's unbounded behaviour.

Start with the implementation order and read accumulator.rs, upstream.rs, inference.rs, types/io/output.rs, and function_sse.rs. Begin with the slot lifecycle and its duplicate-done behavior, checking split_execution_integration.rs:274-278, then follow the listed stages through AgentPipeline. Done means the live and relay paths share the pipeline, existing ordering and validation behavior remain intact, and the channel-bound decision is resolved and tested.