From f13705e525eca7683625158b495d62b7b37973e2 Mon Sep 17 00:00:00 2001 From: Matthew Jackson Date: Sat, 8 Aug 2026 11:22:18 -0700 Subject: [PATCH 1/2] release: keep Cargo.lock in step with the pinned busbar ref, and never publish an assetless release The v1.0.4 release run failed on every target with: error: cannot update the lock file .../Cargo.lock because --locked was passed Cargo.toml's busbar-* dependencies are path deps into a sibling ../busbarAI checkout. release-on-upstream re-pins that checkout to a new commit in .busbar-ref but never regenerated Cargo.lock, so when busbar 1.5.3 pulled `valuable` in behind `tracing`, the committed lock described a graph that no longer resolved and release.yml's `--locked` build correctly refused to proceed. `--locked` is kept deliberately: it is what makes a release build exactly the graph that was resolved at pin time. Instead the lock is now refreshed in the same step that re-pins, against the exact commit being pinned, and committed alongside .busbar-ref so the two can never drift apart. Plain cargo (no --locked) is used for that refresh so the resolution is minimal, and the result is proven to satisfy --locked before anything is committed or tagged. Cargo.lock here is that refresh for the busbar 1.5.3 ref already recorded on main, which is otherwise unbuildable. Separately, create-release published the Release before the build matrix ran, so a total build failure left a tag whose releases/latest carried zero assets. verify-assets already detected that, but only after the empty release was public. The Release is now created as a draft, which releases/latest and releases/tags/ do not resolve, and verify-assets promotes it to published only once assets are provably attached. A release is now either complete or absent. Also brings two test fixtures up to busbar 1.5.3: RoutingRequest/Candidate gained request_id and signals, and inline module entries under auth.admin_auth were retired in favour of a named identity-providers definition. --- .github/workflows/release-on-upstream.yml | 38 +++++++++++++++++-- .github/workflows/release.yml | 46 +++++++++++++++-------- Cargo.lock | 14 ++++++- tests/e2e.rs | 7 ++++ tests/full_stack_e2e.rs | 9 ++++- 5 files changed, 91 insertions(+), 23 deletions(-) diff --git a/.github/workflows/release-on-upstream.yml b/.github/workflows/release-on-upstream.yml index 041e48e..789497d 100644 --- a/.github/workflows/release-on-upstream.yml +++ b/.github/workflows/release-on-upstream.yml @@ -156,6 +156,36 @@ jobs: *) echo "::error::dry-run: ${TAG} is not a valid semver tag"; exit 1 ;; esac + # Re-pinning .busbar-ref moves the SIBLING ../busbarAI checkout that every busbar-* entry in + # Cargo.toml is a path dependency on, and that move can change the resolved dependency graph + # (busbar 1.5.2 -> 1.5.3, for instance, pulled `valuable` in behind `tracing`). Cargo.lock is + # committed in this repo and release.yml builds with `--locked`, so a lock left describing the + # OLD graph makes `cargo build --locked` refuse to update it and every release target fails. + # + # So the lock is refreshed HERE, against the exact commit being pinned, and committed in the + # SAME commit as .busbar-ref. `--locked` in release.yml is deliberately kept: it is what + # guarantees a release builds exactly the graph resolved and reviewed at pin time, and dropping + # it would turn this loud failure into a silent drift. + - name: Refresh Cargo.lock against the busbar ref being pinned + if: steps.guard.outputs.exists == 'no' && steps.resolve.outputs.in_sha != '' && github.event.inputs.dry_run != 'true' + env: + IN_SHA: ${{ steps.resolve.outputs.in_sha }} + run: | + set -euo pipefail + # Cargo.toml's path deps point at ../busbarAI, a SIBLING of this checkout. actions/checkout + # cannot write outside the workspace, and GetBusbar/busbar is public, so clone it directly. + sib="$(cd .. && pwd)/busbarAI" + rm -rf "$sib" + git clone --quiet --no-checkout https://github.com/GetBusbar/busbar.git "$sib" + git -C "$sib" checkout --quiet "${IN_SHA}" + # Plain cargo (NO --locked) performs the MINIMAL resolution: every existing registry pin is + # preserved and only what the new busbar graph actually requires is added or moved. + cargo metadata --format-version 1 >/dev/null + # Prove the refreshed lock satisfies the exact flag release.yml will build under, BEFORE we + # commit and tag. If this fails, no tag is pushed and no phantom release can be created. + cargo metadata --format-version 1 --locked >/dev/null + echo "::notice::Cargo.lock is fresh against busbar ${IN_SHA} and satisfies --locked" + - name: Record the new busbar ref in .busbar-ref if: steps.guard.outputs.exists == 'no' && steps.resolve.outputs.in_sha != '' && github.event.inputs.dry_run != 'true' env: @@ -165,10 +195,12 @@ jobs: run: | set -euo pipefail printf '%s %s\n' "${IN_SHA}" "${IN_VER}" > .busbar-ref - if git diff --quiet .busbar-ref; then - echo "::notice::.busbar-ref already at ${IN_SHA} ${IN_VER}, no commit needed" + # Cargo.lock rides along with the ref it was resolved against: the two must never be + # committed apart, or release.yml's --locked build sees a lock for a different busbar. + if git diff --quiet -- .busbar-ref Cargo.lock; then + echo "::notice::.busbar-ref already at ${IN_SHA} ${IN_VER} and Cargo.lock already fresh, no commit needed" else - git add .busbar-ref + git add .busbar-ref Cargo.lock git commit -m ".busbar-ref: record busbar ${IN_VER} (${IN_SHA}) for the ${TAG} release" git push origin HEAD:main fi diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index fea3ffd..df0fb0b 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -31,18 +31,25 @@ permissions: jobs: # Create the Release first so the parallel per-target upload jobs have something to attach to # (uploading from a matrix without a pre-existing release races -> "release not found"). + # + # It is created as a DRAFT. A draft is addressable by tag to `gh release upload` / `gh release + # view` with this token, but it is NOT returned by the public `releases/latest` or + # `releases/tags/` endpoints -- so while the build matrix is still running, and for good if it + # fails, nothing following `releases/latest` can observe an assetless release. `verify-assets` + # publishes it only once assets are actually attached, which is what makes a release here either + # complete or absent, never an empty shell. create-release: runs-on: ubuntu-latest steps: - uses: actions/checkout@v7 - - name: Create GitHub Release + - name: Create GitHub Release (draft) env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} run: | gh release create "${GITHUB_REF_NAME}" \ --repo "${GITHUB_REPOSITORY}" \ --title "busbar-webrequest ${GITHUB_REF_NAME}" \ - --verify-tag --generate-notes \ + --draft --verify-tag --generate-notes \ || gh release view "${GITHUB_REF_NAME}" --repo "${GITHUB_REPOSITORY}" # One signed .tar.gz per target: {cdylib + manifest.json}, packed by busbar-plugin-pack and @@ -153,19 +160,20 @@ jobs: with: subject-path: "plugin-dist/*.tar.gz" - # PHANTOM-RELEASE GUARD: assert the published Release actually carries assets before we treat this - # as a real release. The per-target build/upload jobs run with fail-fast:false, and `create-release` - # always makes the (initially empty) Release up front — so a build/pack failure on EVERY target - # (e.g. a stale Cargo.lock tripping `--locked`, as happened on the first v1.0.3 cut) leaves a tag + - # Release with ZERO assets: a "phantom" that silently breaks busbar's plugin-registry-gate. This job - # fails the whole release run loud if assets == 0, so a phantom can never ship (or notify marketing) - # unnoticed again. It depends on the build matrix but does NOT inherit its fail-fast:false — one - # green target is enough to have assets, but zero across the board must hard-fail here. + # PHANTOM-RELEASE GATE: this is the only step that turns the draft into a published release, and it + # does so only after proving assets are attached. The per-target build/upload jobs run with + # fail-fast:false and `create-release` makes the (initially empty) Release up front, so a build/pack + # failure on EVERY target (e.g. a stale Cargo.lock tripping `--locked`) leaves a Release with ZERO + # assets. Previously that Release was published immediately and such a failure shipped a "phantom": + # a tag whose `releases/latest` carried nothing, silently breaking busbar's plugin-registry-gate. + # Now the empty Release is a draft nothing can resolve, and this job either publishes it or fails + # the run with the draft left unpublished. It depends on the build matrix but does NOT inherit its + # fail-fast:false: one green target is enough to have assets, zero across the board must hard-fail. verify-assets: needs: [webrequest-plugin] runs-on: ubuntu-latest steps: - - name: Assert the Release has at least one asset + - name: Assert the Release has at least one asset, then publish it env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} run: | @@ -175,13 +183,19 @@ jobs: --json assets --jq '.assets | length')" echo "Release ${GITHUB_REF_NAME} has ${count} asset(s)." if [ "${count}" -eq 0 ]; then - echo "::error::PHANTOM RELEASE: ${GITHUB_REF_NAME} was published with 0 assets." \ - "Every build/pack target failed to upload a tarball. Failing the release run so this" \ - "tag is not mistaken for a real release by busbar's plugin-registry-gate. Fix the" \ - "build (check Cargo.lock freshness vs --locked and the plugin cdylib build step)," \ - "delete this tag+release, and re-cut." >&2 + echo "::error::PHANTOM RELEASE PREVENTED: ${GITHUB_REF_NAME} has 0 assets." \ + "Every build/pack target failed to upload a tarball. The release is still a DRAFT," \ + "so it was never visible to anything following releases/latest and there is no" \ + "phantom to clean up -- just fix the build (check Cargo.lock freshness vs --locked" \ + "and the plugin cdylib build step) and re-run. Failing the release run." >&2 exit 1 fi + # Only now, with assets provably attached, does this stop being a draft and become the + # release that `releases/latest` resolves to. + gh release edit "${GITHUB_REF_NAME}" \ + --repo "${GITHUB_REPOSITORY}" \ + --draft=false --latest + echo "::notice::Published ${GITHUB_REF_NAME} with ${count} asset(s)." # Instant marketing-site rebuild the moment this plugin ships a real release -- marketing's # deploy.yml listens for this exact repository_dispatch event type (plus its own daily-poll diff --git a/Cargo.lock b/Cargo.lock index c83c7e0..a41cf51 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -128,6 +128,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.11.0", + "smallvec", "zeroize", ] @@ -166,6 +167,8 @@ dependencies = [ "busbar-api", "busbar-plugin-abi", "serde_json", + "tracing", + "tracing-core", ] [[package]] @@ -772,9 +775,9 @@ checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" [[package]] name = "libloading" -version = "0.8.9" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55" +checksum = "754ca22de805bb5744484a5b151a9e1a8e837d5dc232c2d7d8c2e3492edc8b60" dependencies = [ "cfg-if", "windows-link", @@ -1497,6 +1500,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", ] [[package]] @@ -1541,6 +1545,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "version_check" version = "0.9.5" diff --git a/tests/e2e.rs b/tests/e2e.rs index ef695d7..3e71e57 100644 --- a/tests/e2e.rs +++ b/tests/e2e.rs @@ -267,6 +267,11 @@ fn cfg(url: &str) -> String { fn req_with_prompt(text: &str) -> RoutingRequest<'static> { RoutingRequest { + // busbar 1.5.3 added the correlation id and the declared-signal bag to the hook projection. + // These fixtures pin a fixed id and an EMPTY bag: no test here declares a signal, and an + // empty bag serialises to nothing, so the forwarded envelope is unchanged by their presence. + request_id: 1, + signals: busbar_api::SignalBag::new(), pool: "p", ingress_protocol: "anthropic", requested_model: None, @@ -299,6 +304,8 @@ fn cand(idx: usize) -> Candidate<'static> { available_concurrency: 1, budget_remaining: None, rate_headroom: None, + // Same contract as RoutingRequest::signals above: empty unless a consumer declares one. + signals: busbar_api::SignalBag::new(), } } diff --git a/tests/full_stack_e2e.rs b/tests/full_stack_e2e.rs index 9510cc8..89b9b00 100644 --- a/tests/full_stack_e2e.rs +++ b/tests/full_stack_e2e.rs @@ -327,15 +327,20 @@ mockup: // `auth.chain: []` (open relay) on purpose: this test's whole point is the admin-install -> // plugin-load -> hook-invocation -> webhook round trip, not the client auth chain (covered // elsewhere in busbar's own suite) — narrowing scope here keeps the failure surface honest. + // + // `admin-tokens` is DEFINED once under `identity-providers:` and REFERENCED by bare name from + // `auth.admin_auth`. busbar 1.5.3 retired the inline-module form this used to use, and refuses + // to boot a config still carrying it, so the inline shape would fail this test at startup. let config_yaml = format!( r#" listen: "127.0.0.1:{data_port}" admin_listen: "127.0.0.1:{admin_port}" +identity-providers: + admin-tokens: {{ module: admin-tokens, token: {{ env: BUSBAR_E2E_ADMIN_TOKEN }} }} auth: chain: [] signing_key: {{ env: BUSBAR_SIGNING_KEY }} - admin_auth: - - admin-tokens: {{ token: {{ env: BUSBAR_E2E_ADMIN_TOKEN }} }} + admin_auth: [admin-tokens] plugins: enabled: true dir: "{plugins_dir}" From 2e933c0048f7f6517fdefa9fb08e9c5a3a68f17d Mon Sep 17 00:00:00 2001 From: matthew Date: Sat, 8 Aug 2026 16:07:11 -0700 Subject: [PATCH 2/2] tests: track busbar 1.5.3's per-stage notify contract full_stack_e2e asserted that one real request produces exactly one webhook call. Against busbar 1.5.3 it produces four. That is a documented breaking change, not a regression: busbarAI CHANGELOG.md, "[1.5.3] Breaking changes": "A hand-written hook with no stage list now fires at all four stages rather than once per request; set `phase: [request]` for the old behaviour." The test registered its tap over the admin API with no stage scoping, so it was getting the new all-stages fan-out: the stage-less request envelope plus `stage.at` of candidate, routing and response. The assertion was encoding the pre-1.5.3 default, not a property of this plugin. Three changes. 1. Pin the tap to the request stage with `at: "request"`. `phase:` is the config-file spelling of that opt-out; `at:` is the admin API's, and busbar resolves stages as: a non-empty `phase:` list wins, else the single `at:`, else all four core stages. The pin restores "one call per request" as a correct expectation rather than a stale one, and it keeps this test aimed at what it exists to prove: that the real engine's wire projection, with real prompt content under the `prompt: ro` grant, reaches a real webhook target. Only the request-stage envelope carries content, so only it can carry that proof. 2. Settle the capture instead of breaking on the first sighting. A tap is fire-and-forget: busbar spawns the POST detached, so the deliveries for one request land over a window. The old loop returned the instant the sink went non-empty, which sampled the middle of that window. That is why the count assertion could pass by accident on slow delivery: it was guarding "at least one call arrived and we looked early", not the count. The new helper requires the count to hold steady across consecutive polls, so a late extra envelope fails the assertion rather than being raced past. 3. Add tests/stage_fanout_e2e.rs, which covers the new contract positively. It registers a tap with NO stage scoping (the configuration an operator writes when they have not thought about stages) and asserts what such a hook really receives: all four core stages for one request; prompt content on the request envelope and on none of the others; the stage envelopes shape-only (no candidates, no messages, no user); the documented per-stage fields (remaining_candidates, attempt_number + model, outcome + status); and one shared request_id join key across all four, which is the only thing that lets a sidecar correlate four POSTs back into one request. Without 3, the repo would merely tolerate the change: full_stack_e2e would go green because it opted out, and nothing would assert the fan-out happens or has the shape sidecar authors are told to expect. Red before green, both captured: full_stack_e2e failed on the count assertion before the pin and passes after; stage_fanout_e2e fails on the stage-set assertion when its hook is scoped to the request stage and passes unscoped. --- tests/full_stack_e2e.rs | 116 +++++-- tests/stage_fanout_e2e.rs | 684 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 781 insertions(+), 19 deletions(-) create mode 100644 tests/stage_fanout_e2e.rs diff --git a/tests/full_stack_e2e.rs b/tests/full_stack_e2e.rs index 89b9b00..70db18a 100644 --- a/tests/full_stack_e2e.rs +++ b/tests/full_stack_e2e.rs @@ -24,9 +24,12 @@ //! 3. `POST /api/v1/admin/plugins` the signed tarball — confirm `201` + `trust: "trusted"`. //! 4. `POST /api/v1/admin/plugins/reload` — confirm the plugin registry picks it up. //! 5. `POST /api/v1/admin/hooks` — register a `global` tap naming this plugin's alias -//! (`webrequest`), `settings.url` pointing at a local mock HTTP server standing in for the -//! operator's webhook target (this repo's own established mocking convention — see -//! `tests/e2e.rs`'s `mock_target`/`capturing_target`). +//! (`webrequest`), pinned to the REQUEST stage with `at: "request"`, `settings.url` pointing +//! at a local mock HTTP server standing in for the operator's webhook target (this repo's own +//! established mocking convention, see `tests/e2e.rs`'s `mock_target`/`capturing_target`). +//! The stage pin is deliberate and documented at the registration site: since busbar 1.5.3 an +//! unscoped hook fires at every core stage, and only the request-stage envelope carries prompt +//! content. `tests/stage_fanout_e2e.rs` covers the unscoped, all-stages case. //! 6. `POST /{model}/v1/messages` — a real Anthropic-shaped chat request through the real router, //! against a real (mocked) upstream model. //! 7. Confirm the mock webhook server received a REAL POST with the REAL engine-built envelope @@ -281,6 +284,61 @@ async fn wait_for_healthz(admin_addr: &str) { } } +/// Poll the captured-webhook sink until the observed CALL COUNT has stopped moving, then return the +/// settled batch. Returns as soon as a non-zero count has held steady across [`STABLE_POLLS`] +/// consecutive polls; panics only if NOTHING ever arrived within `timeout`. +/// +/// Why a settle and not a "first non-empty wins" read: a tap is fire-and-forget. Busbar spawns the +/// POST on a detached task and never awaits it, so the calls for a single request land over a short +/// window with no ordering or delivery guarantee. Reading the sink the instant it turns non-empty +/// samples the MIDDLE of that window, which makes any assertion about the batch (its size, which +/// stages are in it, which envelope carries what) a coin flip that usually lands the way the author +/// expected. Requiring the count to hold still first turns "we looked too early" into a real, +/// reproducible observation. +/// +/// A late arrival resets the streak, so an EXTRA call is surfaced to the caller's assertion rather +/// than raced past. On timeout with at least one call in hand, the settled-so-far batch is returned +/// and the caller's own assertion reports the mismatch (a more useful failure than a generic +/// timeout); with zero calls it panics, preserving the original "the chain never completed" +/// diagnostic. +async fn settle_captured( + captured: &Arc>>, + timeout: Duration, +) -> Vec { + /// Gap between polls. + const POLL: Duration = Duration::from_millis(100); + /// Consecutive equal, non-zero counts required before the batch is considered settled. + const STABLE_POLLS: u32 = 5; + + let deadline = std::time::Instant::now() + timeout; + let mut last_len = 0usize; + let mut stable = 0u32; + loop { + tokio::time::sleep(POLL).await; + let len = captured.lock().unwrap().len(); + if len > 0 && len == last_len { + stable += 1; + if stable >= STABLE_POLLS { + return captured.lock().unwrap().clone(); + } + } else { + // Growth (or still nothing): restart the streak against the new count. + stable = 0; + last_len = len; + } + if std::time::Instant::now() > deadline { + if last_len > 0 { + return captured.lock().unwrap().clone(); + } + panic!( + "the mock webhook target never received a call from the real, admin-API-installed \ + webrequest hook within {timeout:?}: the real install -> load -> invoke -> webhook \ + chain did not complete" + ); + } + } +} + /// THE full-stack proof. `#[ignore]`-free and part of the normal `cargo test` run: this is the /// "prod ready" bar, not an opt-in extra. #[tokio::test(flavor = "multi_thread")] @@ -450,12 +508,33 @@ models: // than a listing row would be. // ── 6. Register a global hook naming this plugin, over the REAL admin API ─────────────────── + // `at: "request"` PINS this tap to the request stage, and it is load-bearing for the + // exactly-one-call assertion below. Busbar 1.5.3 made an unscoped hook fire at EVERY core + // stage (request, candidate, routing, response) instead of once per request: + // + // busbarAI CHANGELOG.md, "[1.5.3] Breaking changes": + // "A hand-written hook with no stage list now fires at all four stages rather than once + // per request; set `phase: [request]` for the old behaviour." + // + // `phase:` is the config-file spelling of that opt-out; on THIS surface, the admin API, the + // equivalent is the single-valued `at:` (busbar resolves a hook's stages as: a non-empty + // `phase:` list wins, else the single `at:`, else all four core stages - see + // `HookCfg::fires_at_stage` in the engine). Without the pin this tap observes four envelopes + // and only one of them carries prompt text, which is not what this test is here to prove. + // + // What this test proves is the REQUEST-stage envelope specifically: that the real engine's own + // wire projection, carrying real prompt content under the `prompt: ro` grant, reached a real + // webhook target through a real admin-API plugin install. The candidate/routing/response + // envelopes are shape-only by construction (the engine sends no prompt, no identity, no + // candidates on them), so they cannot carry that proof. `tests/stage_fanout_e2e.rs` is the + // test that covers them, asserting the 1.5.3 fan-out positively. let hook_body = serde_json::json!({ "name": "webrequest-e2e-tap", "config": { "kind": "tap", "plugin": "webrequest", "global": true, + "at": "request", "prompt": "ro", "settings": { "url": webhook_url }, } @@ -520,26 +599,25 @@ models: ); // ── 8. Confirm the mock webhook ACTUALLY received the real HTTP round trip ────────────────── - let deadline = std::time::Instant::now() + Duration::from_secs(10); - let received = loop { - let snapshot = captured.lock().unwrap().clone(); - if !snapshot.is_empty() { - break snapshot; - } - if std::time::Instant::now() > deadline { - panic!( - "the mock webhook target never received a call from the real, admin-API-installed \ - webrequest hook within 10s — the real install -> load -> invoke -> webhook chain \ - did not complete" - ); - } - tokio::time::sleep(Duration::from_millis(100)).await; - }; + // SETTLE, do not break on the first sighting. A tap is fire-and-forget: busbar spawns the POST + // detached and never waits for it, so the deliveries for one request land over a short window + // and `captured` grows asynchronously. A loop that stopped as soon as the vec was non-empty + // read a partially delivered batch, which is why the stale `len() == 1` assertion below could + // pass by accident on slow delivery: it was not really guarding the count at all, it was + // guarding "at least one call arrived, and we looked before the rest showed up". + // + // So: wait for at least one call, then require the count to hold steady across consecutive + // polls before trusting it. Any late arrival resets the streak, so an extra envelope is + // observed and fails the assertion instead of being raced past. + let received = settle_captured(&captured, Duration::from_secs(10)).await; assert_eq!( received.len(), 1, - "exactly one real webhook call expected for one real request: {received:?}" + "exactly one real webhook call expected for one real request: a hook pinned to the \ + request stage (`at: \"request\"`) observes each request once. More than one here means \ + the stage pin stopped taking effect and this tap is seeing busbar 1.5.3's per-stage \ + fan-out: {received:?}" ); let envelope = &received[0]; assert_eq!( diff --git a/tests/stage_fanout_e2e.rs b/tests/stage_fanout_e2e.rs new file mode 100644 index 0000000..60c5d0f --- /dev/null +++ b/tests/stage_fanout_e2e.rs @@ -0,0 +1,684 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! THE 1.5.3 STAGE FAN-OUT PROOF: an UNSCOPED hook, registered over the real admin API into a real +//! running `busbar`, observes one request at EVERY core stage, and the stage envelopes are +//! shape-only while the request envelope is the one carrying prompt content. +//! +//! `tests/full_stack_e2e.rs` (the sibling file) proves the install -> load -> invoke -> webhook +//! chain for the REQUEST stage specifically, and pins its tap there with `at: "request"` so it +//! observes exactly one call. That pin is the documented opt-out from the change this file exists +//! to cover: +//! +//! busbarAI CHANGELOG.md, "[1.5.3] Breaking changes": +//! "A hand-written hook with no stage list now fires at all four stages rather than once per +//! request; set `phase: [request]` for the old behaviour." +//! +//! Without this file the plugin repo would merely TOLERATE that change: `full_stack_e2e` would go +//! green because it opted out, and nothing anywhere would assert that the fan-out actually happens +//! or that its envelopes have the shape a sidecar author is told to expect. That is the gap this +//! closes. It is a POSITIVE test: it registers a tap with no stage scoping at all, the exact +//! configuration an operator writes when they have not thought about stages, and asserts what such +//! a hook really receives. +//! +//! What is asserted, and why each part matters to a sidecar author: +//! +//! 1. All four core stages arrive for ONE request: the stage-less request envelope, plus +//! `stage.at` values covering `candidate`, `routing` and `response`. A sidecar sized for +//! one call per request is sized wrong by a factor of four. +//! 2. Exactly ONE envelope carries prompt text, and it is the stage-less (request-stage) one. +//! The engine sends `system`/`messages`/`user` as absent on stage taps regardless of grant, +//! so a `prompt: ro` hook does NOT get content four times over. A sidecar that screens or +//! logs content keys off the request envelope; one that assumed every notify carries content +//! would silently screen nothing on three quarters of its traffic. +//! 3. The stage envelopes are shape-only in the other direction too: no `candidates`, and the +//! documented per-stage fields are present (`remaining_candidates` on `candidate`, +//! `attempt_number` + `model` on `routing`, `outcome` + `status` on `response`). +//! 4. Every envelope carries the SAME `request.request_id`. That is the documented join key, and +//! it is the only thing that lets a sidecar correlate four separate POSTs back into one +//! request. If it ever drifted per stage, stage taps would be unusable for audit. +//! +//! The harness below (binary discovery, signing, boot, mocks) is a deliberate copy of +//! `full_stack_e2e.rs`'s, following this repo's established convention for integration tests: each +//! `tests/*.rs` is its own independent test binary and carries its own helpers rather than sharing +//! a `tests/common` module (see the `plugin_path()` doc comment in `tests/e2e.rs`, which makes the +//! same call for the same reason). Only the hook REGISTRATION and the ASSERTIONS differ, and both +//! differences are the point of the file. + +use axum::routing::post; +use axum::Router; +use busbar_plugin_sign::{HookNeeds, Manifest, NeedLevel, SigningKey}; +use std::collections::BTreeSet; +use std::path::PathBuf; +use std::process::{Child, Command, Stdio}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +/// Fixed ed25519 signing secret (64 hex = 32 bytes) for this e2e test. 1.5.1 requires an +/// explicit signing key to mint virtual keys; busbar no longer auto-generates one. +const TEST_SIGNING_KEY: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +/// The sibling `busbarAI` monorepo checkout - same interim path convention as `Cargo.toml`'s +/// `busbar-plugin-sdk` dependency. +fn busbarai_root() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../busbarAI") +} + +/// Locate (building on demand) the real `busbar` engine binary from the sibling checkout. Same +/// discipline as `full_stack_e2e.rs`: under CI a missing checkout or a failed build is a hard +/// panic, never a silent skip, because a test that quietly no-ops is worse than no test. +fn busbar_bin() -> Option { + let root = busbarai_root(); + if !root.join("Cargo.toml").exists() { + if std::env::var_os("CI").is_some() { + panic!( + "stage_fanout_e2e: sibling busbarAI checkout not found at {} under CI; refusing to \ + silently skip the only coverage of the 1.5.3 per-stage notify fan-out", + root.display() + ); + } + eprintln!( + "skip: sibling busbarAI checkout not found at {} (run under the plugin-ci layout)", + root.display() + ); + return None; + } + let bin = root.join("target").join("debug").join("busbar"); + if !bin.exists() { + eprintln!( + "stage_fanout_e2e: building the busbar binary from the sibling checkout (first run \ + only)..." + ); + let status = Command::new("cargo") + .args(["build", "--bin", "busbar"]) + .current_dir(&root) + .status() + .expect("run `cargo build --bin busbar` in the sibling checkout"); + if !status.success() || !bin.exists() { + if std::env::var_os("CI").is_some() { + panic!( + "stage_fanout_e2e: failed to build the busbar binary from the sibling checkout \ + under CI" + ); + } + eprintln!("skip: failed to build the busbar binary locally"); + return None; + } + } + Some(bin) +} + +/// Locate the built `webrequest` cdylib. Checks BOTH the uplifted `/` copy and +/// the raw `/deps/` compiler output, newest wins - a bare `cargo test` does not +/// uplift the top-level copy, so checking only that path finds nothing or something stale and this +/// test silently no-ops. See `tests/e2e.rs`'s `plugin_path()` for the full story. +fn webrequest_cdylib() -> Option { + let candidate = (|| { + let exe = std::env::current_exe().ok()?; + let profile_dir = exe.parent()?.parent()?; + let name = busbar_plugin_loader::plugin_library_filename("busbar_webrequest_hook_plugin"); + let uplifted = profile_dir.join(&name); + let raw = profile_dir.join("deps").join(&name); + [uplifted, raw] + .into_iter() + .filter_map(|p| { + std::fs::metadata(&p) + .and_then(|m| m.modified()) + .ok() + .map(|mtime| (p, mtime)) + }) + .max_by_key(|(_, mtime)| *mtime) + .map(|(p, _)| p) + })(); + if candidate.is_none() && std::env::var_os("CI").is_some() { + panic!( + "stage_fanout_e2e: the webrequest-hook plugin cdylib is not built under CI: \ + `cargo test` must build busbar_webrequest_hook_plugin (checked both the uplifted \ + target dir and target/deps)." + ); + } + candidate +} + +/// Grab an ephemeral free TCP port by binding to port 0 and reading it back, then dropping the +/// listener. Small TOCTOU window, acceptable for a test: a real collision fails loudly. +fn free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .expect("bind ephemeral port") + .local_addr() + .expect("local_addr") + .port() +} + +/// A local mock HTTP server standing in for the operator's webhook target. Captures every POSTed +/// body (parsed as JSON) and always replies `{}`. +async fn mock_webhook() -> (String, Arc>>) { + let captured: Arc>> = Arc::new(Mutex::new(Vec::new())); + let sink = captured.clone(); + let app = Router::new().route( + "/", + post(move |body: axum::body::Bytes| { + let sink = sink.clone(); + async move { + if let Ok(v) = serde_json::from_slice::(&body) { + sink.lock().unwrap().push(v); + } + ( + axum::http::StatusCode::OK, + [(axum::http::header::CONTENT_TYPE, "application/json")], + "{}", + ) + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (format!("http://{addr}/"), captured) +} + +/// A mock upstream "model" server: replies to `POST /v1/messages` with a minimal valid Anthropic +/// Messages response, enough for busbar's ingress/egress translation to produce a real 200. +async fn mock_upstream() -> String { + let app = Router::new().route( + "/v1/messages", + post(|| async { + ( + axum::http::StatusCode::OK, + [(axum::http::header::CONTENT_TYPE, "application/json")], + r#"{"id":"msg_1","type":"message","role":"assistant","content":[{"type":"text","text":"hi"}],"stop_reason":"end_turn","usage":{"input_tokens":11,"output_tokens":7}}"#, + ) + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + format!("http://{addr}") +} + +/// Build a GENUINELY signed plugin tarball around `lib_bytes`, the exact way the release pipeline +/// does, so the REAL signature-verification path runs rather than the `allow_unsigned` escape. +fn build_signed_tarball(lib_bytes: &[u8]) -> (Vec, String) { + let seed = [0x42u8; 32]; + let key = SigningKey::from_bytes(&seed); + let manifest = Manifest { + name: "busbar-webrequest-hook-plugin".to_string(), + alias: "webrequest".to_string(), + kind: "hook".to_string(), + version: "1.5.0".to_string(), + publisher: "acme-e2e".to_string(), + abi_version: *busbar_plugin_loader::supported_abi("hook") + .iter() + .max() + .unwrap(), + sha256: String::new(), + signature: String::new(), + description: "stage_fanout_e2e signed test tarball".to_string(), + homepage: String::new(), + license: "Apache-2.0".to_string(), + needs: HookNeeds { + prompt: NeedLevel::Ro, + user: NeedLevel::No, + }, + settings_schema: None, + schema_derived: false, + host: None, + }; + let signed = busbar_plugin_sign::sign(&key, manifest, lib_bytes); + let tarball = busbar_plugin_loader::tarball::package(&signed, "lib.so", lib_bytes) + .expect("package signed tarball"); + let pubkey_hex = hex_encode(&key.verifying_key().to_bytes()); + (tarball, pubkey_hex) +} + +/// Lowercase-hex encode - avoids pulling in a `hex` crate for one call site. +fn hex_encode(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +/// A child-process guard: always kill + reap `busbar` on drop (including on test panic), so a +/// failing assertion never leaks a live busbar process holding ports open. +struct BusbarProcess { + child: Child, +} + +impl Drop for BusbarProcess { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +/// Poll `GET /healthz` on the admin listener until it answers, or panic after a generous timeout. +async fn wait_for_healthz(admin_addr: &str) { + let client = reqwest::Client::new(); + let deadline = std::time::Instant::now() + Duration::from_secs(30); + loop { + if let Ok(resp) = client + .get(format!("http://{admin_addr}/healthz")) + .timeout(Duration::from_secs(2)) + .send() + .await + { + if resp.status().is_success() { + return; + } + } + if std::time::Instant::now() > deadline { + panic!("busbar did not become healthy within 30s"); + } + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +/// Poll the captured-webhook sink until the observed CALL COUNT has stopped moving, then return the +/// settled batch. Same helper (and same reasoning) as `full_stack_e2e.rs`'s. +/// +/// This test needs the settle even more than that one does: it asserts a SET of stages, so reading +/// the sink the moment it turns non-empty would routinely see the request and candidate envelopes +/// and miss routing and response, failing for a reason that has nothing to do with the contract. +/// A late arrival resets the streak, so an unexpected EXTRA envelope is surfaced to the assertions +/// rather than raced past. +async fn settle_captured( + captured: &Arc>>, + timeout: Duration, +) -> Vec { + /// Gap between polls. + const POLL: Duration = Duration::from_millis(100); + /// Consecutive equal, non-zero counts required before the batch is considered settled. + const STABLE_POLLS: u32 = 5; + + let deadline = std::time::Instant::now() + timeout; + let mut last_len = 0usize; + let mut stable = 0u32; + loop { + tokio::time::sleep(POLL).await; + let len = captured.lock().unwrap().len(); + if len > 0 && len == last_len { + stable += 1; + if stable >= STABLE_POLLS { + return captured.lock().unwrap().clone(); + } + } else { + stable = 0; + last_len = len; + } + if std::time::Instant::now() > deadline { + if last_len > 0 { + return captured.lock().unwrap().clone(); + } + panic!( + "the mock webhook target never received a call from the real, admin-API-installed \ + webrequest hook within {timeout:?}: the real install -> load -> invoke -> webhook \ + chain did not complete" + ); + } + } +} + +/// The 1.5.3 fan-out proof. `#[ignore]`-free and part of the normal `cargo test` run. +#[tokio::test(flavor = "multi_thread")] +async fn unscoped_hook_observes_every_core_stage() { + let Some(busbar_bin) = busbar_bin() else { + return; // logged above; CI already hard-panics instead of reaching here + }; + let Some(cdylib_path) = webrequest_cdylib() else { + return; // logged above; CI already hard-panics instead of reaching here + }; + + // --- 1. A genuinely signed tarball around the REAL built cdylib ----------------------------- + let lib_bytes = std::fs::read(&cdylib_path).expect("read webrequest cdylib"); + let (tarball, publisher_pubkey_hex) = build_signed_tarball(&lib_bytes); + + // --- 2. Mock targets: the webhook this hook forwards to, and the upstream model ------------- + let (webhook_url, captured) = mock_webhook().await; + let upstream_base = mock_upstream().await; + + // --- 3. Generate a real config.yaml/providers.yaml and boot a real busbar process ----------- + let workdir = std::env::temp_dir().join(format!( + "busbar-webrequest-stage-fanout-e2e-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&workdir); + let plugins_dir = workdir.join("plugins"); + std::fs::create_dir_all(&plugins_dir).expect("create plugins dir"); + + let data_port = free_port(); + let admin_port = free_port(); + let admin_addr = format!("127.0.0.1:{admin_port}"); + let admin_token = format!("stage-fanout-admin-token-{}", std::process::id()); + + let providers_yaml = format!( + r#" +mockup: + protocol: anthropic + base_url: "{upstream_base}" + error_map: {{}} +"# + ); + std::fs::write(workdir.join("providers.yaml"), providers_yaml).expect("write providers.yaml"); + + // `auth.chain: []` (open relay) on purpose, and `admin-tokens` DEFINED once under + // `identity-providers:` then REFERENCED by bare name from `auth.admin_auth` - busbar 1.5.3 + // retired the inline-module form and refuses to boot a config still carrying it. + let config_yaml = format!( + r#" +listen: "127.0.0.1:{data_port}" +admin_listen: "127.0.0.1:{admin_port}" +identity-providers: + admin-tokens: {{ module: admin-tokens, token: {{ env: BUSBAR_E2E_ADMIN_TOKEN }} }} +auth: + chain: [] + signing_key: {{ env: BUSBAR_SIGNING_KEY }} + admin_auth: [admin-tokens] +plugins: + enabled: true + dir: "{plugins_dir}" + trust: + publishers: + - name: acme-e2e + public_key: "{publisher_pubkey_hex}" +providers: + mockup: + api_key: {{ env: BUSBAR_E2E_UPSTREAM_KEY }} +models: + m: + provider: mockup + max_concurrent: 5 + max_requests: -1 +"#, + plugins_dir = plugins_dir.display(), + ); + std::fs::write(workdir.join("config.yaml"), config_yaml).expect("write config.yaml"); + + let mut cmd = Command::new(&busbar_bin); + cmd.env("BUSBAR_CONFIG", workdir.join("config.yaml")) + .env("BUSBAR_PROVIDERS", workdir.join("providers.yaml")) + .env("BUSBAR_E2E_ADMIN_TOKEN", &admin_token) + .env("BUSBAR_E2E_UPSTREAM_KEY", "sk-e2e-fake-upstream-key") + .env("BUSBAR_SIGNING_KEY", TEST_SIGNING_KEY) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let mut child = cmd.spawn().expect("spawn busbar"); + // Mirror the real busbar process's stdout/stderr onto this test's stderr, prefixed, so + // `cargo test`'s capture shows them whenever this test fails. + for (label, pipe) in [ + ( + "busbar/out", + child + .stdout + .take() + .map(|s| Box::new(s) as Box), + ), + ( + "busbar/err", + child + .stderr + .take() + .map(|s| Box::new(s) as Box), + ), + ] { + if let Some(pipe) = pipe { + std::thread::spawn(move || { + use std::io::BufRead; + for line in std::io::BufReader::new(pipe).lines().map_while(Result::ok) { + eprintln!("[{label}] {line}"); + } + }); + } + } + let _busbar = BusbarProcess { child }; + + wait_for_healthz(&admin_addr).await; + + let client = reqwest::Client::new(); + let admin = format!("http://{admin_addr}/api/v1/admin"); + + // --- 4. POST the signed tarball to the REAL admin API --------------------------------------- + let install_body = serde_json::json!({ + "file": "webrequest-stage-fanout.tar.gz", + "tarball_b64": base64::Engine::encode(&base64::engine::general_purpose::STANDARD, &tarball), + }); + let resp = client + .post(format!("{admin}/plugins")) + .header("x-admin-token", &admin_token) + .json(&install_body) + .send() + .await + .expect("POST /plugins"); + let status = resp.status(); + let body: serde_json::Value = resp.json().await.expect("install response json"); + assert_eq!( + status, 201, + "plugin install must return 201 Created, got {status}: {body}" + ); + assert_eq!( + body["trust"], "trusted", + "a validly signed, allowlisted-publisher plugin must install as trusted: {body}" + ); + + // --- 5. Reload the plugin registry so the freshly installed tarball is actually loaded ------- + let resp = client + .post(format!("{admin}/plugins/reload")) + .header("x-admin-token", &admin_token) + .send() + .await + .expect("POST /plugins/reload"); + assert!( + resp.status().is_success(), + "plugin reload must succeed: {}", + resp.status() + ); + + // --- 6. Register an UNSCOPED global tap: NO `at:`, NO `phase:` ------------------------------ + // This is the whole point of the file. The omission is deliberate, not an oversight, and it is + // the configuration an operator writes when they have not thought about stages at all. Under + // busbar 1.5.3 that means the hook fires at every core stage rather than once per request (see + // the CHANGELOG quote in this file's header). If either key is ever added here, this test stops + // testing anything and the assertions below will say so rather than pass vacuously. + let hook_body = serde_json::json!({ + "name": "webrequest-stage-fanout-tap", + "config": { + "kind": "tap", + "plugin": "webrequest", + "global": true, + "prompt": "ro", + "settings": { "url": webhook_url }, + } + }); + let resp = client + .post(format!("{admin}/hooks")) + .header("x-admin-token", &admin_token) + .json(&hook_body) + .send() + .await + .expect("POST /hooks"); + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + assert_eq!( + status, 201, + "hook registration must return 201 Created, got {status}: {body}" + ); + + // --- 7. Drive a REAL data-plane request through the REAL router ------------------------------ + // Registering a hook with a CONTENT grant (`prompt: ro`) flips governance on even though + // `auth.chain` is empty, so a real client request needs a real minted key. + let resp = client + .post(format!("{admin}/keys")) + .header("x-admin-token", &admin_token) + .json(&serde_json::json!({ "name": "stage-fanout-e2e-client" })) + .send() + .await + .expect("POST /keys"); + let status = resp.status(); + let mint: serde_json::Value = resp.json().await.expect("mint key json"); + assert!( + status.is_success(), + "minting a client key must succeed: {status}: {mint}" + ); + let client_key = mint["token"] + .as_str() + .unwrap_or_else(|| panic!("mint response carries no token: {mint}")) + .to_string(); + + let prompt_text = "hello from stage_fanout_e2e - prove the per-stage notify fan-out"; + let chat_body = serde_json::json!({ + "model": "m", + "max_tokens": 16, + "messages": [ { "role": "user", "content": prompt_text } ], + }); + let resp = client + .post(format!("http://127.0.0.1:{data_port}/m/v1/messages")) + .header("x-api-key", &client_key) + .json(&chat_body) + .send() + .await + .expect("POST /m/v1/messages"); + let status = resp.status(); + let body: serde_json::Value = resp.json().await.expect("chat response json"); + assert!( + status.is_success(), + "the real chat request must succeed (proves the request actually reached the mock \ + upstream and came back through busbar): {status}: {body}" + ); + + // --- 8. The fan-out assertions --------------------------------------------------------------- + let received = settle_captured(&captured, Duration::from_secs(15)).await; + + // Every delivery is a `notify`. A tap is never asked to decide or transform, so an envelope + // with any other `op` here would mean the engine routed the wrong message kind to a tap. + for envelope in &received { + assert_eq!( + envelope["op"], "notify", + "every stage delivery to a tap must be a notify: {envelope}" + ); + } + + // Split by the presence of the top-level `stage` key. Its ABSENCE is what marks the + // request-stage envelope: the engine omits `stage` entirely there (and on every gate payload), + // which is what keeps the pre-1.5.3 request-stage wire byte-identical. Keying on presence, not + // on a null or a sentinel `at` value, is exactly what the contract tells sidecar authors to do. + let (staged, unstaged): (Vec<_>, Vec<_>) = received + .iter() + .partition(|e| e.get("stage").is_some_and(|s| !s.is_null())); + + assert_eq!( + unstaged.len(), + 1, + "exactly one stage-less (request-stage) envelope expected for one request: {received:?}" + ); + + let stages: BTreeSet<&str> = staged + .iter() + .filter_map(|e| e["stage"]["at"].as_str()) + .collect(); + let expected: BTreeSet<&str> = ["candidate", "routing", "response"].into_iter().collect(); + assert_eq!( + stages, expected, + "an unscoped hook must observe the candidate, routing and response stages (plus the \ + stage-less request envelope asserted above): {received:?}" + ); + + // One dispatch attempt against a one-member pool that answers 200: no failover, so exactly one + // envelope per stage and four in total. Asserted AFTER the set comparison so a duplicate stage + // is reported as a count mismatch here rather than being swallowed by the set dedup above. + assert_eq!( + received.len(), + 4, + "one request, one successful dispatch attempt: one envelope per core stage: {received:?}" + ); + + // PROMPT CONTENT lands on the request stage and NOWHERE else. Stage taps are shape-only by + // construction in the engine (`system`/`messages`/`user` are sent as absent regardless of the + // hook's grant), so a `prompt: ro` hook does not receive content four times over. A sidecar + // that screens or logs content keys off this envelope. + let request_envelope = unstaged[0]; + assert!( + request_envelope.to_string().contains(prompt_text), + "the request-stage envelope must carry the real prompt content (granted via prompt: ro): \ + {request_envelope}" + ); + for envelope in &staged { + assert!( + !envelope.to_string().contains(prompt_text), + "a stage envelope must be shape-only and must never carry prompt content: {envelope}" + ); + assert!( + envelope["request"].get("messages").is_none(), + "a stage envelope must omit `messages` entirely, not send it empty: {envelope}" + ); + assert!( + envelope["request"].get("user").is_none(), + "a stage envelope must omit `user` entirely: {envelope}" + ); + assert_eq!( + envelope["candidates"], + serde_json::json!([]), + "a stage envelope carries no candidate projection: {envelope}" + ); + } + + // The documented per-stage payload fields. These are what make each stage worth observing at + // all: without them a stage notify is an empty ping. + let by_stage = |at: &str| -> &serde_json::Value { + staged + .iter() + .find(|e| e["stage"]["at"] == at) + .unwrap_or_else(|| panic!("no `{at}` stage envelope in {received:?}")) + }; + + // `candidate`: the surviving candidate count after the decision reconcile. One member pool, + // no gate restricted anything, so one candidate survived. + assert_eq!( + by_stage("candidate")["stage"]["remaining_candidates"], + 1, + "the candidate stage reports the post-reconcile surviving candidate count" + ); + + // `routing`: the failover story for this dispatch attempt. First attempt, so `attempt_number` + // is 1 and `previous_failure` is absent (there is no previous attempt to report). + let routing = by_stage("routing"); + assert_eq!( + routing["stage"]["attempt_number"], 1, + "the first dispatch attempt is attempt_number 1: {routing}" + ); + assert_eq!( + routing["stage"]["model"], "m", + "the routing stage names the DISPATCHED member: {routing}" + ); + assert!( + routing["stage"].get("previous_failure").is_none(), + "there is no previous failure on the first attempt, so the field is absent: {routing}" + ); + + // `response`: the outcome. The upstream answered 200, and no gate rejected, so this is the + // plain `ok` outcome rather than a synthetic `rejected_by_gate` / `rejected_by_auth`. + let response = by_stage("response"); + assert_eq!( + response["stage"]["outcome"], "ok", + "a served 200 is the `ok` outcome: {response}" + ); + assert_eq!( + response["stage"]["status"], 200, + "the response stage carries the real response status: {response}" + ); + + // THE JOIN KEY. Four separate POSTs are only usable if a sidecar can correlate them back to one + // request, and `request.request_id` is the documented handle for that. If it ever varied per + // stage, stage taps would be useless for audit and nothing else in this test would notice. + let ids: BTreeSet = received + .iter() + .filter_map(|e| e["request"]["request_id"].as_u64()) + .collect(); + assert_eq!( + ids.len(), + 1, + "every stage envelope for one request must carry the SAME request_id join key: {received:?}" + ); + + let _ = std::fs::remove_dir_all(&workdir); +}