Plan 008: Behaviorally test the relay streaming core

[duyetbot]

Plan 008: Behaviorally test the relay streaming core

Executor instructions: Follow this plan step by step. Run every
verification command and confirm the expected result before moving to the
next step. If anything in the "STOP conditions" section occurs, stop and
report — do not improvise. When done, update the status row for this plan
in plans/README.md — unless a reviewer dispatched you and told you they
maintain the index.

Drift check (run first): git diff --stat 61ee3c7..HEAD -- src/relay.rs tests/cli.rs
If either file changed since this plan was written, compare "Current state"
excerpts against live code; on mismatch, STOP.

Status

• Priority: P2
• Effort: M
• Risk: LOW (tests only)
• Depends on: none
• Category: tests
• Planned at: commit 61ee3c7, 2026-08-26

Why this matters

The relay (anyr relay start) is the newest flagship feature: it bridges
cloud chat-completion requests to a local OpenAI-compatible server, streaming
responses back over one WebSocket. Its existing tests cover frame parsing and
config plumbing only. The actual serving behavior — forwarding to the local
target, incremental chunk streaming, UTF-8-safe splitting, cancel handling,
error frames when the target is down — has zero coverage. Any refactor can
regress it with green CI. This plan adds an end-to-end test harness: a local
WebSocket fake-cloud plus a stub HTTP target, driving the real
handle_request path.

Current state

• src/relay.rs key pieces (verified at 61ee3c7):
  • RequestFrame { id, path, body } parsed by parse_server_frame(&str) -> Option<Result<RequestFrame, String>> (line 168). "cancel" frames map to
    Some(Err(id)).
  • ClientFrame enum with to_json() (lines 106–162): Head{id,status,contentType},
    Chunk{id,data}, Done{id}, Error{id,message}, Hello{models,maxConcurrency}.
  • handle_request(tx, frame, target, cancel) (lines 494–611): POSTs
    frame.body to {target}{path} via a per-request ureq agent
    (connect timeout 15 s), sends Head then Chunk frames as the body streams
    in (8 KiB reads, UTF-8 boundary-aware flush via utf8_flush_len), then
    Done; non-2xx still relays status+body; transport error → Error frame;
    cancel flag checked between reads.
  • Worker spawn: spawn_request(state, frame, target) (line 831).
• Test conventions: inline mod tests in each module
  (src/relay.rs:1017); integration tests spawn the real binary from
  tests/cli.rs using env!("CARGO_BIN_EXE_anyr"), temp ANYROUTER_HOME,
  and env seams (ANYR_NO_UPDATE=1). No test in the repo currently opens a
  socket — this plan introduces the first local-server harness.
• Dependencies available: tungstenite (client AND server accept via
  tungstenite::handshake::server), std TcpListener. Both already in the
  dependency tree under the native feature.

Commands you will need

Purpose	Command	Expected on success
New tests	cargo test --locked --lib relay::tests	all pass incl. new
Full suite	cargo test --locked --all-targets	all pass

Scope

In scope:

• src/relay.rs (only appending tests + minimal #[cfg(test)]-gated helpers)
• Cargo.toml / Cargo.lock ONLY if you must add [dev-dependencies]
  (prefer zero additions; tungstenite is already there behind the native
  feature which is default)

Out of scope:

• Changing any production logic in relay.rs. If a test reveals a real bug,
  STOP and report it instead of fixing inside this plan.
• tests/cli.rs (integration binary tests) — keep the harness unit-level.

Git workflow

• Branch: advisor/008-relay-streaming-tests
• Commit style: e.g. test(relay): behavioral coverage for request streaming, cancel, and error paths
• Do NOT push or open a PR.

Steps

Step 1: Build the stub local target (HTTP) helper

Inside #[cfg(test)] mod tests, write a helper that binds 127.0.0.1:0,
returns (addr, join_handle), and serves N scripted responses:

fn spawn_stub_target(responses: Vec<(u16, String)>) -> (String, std::thread::JoinHandle<()>) {
    // TcpListener on 127.0.0.1:0; thread loop: accept, read headers until \r\n\r\n,
    // respond with the next scripted (status, body) as a valid HTTP/1.1 response
    // with Content-Length; close.
}

Keep it minimal — no chunked encoding needed; plain bodies suffice because
ureq's reader streams whatever arrives (Content-Length bodies exercise the
same read loop).

Step 2: Build the fake-cloud WS assertion helper

A second helper that accepts ONE WebSocket connection
(TcpListener + tungstenite::server::accept(stream)), receives the hello
frame, then:

struct FakeCloud { /* holds the accepted ws */ }
impl FakeCloud {
    fn send_request(&mut self, id: &str, path: &str, body: &str);
    fn send_cancel(&mut self, id: &str);
    fn drain_until_done(&mut self, id: &str, timeout_secs: u64) -> Vec<ClientFrame>;
        // reads text messages, parse_client_side via serde_json into ClientFrame-like
        // structs (define a small mirror struct or reuse to_json round-trip),
        // collects until Done{id} or Error{id} or timeout
}

Parsing outbound frames: define a #[derive(Deserialize)] struct OutFrame { r#type: String, id: String, status: Option<u16>, content_type: Option<String>, data: Option<String>, message: Option<String> } in the tests module.

Step 3: Write the behavioral tests

1. streams_local_response_back_as_head_chunk_done — stub returns
   (200, "hello world"); drive handle_request directly with a channel
   (no threads beyond the stub): create mpsc::channel, build RequestFrame,
   call handle_request(&tx, &frame, &target_url, &AtomicBool::new(false));
   assert drained frames == [Head{200}, Chunk{"hello world"}, Done].
2. non_2xx_from_target_relays_status_and_body — stub returns
   (502, "bad gateway"); expect Head{502} + Chunk + Done.
3. unreachable_target_produces_error_frame — target URL points at a closed
   port (bind then drop the listener); expect exactly one Error frame whose
   message contains "Local target unreachable".
4. cancel_flag_stops_streaming — stub returns a LARGE body (e.g. 1 MiB);
   set cancel=true BEFORE calling handle_request; assert NO frames were sent
   (or only none beyond Head — assert empty vec of Chunk/Done).
5. utf8_split_body_round_trips — stub returns a multibyte-heavy body
   (e.g. "héllo 🚀".repeat(1000)); collect all Chunk data and concatenate;
   assert equals original string (proves lossless reassembly).

Use direct handle_request calls (public within crate? if private, make the
tests module access it as today — same module tree, fine).

Verify: cargo test --locked --lib relay::tests → ≥5 new tests pass.

Step 4: Full suite + flake guard

Run the new tests 5 times to shake port/timing flakes:

Verify: for i in 1 2 3 4 5; do cargo test --locked --lib relay::tests || break; done → 5 consecutive passes.

Then cargo test --locked --all-targets → exit 0.

Test plan

The five tests above ARE the plan. Structural pattern: follow existing relay
unit tests' style (plain asserts, no snapshotting); helpers stay inside
mod tests.

Done criteria

ALL must hold:

• cargo test --locked --lib relay::tests lists ≥5 new passing tests with the names above.
• 5 consecutive reruns pass (no flakes).
• cargo test --locked --all-targets exits 0.
• Production code diff in relay.rs is EMPTY (git diff 61ee3c7..HEAD -- src/relay.rs shows only additions inside #[cfg(test)] region — verify by reading the diff hunk headers).
• No new dependencies added (unless Step-scope allowed dev-dep and you document why).

STOP conditions

Stop and report if:

• Driving handle_request requires making anything public or refactoring
  production signatures (that would violate tests-only scope — report the
  friction; the reviewer will decide).
• A test exposes a REAL bug in streaming/cancel/error handling (report the
  exact scenario and observed frames — do not fix).
• Port binding is unavailable/flaky in your sandbox after retries.

Maintenance notes

• The stub helpers are reusable for future relay features (pool flow, auth on
  connect); keep them small and self-contained.
• Reviewer: confirm assertions check FRAME SEQUENCE and CONTENT, not just
  "something arrived".