fix: normalize multi-round gateway streaming

[harivilasp]

Summary

Revives and updates the GatewayAccumulator work for #119 on top of current main.

The gateway streaming path currently forwards each upstream round as if it were an independent client-visible response. In a multi-round gateway/tool loop that leaks upstream lifecycle and indexing details to clients: duplicate response lifecycle events, sequence numbers that reset between rounds, and output indexes that collide across rounds.

This PR introduces a single stream accumulation layer for gateway responses so the public SSE stream is owned and normalized at one boundary before it reaches the client.

Closes #119.

What Changed

• Added gateway stream accumulation for public SSE events across gateway rounds.
• Deduplicated client-visible lifecycle events so a multi-round stream emits one logical response lifecycle.
• Assigned monotonic public sequence numbers across forwarded upstream frames, synthetic gateway frames, and the terminal event.
• Rebased output indexes so final client-visible output items remain contiguous across rounds instead of restarting at each upstream call.
• Routed forwarded upstream frames, synthetic tool/search frames, and terminal response events through the same normalization path.
• Added a multi-round streaming regression test covering a gateway tool round followed by a final text round.

Why

Issue #119 documents three symptoms with the same root cause: no stage owns the final client-visible SSE sequence for multi-round gateway execution.

Without this, clients that initialize on response.created can reset more than once, clients that order or dedupe by sequence_number see the sequence reset mid-stream, and clients keyed by output_index can overwrite prior round output when later rounds reuse the same upstream indexes.

Validation

Ran the following locally from agent/issue-119-gateway-accumulator-verified:

• cargo fmt --check
• cargo test -p agentic-server-core --test web_search_tool_test multi_round_stream_has_single_lifecycle_and_monotonic_public_sequence
• cargo test -p agentic-server-core --test web_search_tool_test
• cargo clippy -p agentic-server-core --all-targets -- -D warnings
• cargo test -p agentic-server-core
• cargo clippy --all-targets -- -D warnings
• cargo test
• uvx pre-commit run --all-files

Results:

• Targeted regression: 1 passed, 17 filtered out
• Full web_search_tool_test: 18 passed
• agentic-server-core: 337 passed, 1 ignored
• Workspace tests: 376 passed, 1 ignored
• Clippy: no issues found
• Pre-commit: all hooks passed

Notes

This was previously explored in draft PR #129 and cancelled because it was considered too narrow relative to #121. This PR keeps the #119 behavior fix focused, but includes the regression coverage needed to prove the multi-round public stream invariants. It also updates the revived test for the current RequestPayload.cache_salt field added on main.

[ashwing]

The per-round lifecycle leak and colliding output_index are the sharp edges from #119, and one normalization boundary is the right shape. Traced it against the three acceptance criteria — single lifecycle ✅, contiguous output_index ✅, monotonic sequence_number holds on the success path. One gap and two minor notes inline.

[ashwing]

#119's criterion 2 was "monotonic sequence_number on every emitted frame." These error_sse_chunk frames (this arm and the Ok(Err) one) bypass the accumulator, so if a round already emitted frames (seq 0..k) and a later round fails, the client sees an error frame with no sequence_number after a numbered sequence. The accumulator is moved into the spawned task and only handed back on Ok(Ok(...)), so these arms don't have a handle to it today — worth restructuring so error frames get stamped too, or calling it out as a known follow-up before this closes #119.

[ashwing]

stream_events and stream_accumulator are separate Options but always travel together (both Some on the stream path, both None on blocking). This guard — and the matching one in emit_gateway_completed_events — silently drops every synthetic gateway event if a caller ever passes stream_events: Some with stream_accumulator: None. Bundling them into one type (the way StreamEmitContext already does on the upstream path) would make "both or neither" unrepresentable rather than convention.

[ashwing]

Two multi-round edges worth covering since this PR owns the invariant now: (a) a round that emits multiple public output items before the final round — the current test only ever has one item per round, so it never checks output_offset advancing by item count vs. +1; (b) the response.incomplete/max-rounds terminal path — confirms the terminal still gets a monotonic sequence_number after N rounds. Not blocking.

[maralbahari]

@harivilasp Thanks for the fix. Four structural requests, each building on the previous:

1. Move GatewayStreamAccumulator to its own file. It mirrors ResponseAccumulator, which lives in executor/accumulator.rs, so give it symmetric placement (e.g. executor/gateway_accumulator.rs) and take rebase_output_index (gateway.rs:165) and emit_sse_json (gateway.rs:349) with it. The duplicated Some/None branches in execute_and_emit_output_calls should clean up in the process.

2. Parse each SSE line once. Today every line is parsed up to three times: normalize_sse_line at upstream.rs:115, again in emit_stream_line (upstream.rs:143), and a third time via acc.process_sse_line (upstream.rs:89). normalize_sse_line already has the full JSON in hand and drops it; keep it on the frame, as a typed envelope rather than a bare Value:

#[derive(Serialize, Deserialize)]
pub struct WireEvent {
    #[serde(rename = "type")]
    pub event_type: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub sequence_number: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub output_index: Option<u64>,
    #[serde(flatten)]
    pub rest: serde_json::Map<String, Value>,
}

Renumbering becomes wire.sequence_number = Some(n) instead of value["sequence_number"] = ..., and the flatten catch-all keeps the relay lossless. Don't deserialize relayed events into the existing item types (OutputItem etc.): they have no flatten catch-alls and #[serde(other)] Unknown (types/io/output.rs:358) discards data, so re-emitting through them would silently strip unknown fields from the client stream. The namespace restorer is the only non-trivial adaptation, and it's close to mechanical: restore_response_value_with_map (tool/codex.rs:447) already goes through as_object_mut() at every step, so it can take the rest map directly. The WireEvent envelope is part of this PR; what can wait for a follow-up PR is extending the typing beyond it, promoting fields out of rest into dedicated structs per event and item type (with flatten catch-alls so the relay stays lossless).

3. Mirror ResponseAccumulator's two-level API. Give the gateway accumulator a line-based entry point process_sse_line that normalizes internally and delegates to a frame-based core process_event(&mut EventFrame), the same shape as accumulator.rs:211-229. In the fetch_stream_payload loop, normalize once and call the frame-based process_event of BOTH accumulators; calling the two line-based wrappers there would reintroduce the double parse.

4. Same frame currency for synthetic and deferred events. The gateway-built events (emit_gateway_start_events gateway.rs:360, emit_gateway_completed_events gateway.rs:411, terminal_response_chunk gateway.rs:143) are bare json!, yet every type they emit already has an SSEEventType variant. Add an EventFrame::synthetic(event_type, wire) constructor in events and route them through the same process_event, which also shares sequence numbering automatically. Likewise, pending_unnamed_function_events (upstream.rs:105) should park EventFrames instead of raw Strings that flush_pending_function_events (upstream.rs:211) re-parses on release.

[harivilasp]

will try to address by tomrrow