refactor(server): make the SSE keepalive attach unforgettable via a shared constructor instead of testing for it #1107

Summary

Five route handlers each attach the SSE keepalive by hand with the same three-line tail. A sixth route that forgot would compile and pass the whole suite. A TODO at src/server/streaming_tests.rs:358 asks for an axum integration test to cover this, but the investigation below argues the test is the wrong tool: replace the five copy-paste sites with one shared constructor so forgetting the keepalive stops being expressible.

Background

src/server/streaming_tests.rs:358-364 records the coverage gap and asks for an axum integration harness to drive Sse::new(stream).keep_alive(...) end to end, skipped "to avoid the additional test infra."

Five route handlers each end with the byte-identical tail Sse::new(stream).keep_alive(keepalive.into_inner()).into_response():

• src/server/routes/chat.rs:973
• src/server/routes/completions.rs:443
• src/server/routes/native_completion.rs:254
• src/server/routes/responses.rs:853
• src/server/routes/anthropic.rs:553

No route varies the stream type, the keepalive, or the ordering; this is five copies of one expression. The property that can actually regress ("does every route attach the keepalive?") is caught by an integration test only for the routes someone remembers to write a test for, so a test-per-route is the same duplication moved into the test module. (The other half of the TODO's concern, "does axum emit a comment frame while idle," tests axum's KeepAlive, which upstream already covers.)

Proposed Solution

Delete the duplication instead of testing it. Add one constructor next to the channel helpers in src/server/streaming.rs:

/// Build the streaming HTTP response, attaching the keepalive that the matching
/// `*_sse_channel` produced. Route handlers call this instead of assembling
/// `Sse::new(..).keep_alive(..)` by hand, so the keepalive cannot be forgotten.
pub(crate) fn sse_response<S>(stream: S, keepalive: impl IntoKeepAlive) -> Response
where S: Stream<Item = ...> + Send + 'static
{
    Sse::new(stream).keep_alive(keepalive.into_keep_alive()).into_response()
}

Give the three keepalive newtypes a one-line IntoKeepAlive impl each. This preserves the property #1105 asks to keep (a route cannot attach another surface's keepalive, because the type it receives comes from the channel constructor it called) while removing the hand-assembly. The five call sites become sse_response(stream, keepalive), and forgetting the keepalive stops being expressible: there is no other public way to turn one of these streams into a response. Then delete the TODO at streaming_tests.rs:358-364 with a one-line note that the hazard was removed structurally.

Implementation Notes

• tower = { version = "0.4", features = ["util"] } is already a direct dependency (Cargo.toml:224), so the fallback integration-test approach needs no new dependency; it is still the weaker option.
• Each of the three newtypes already exposes into_inner() (streaming.rs:81, streaming_responses.rs:98, streaming_anthropic.rs:90).
• Complementary to refactor(server): consolidate the three duplicated SSE keepalive constants under one proxy-timeout assertion #1105 (which consolidates the interval constants). Same files, cleanest landed after refactor(server): consolidate the three duplicated SSE keepalive constants under one proxy-timeout assertion #1105, but neither blocks the other.

Acceptance Criteria

• No Sse::new(..).keep_alive(..) remains in src/server/routes/.
• All three keepalive newtypes reach the shared constructor.
• Deleting .keep_alive(..) from the shared constructor is the only way to break the invariant, and it breaks it for every route at once.
• The TODO at streaming_tests.rs:358-364 is removed.
• cargo test --workspace --profile test-fast --features metal,accelerate server:: is green.

Original Suggestion

Title: refactor(server): make the SSE keepalive attach unforgettable instead of testing for it

Supersedes the framing in the TODO at src/server/streaming_tests.rs:358, which asks
for an axum integration test. Investigation below argues the test is the wrong tool.

Problem / Background

src/server/streaming_tests.rs:358-364 records a known coverage gap:

// TODO: add a test that verifies the SSE keepalive comment frame is emitted
// before the first event arrives (LOW #2). This requires an axum
// integration test harness to drive `Sse::new(stream).keep_alive(...)` end-to-end
// and read raw SSE frames from the response body. ...
// skipped here to avoid the additional test infra.

The risk it names is real. Five route handlers each attach the keepalive by hand, and a
sixth that forgot would compile and pass the whole suite:

• src/server/routes/chat.rs:973
• src/server/routes/completions.rs:443
• src/server/routes/native_completion.rs:254
• src/server/routes/responses.rs:853
• src/server/routes/anthropic.rs:553

What the investigation found

1. The five sites are byte-identical. Every one of them is the same three-line tail:

Sse::new(stream)
    .keep_alive(keepalive.into_inner())
    .into_response()

No route varies the stream type, the keepalive, or the ordering. This is five copies of
one expression, not five route-specific constructions.

2. The TODO's stated cost is lower than when it was written.
tower = { version = "0.4", features = ["util"] } is already a direct dependency
(Cargo.toml:224), and util is exactly the feature providing ServiceExt::oneshot. So
the suggested harness needs no new dependency and no cargo deny exposure. Worth knowing,
but see below — it turns out not to matter.

3. The property worth testing splits in two, and neither half wants this test.

• Does axum emit a comment frame while the stream is idle? Testable in a few lines —
  but this tests axum's KeepAlive, not mlxcel. Upstream already tests it.
• Does every route attach the keepalive? This is the one that can actually regress, and
  an integration test catches it only for the routes someone remembers to write a test
  for. A test per route is five tests guarding five copy-paste sites — the same
  duplication, moved into the test module.

Proposed solution

Delete the duplication instead of testing it. Add one constructor next to the channel
helpers in src/server/streaming.rs:

/// Build the streaming HTTP response, attaching the keepalive that the matching
/// `*_sse_channel` produced. Route handlers call this instead of assembling
/// `Sse::new(..).keep_alive(..)` by hand, so the keepalive cannot be forgotten.
pub(crate) fn sse_response<S>(stream: S, keepalive: impl IntoKeepAlive) -> Response
where S: Stream<Item = ...> + Send + 'static
{
    Sse::new(stream).keep_alive(keepalive.into_keep_alive()).into_response()
}

The three keepalive newtypes each need a one-line IntoKeepAlive impl. This preserves the
property #1105 asks to keep — that a route cannot attach another surface's keepalive,
because the type it receives comes from the channel constructor it called — while removing
the hand-assembly the TODO was worried about.

Then the five call sites become sse_response(stream, keepalive), and forgetting the
keepalive stops being expressible: there is no other public way to turn one of these
streams into a response.

Finally, delete the TODO at src/server/streaming_tests.rs:358-364 with a one-line note
saying the hazard was removed structurally rather than covered by a test.

Acceptance criteria

• No Sse::new(..).keep_alive(..) remains in src/server/routes/.
• All three keepalive newtypes reach the shared constructor.
• Deleting .keep_alive(..) from the shared constructor is the only way to break the
  invariant, and it breaks it for every route at once.
• The TODO at streaming_tests.rs:358-364 is removed.
• cargo test --workspace --profile test-fast --features metal,accelerate server:: is green.

Relationship to #1105

#1105 consolidates the three duplicated interval constants; this consolidates the three
duplicated attach expressions. They touch the same files and are cleanest landed in
order (#1105 first), but neither blocks the other. #1105 states the three newtypes should
stay distinct — this issue keeps them distinct and unifies only the call shape.

Alternative, if the constructor is rejected

If a maintainer prefers the newtypes stay fully independent with no shared trait, the
fallback is the TODO's original plan: one tower::ServiceExt::oneshot test per route,
with a short keepalive interval so it runs in milliseconds. That is five tests and a test
module that must be updated whenever a route is added — strictly more to maintain than the
constructor, which is why it is the fallback rather than the proposal.