# Test-labeler Implementation Plan — Phase 5

**Goal:** Add the subscription stage — a two-connection strategy against `com.atproto.label.subscribeLabels` with a configurable per-connection time budget. First connection does backfill with `cursor=0`; a second connection runs the live-tail check if the backfill did not also exercise the live tail.

**Architecture:** New module `src/commands/test/labeler/subscription.rs`. Contains the two-CBOR-block frame parser (header + payload), the `BackfillOutcome` and `LiveTailOutcome` enums, the `SubscriptionFacts`, and the stage `run` function. The stage is written against a narrow `WebSocketClient` trait so unit tests replay recorded frame sequences from a fake. The real implementation uses `tokio-tungstenite` with rustls native-roots. `ciborium` decodes each of the two CBOR blocks; the stage validates header `op`/`t` fields and routes `#labels` and `#info` message types.

**Tech Stack:** adds `tokio-tungstenite` 0.29 (`rustls-tls-native-roots`), `ciborium` 0.2 (strict-decoding mode), `futures-util` (for `StreamExt`), `bytes`. No new dev-dependencies.

**Scope:** Phase 5 of 6.

**Codebase verified:** 2026-04-13 (post Phase 4). `src/commands/test/labeler/subscription.rs` does **not** yet exist. Pipeline driver calls Phase 4 HTTP stage and still emits `Skipped("not yet implemented (phase 5)")` for subscription.

---

## Acceptance Criteria Coverage

### test-labeler.AC4: Subscription-layer checks (`subscribeLabels`)
- **test-labeler.AC4.1 Success (backfill completes within budget):** A labeler whose backfill flushes within `--subscribe-timeout` followed by a ≥500ms idle gap produces a `Pass` backfill check and an implicit `Pass` live-tail check ("live tail observed after backfill completed").
- **test-labeler.AC4.2 Success (backfill exceeds budget):** A labeler whose backfill is still producing frames at the end of the budget produces an `Advisory` backfill check and triggers a second connection for the live-tail check. A clean live-tail connection (any frames decoded, or connection held open with no decode errors) produces a `Pass` live-tail check.
- **test-labeler.AC4.3 Success (empty labeler):** A labeler producing no frames at all during the budget produces an `Advisory` backfill check ("labeler has no published labels") and a `Skipped` live-tail check with the same reason.
- **test-labeler.AC4.4 Failure:** A frame that fails the two-CBOR-block decode, or a `#labels` payload that fails lexicon decoding, produces a `SpecViolation` `CheckResult` independent of the backfill/live-tail pass/fail dimension. The diagnostic carries the offending frame bytes as `NamedSource`.
- **test-labeler.AC4.5 Failure:** A WebSocket handshake that succeeds but whose first frame has `op: -1` (error) with a malformed `#info` payload produces a `SpecViolation` on decode.
- **test-labeler.AC4.6 NetworkError:** An unreachable WebSocket endpoint or a handshake TLS failure produces a `NetworkError` rather than a subscription `Fail`.
- **test-labeler.AC4.7 Edge:** Passing `--subscribe-timeout 0` (or another invalid duration) produces a clap parse error; values below a reasonable floor (e.g., 1s) are rejected with a helpful message.

AC4.7 is structurally in place since Phase 1 (the `parse_subscribe_timeout` value parser rejects sub-1s). Phase 5 adds an **integration test** asserting the clap rejection still fires and stderr mentions "at least 1 second".

---

## Codebase state at start of phase

Phase 1 → Phase 4 complete. `subscription.rs` does not exist. The Phase 1 `LabelerCmd::subscribe_timeout` flag with `parse_subscribe_timeout` value parser and the 1-second floor remain in `src/commands/test/labeler.rs`; Phase 5 consumes the already-parsed `Duration` through `LabelerOptions::subscribe_timeout`.

---

## External dependency notes

- **`tokio-tungstenite` 0.29.0**:
  - `tokio_tungstenite::connect_async(request)` returns `Result<(WebSocketStream<MaybeTlsStream<TcpStream>>, Response), WsError>`. Build the request with `tokio_tungstenite::tungstenite::client::IntoClientRequest::into_client_request(&url)` so custom headers are possible even though Phase 5 does not need any.
  - `WebSocketStream` implements `Stream<Item = Result<Message, WsError>>` via `futures-util::StreamExt`. Each `Message::Binary(Bytes)` is a full WebSocket binary frame; text frames are ignored with a `SpecViolation` diagnostic.
  - Features: enable `rustls-tls-native-roots` (default feature set for 0.29 includes the non-TLS transports only). Do **not** enable the default `connect` feature with native-tls.
  - Build with `url::Url::parse(...)?` then mutate the scheme from `https` to `wss`. (`com.atproto.label.subscribeLabels` lives under `wss://<endpoint>/xrpc/com.atproto.label.subscribeLabels[?cursor=0]`.)
- **`ciborium` 0.2.2**:
  - `ciborium::de::from_reader::<Value, _>(&mut &bytes[..])` decodes one top-level CBOR item and returns the remainder of the slice (the `Read` impl for `&[u8]` advances a cursor). To decode **two sequential** blocks from a single WebSocket frame, wrap the frame bytes in a `&mut &[u8]` cursor and call `from_reader` twice; the cursor advances naturally.
  - We also use `ciborium::Value` for the header and decode the payload to a concrete lexicon type via `ciborium::de::from_reader::<PayloadStruct, _>` for the `#labels` case. Two possible approaches:
    1. Decode the header into `HeaderData { op: i64, t: Option<String> }`, then pick a decoder for the payload based on `header.t`.
    2. Decode header as `ciborium::Value`, match on the `t` map entry manually, then decode the payload.
  - Prefer approach 1 with a typed `HeaderData` struct; it is the pattern atrium uses.
- **`futures-util` 0.3** — needed for `StreamExt::next()` on the WebSocket stream. Add as a dependency; nothing else in the crate uses it yet.
- **`bytes` 1.x** — `tokio-tungstenite` re-exports this. The raw frame payload arrives as `bytes::Bytes`; convert with `.into()` to `Vec<u8>`/`Arc<[u8]>` for diagnostics.
- **atproto subscribeLabels frame format**:
  - Each WebSocket binary frame is **two concatenated CBOR items**: a header map and a payload map.
  - Header map: `{ "op": Int, "t": Option<Text> }`.
    - `op == 1` → message frame; `t` is a `#`-prefixed lexicon type NSID fragment like `"#labels"` or `"#info"`.
    - `op == -1` → error frame; payload has shape `{ "error": Text, "message": Option<Text> }`. AC4.5 specifically requires us to decode such frames and report a malformed one as `SpecViolation`.
  - Payload for `#labels`: `{ "seq": Int, "labels": Array<Label> }` where each label has the canonical atproto label shape. For Phase 5 decoding, we reuse `atrium_api::com::atproto::label::defs::Label` via ciborium/serde — `Label` already derives `serde::Deserialize`, so `ciborium::de::from_reader::<SubscribeLabelsPayload, _>` works out of the box.
  - Payload for `#info`: `{ "name": Text, "message": Option<Text> }`.

---

<!-- START_TASK_1 -->
### Task 1: Phase 5 dependencies

**Files:**
- Modify: `Cargo.toml` `[dependencies]`

**Implementation:**

Append to `[dependencies]`:

```toml
bytes = "1.10"
ciborium = "0.2"
futures-util = { version = "0.3", default-features = false, features = ["std"] }
tokio-tungstenite = { version = "0.29", default-features = false, features = ["connect", "rustls-tls-native-roots"] }
```

**Verification:**

Run: `cargo check`
Expected: new crates resolve.

**Commit:**
```bash
git add Cargo.toml Cargo.lock
git commit -m "Add Phase 5 dependencies (tokio-tungstenite, ciborium)"
```
<!-- END_TASK_1 -->

<!-- START_SUBCOMPONENT_A (tasks 2-4) -->
<!-- START_TASK_2 -->
### Task 2: Frame parser and `SubscriptionFacts` types

**Files:**
- Create: `src/commands/test/labeler/subscription.rs` (types + parser only)
- Modify: `src/commands/test/labeler.rs` add `pub mod subscription;`

**Implementation:**

Types:

- `pub struct FrameHeader { op: i64, t: Option<String> }` with `#[serde(rename_all = "camelCase")]` (actually the raw atproto keys are lowercase already so no rename needed).
- `pub enum DecodedFrame { Labels(SubscribeLabelsPayload), Info(SubscribeInfoPayload), Error(SubscribeErrorPayload) }`.
- `pub struct SubscribeLabelsPayload { seq: i64, labels: Vec<atrium_api::com::atproto::label::defs::Label> }`.
- `pub struct SubscribeInfoPayload { name: String, message: Option<String> }`.
- `pub struct SubscribeErrorPayload { error: String, message: Option<String> }`.
- `pub enum FrameDecodeError { HeaderDecode { raw: Arc<[u8]>, cause: String }, PayloadDecode { header: FrameHeader, raw: Arc<[u8]>, cause: String }, UnknownMessageType { t: String, raw: Arc<[u8]> }, TextFrameRejected(Arc<[u8]>) }`.
- `pub fn decode_frame(bytes: &[u8]) -> Result<DecodedFrame, FrameDecodeError>`:
  1. Wrap the bytes in a `&mut &[u8]` cursor.
  2. `ciborium::de::from_reader::<FrameHeader, _>(&mut cursor)` → header, or `HeaderDecode` error on failure.
  3. Based on `header.op` and `header.t`:
     - `op == 1, t == Some("#labels")` → `ciborium::de::from_reader::<SubscribeLabelsPayload, _>(&mut cursor)` → `DecodedFrame::Labels` or `PayloadDecode` error.
     - `op == 1, t == Some("#info")` → similar for `SubscribeInfoPayload`.
     - `op == -1` → `ciborium::de::from_reader::<SubscribeErrorPayload, _>(&mut cursor)` → `DecodedFrame::Error` or `PayloadDecode` error. AC4.5 hits this path.
     - Any other combination → `UnknownMessageType`.

- `pub struct SubscriptionFacts { pub backfill_outcome: BackfillOutcome, pub live_tail_outcome: Option<LiveTailOutcome>, pub decode_errors: Vec<FrameDecodeError> }`.
- `pub enum BackfillOutcome { CompletedWithIdleGap { frames_observed: usize, idle_gap_ms: u64 }, ExceededBudget { frames_observed: usize }, NoFramesWithinBudget }`.
- `pub enum LiveTailOutcome { FromBackfill, CleanHold { frames_observed: usize }, SkippedEmpty }`.

**Unit tests (in-file `#[cfg(test)]`):**
- `decode_labels_frame_valid` — construct a valid two-block byte buffer by serializing `FrameHeader { op: 1, t: Some("#labels") }` and `SubscribeLabelsPayload { seq: 0, labels: vec![] }` with `ciborium::ser::into_writer`, concatenating the two, then passing to `decode_frame`. Expect `DecodedFrame::Labels` with `seq == 0` and empty labels.
- `decode_info_frame_valid` — similar for `#info`.
- `decode_error_frame_valid` — op=-1, valid error payload. Expect `DecodedFrame::Error`.
- `decode_frame_header_decode_failure` — feed 3 random bytes. Expect `HeaderDecode`, raw bytes preserved.
- `decode_frame_payload_decode_failure` — valid header for `#labels` followed by `0xff` garbage. Expect `PayloadDecode { header, raw, .. }`.
- `decode_frame_unknown_message_type` — valid header with `t = Some("#futureType")`. Expect `UnknownMessageType`.
- `decode_frame_error_payload_malformed` — op=-1 but payload bytes do not decode as `SubscribeErrorPayload`. Expect `PayloadDecode`. This is the programmatic shape of AC4.5.

**Verification:**

Run: `cargo test`
Expected: all new unit tests pass.

**Commit:**
```bash
git add src/commands/test/labeler/subscription.rs src/commands/test/labeler.rs
git commit -m "Add subscribeLabels frame parser and types"
```
<!-- END_TASK_2 -->

<!-- START_TASK_3 -->
### Task 3: `WebSocketClient` trait, real impl, and stage `run`

**Files:**
- Modify: `src/commands/test/labeler/subscription.rs` — add the IO trait, the real impl, and the stage body.
- Modify: `src/commands/test/labeler/pipeline.rs` — replace the subscription stage stub with a real call.

**Implementation:**

```text
#[async_trait]
pub trait WebSocketClient: Send + Sync {
    async fn connect(
        &self,
        url: &Url,
    ) -> Result<Box<dyn FrameStream>, SubscriptionStageError>;
}

#[async_trait]
pub trait FrameStream: Send {
    async fn next_frame(&mut self) -> Option<Result<Vec<u8>, SubscriptionStageError>>;
    async fn close(&mut self);
}
```

- `RealWebSocketClient` — uses `tokio_tungstenite::connect_async(request)` internally. `next_frame` filters text frames (→ error), ping/pong frames (discard), and close frames (→ `None`). Binary frames yield `Vec<u8>`.

Stage `run`:

```text
pub async fn run(
    labeler_endpoint: &Url,
    ws: &dyn WebSocketClient,
    budget_per_connection: Duration,
) -> SubscriptionStageOutput;
```

Logic:

1. Build `backfill_url = labeler_endpoint + "/xrpc/com.atproto.label.subscribeLabels?cursor=0"` with scheme `wss`.
2. **Backfill connection.** Start a `tokio::time::timeout(budget_per_connection, ...)` around the frame-draining loop:
   - `first_frame_at: Option<Instant> = None`
   - `last_frame_at: Option<Instant> = None`
   - `frames_observed: usize = 0`
   - Loop: `tokio::select!` on `next_frame` and a `tokio::time::sleep_until(last_frame_at + 500ms)` once any frame has arrived.
     - Frame arrived → decode via `decode_frame`. On `Ok(DecodedFrame::Labels | Info | Error)`, update counters. On `Err(..)`, push to `decode_errors` (AC4.4) and continue reading; decode errors do not abort the loop.
     - Idle gap timer fired → set `backfill_outcome = CompletedWithIdleGap { frames_observed, idle_gap_ms: 500 }`, mark live tail as `FromBackfill`, break.
     - Top-level timeout fired → if `frames_observed > 0`, `backfill_outcome = ExceededBudget { frames_observed }`; else `NoFramesWithinBudget`. Break.
3. **Live-tail connection** (only if `live_tail_outcome.is_none()`):
   - For `ExceededBudget`, issue a second connection to the same URL _without_ `cursor=0` (live-only). Run for `budget_per_connection`, counting frames. Any decode errors accumulate into `decode_errors`. At end-of-budget → `CleanHold { frames_observed }`. AC4.2.
   - For `NoFramesWithinBudget`, set `live_tail_outcome = SkippedEmpty`. AC4.3. Do not issue a second connection.
4. Emit `CheckResult`s:
   - `subscription::backfill` — `Pass` for `CompletedWithIdleGap`, `Advisory` for `ExceededBudget`, `Advisory` for `NoFramesWithinBudget` (summary differentiates the two).
   - `subscription::live_tail` — `Pass` for `FromBackfill` or `CleanHold`, `Skipped` for `SkippedEmpty` with reason `"labeler has no published labels"`.
   - One additional `SpecViolation` `CheckResult` per unique `FrameDecodeError` variant (dedupe by variant kind so we do not flood the report when thousands of frames fail). AC4.4, AC4.5. Each diagnostic carries the raw frame bytes as `NamedSource` and a `#[label("frame decode failure")]` span covering the first byte.
5. Return the `SubscriptionFacts`.

If the initial `connect_async` fails with a transport/TLS error, emit a single `subscription::endpoint_reachable` `CheckResult` with severity `NetworkError` (AC4.6) and return with no facts.

**Pipeline driver wiring:** Replace the subscription `Skipped` stub with a real call. Route the `subscribe_timeout` from `LabelerOptions::subscribe_timeout` into `run`.

**Verification:**

Run: `cargo build`
Expected: compiles.

Run: `cargo run -- test labeler moderation.bsky.app --subscribe-timeout 3s` (online)
Expected: subscription section populated; backfill and live-tail checks render `[OK]` or `[WARN]` depending on the live labeler's current behaviour.

**Commit:**
```bash
git add src/commands/test/labeler/subscription.rs src/commands/test/labeler/pipeline.rs
git commit -m "Implement subscription stage with two-connection strategy"
```
<!-- END_TASK_3 -->

<!-- START_TASK_4 -->
### Task 4: Snapshot tests for the subscription stage

**Verifies:** AC4.1, AC4.2, AC4.3, AC4.4, AC4.5, AC4.6, AC4.7.

**Files:**
- Create: `tests/fixtures/labeler/subscription/` — one subdirectory per case:
  - `backfill_complete/frames.bin` — pre-encoded CBOR binary frame stream (sequence of two-CBOR-block binary frames) that ends with a 500ms idle gap before the budget runs out. The fixture is generated by a `tests/support/encode_frames.rs` helper at fixture-capture time, **not** at test run time (fixtures are static bytes; the generator is a `#[ignore]`d dev binary). AC4.1.
  - `backfill_exceeds_budget/frames.bin` — continuous frame stream with no idle gap within the budget. AC4.2.
  - `empty_stream/frames.bin` — 0-byte fixture; fake simulates "no frames ever arrive". AC4.3.
  - `malformed_frame/frames.bin` — 2 valid frames followed by 3 random bytes. AC4.4.
  - `error_frame_malformed/frames.bin` — one frame whose header has `op == -1` and whose payload cannot decode as `SubscribeErrorPayload`. AC4.5.
  - `unreachable/` — empty. Fake's `connect` returns `SubscriptionStageError::Transport(_)`. AC4.6.
- Create: `tests/labeler_subscription.rs` — snapshot tests.
- Create: `tests/support/encode_frames.rs` — a tiny `#[bin]` target under `[[bin]]` in `Cargo.toml` is overkill; instead, place the generator as a `#[test] #[ignore]` function in the integration test file. When run with `cargo test --test labeler_subscription -- --ignored gen_fixtures`, it writes all `frames.bin` files from in-memory constructions. Document this in a top comment.

**Test fakes:**

- `FakeWebSocketClient { scripts: Vec<FakeScript> }` where `FakeScript = { frames: Vec<Vec<u8>>, inter_frame_delay: Duration, final_wait: Option<Duration>, transport_error: bool }`. Each call to `.connect` pops the next script. The returned `FakeFrameStream` uses `tokio::time::sleep` (**not** `std::thread::sleep`) for `inter_frame_delay` between frames and `final_wait` after the last frame (used to simulate the 500ms idle gap after backfill completes, or a continuing stream exceeding the budget). Both the fake's sleeps and the stage's `tokio::time::timeout`/`sleep_until` calls must run on the same paused clock.
- **Runtime flavor:** every `#[tokio::test]` function in `labeler_subscription.rs` must be annotated as `#[tokio::test(flavor = "current_thread", start_paused = true)]`. This both matches the production binary's runtime (`current_thread` from Phase 1) and ensures `tokio::time::pause` is active for the whole test, so `tokio::time::advance(...)` drives both the fake's inter-frame sleeps and the stage's budget timer deterministically. Without `start_paused = true`, any `tokio::time::sleep` that fires before the first `advance` will wall-clock block.

**Test list:**

- `backfill_completes_within_budget_passes` — AC4.1. Script: 3 frames, inter-frame 10ms, then a 600ms idle gap. Subscribe timeout 2s. Assert `backfill: [OK]` and `live_tail: [OK]` with "observed after backfill completed" reason.
- `backfill_exceeds_budget_triggers_live_tail` — AC4.2. Script 1: 20 frames, inter-frame 50ms → the 2s budget expires while still producing frames. Script 2: 1 frame then 1s idle, for the live-tail connection. Assert `backfill: [WARN] ExceededBudget`, `live_tail: [OK] CleanHold`.
- `empty_stream_advisories` — AC4.3. Script: no frames, budget runs out. Assert `backfill: [WARN]` with "no published labels" reason and `live_tail: [SKIP]` with the same reason. Assert exit code is 0.
- `malformed_frame_emits_spec_violation` — AC4.4. Script: 1 valid frame, 1 malformed. Assert one `[FAIL] subscription::frame_decode` `CheckResult` with source bytes equal to the malformed fixture.
- `error_frame_malformed_payload_spec_violation` — AC4.5.
- `unreachable_endpoint_network_error` — AC4.6. Assert severity `NetworkError`, summary calls out one network error, exit code 0.
- `subscribe_timeout_below_floor_rejected` — AC4.7. Preferred implementation: unit test on the `parse_subscribe_timeout` helper in `labeler.rs`:
  - Assert `parse_subscribe_timeout("500ms")` returns `Err(_)` with a message containing `"at least 1 second"` (the floor check fires because `humantime` parses `500ms` successfully, then `parse_subscribe_timeout` rejects it as sub-second).
  - Assert `parse_subscribe_timeout("0")` returns `Err(_)` (any error is acceptable — `humantime::parse_duration` rejects bare `0` at parse time with `"time unit needed"`, before the floor check runs).
  - Optional end-to-end complement using `assert_cmd`: `Command::cargo_bin("atproto-devtool").args(["test", "labeler", "--subscribe-timeout", "500ms", "not-a-real-thing"])` → exit code 2, stderr contains `"at least 1 second"`. Deterministic because clap rejects the argument before any network IO runs.

**Verification:**

Run: `cargo insta test --accept` (first run)
Run: `cargo test` (subsequent runs)
Expected: all 7 new tests pass.

**Commit:**
```bash
git add tests/labeler_subscription.rs tests/fixtures/labeler/subscription tests/snapshots
git commit -m "Add subscription stage snapshot tests covering AC4"
```
<!-- END_TASK_4 -->
<!-- END_SUBCOMPONENT_A -->

<!-- START_TASK_5 -->
### Task 5: Final phase verification

**Steps:**
1. `cargo fmt --check` — clean.
2. `cargo clippy -- -D warnings` — clean.
3. `cargo test` — all Phase 3/4/5 tests pass.
4. (Optional, online) `cargo run -- test labeler moderation.bsky.app --subscribe-timeout 5s` — populated subscription section.

**Done when:** steps 1–3 pass.
<!-- END_TASK_5 -->

---

## Phase 5 done-when checklist

- `src/commands/test/labeler/subscription.rs` exists, compiles, and exports the frame parser, the `WebSocketClient` and `FrameStream` traits, the `RealWebSocketClient` impl, the `SubscriptionFacts` struct, and the `run` entry point.
- Pipeline driver calls `subscription::run` with the parsed `subscribe_timeout`.
- All 7 Phase 5 tests pass (6 snapshot + 1 argument parser unit test).
- Running against a real labeler populates the subscription section.
- `cargo clippy -- -D warnings` and `cargo fmt --check` clean.
- Commit history shows the four atomic commits from this phase.
