Add tests and docs for per-connection memory budgets (8.3.6) (#489)

* test(memory-budgets): add comprehensive tests for per-connection memory budgets

Add new async unit tests for apply_memory_pressure covering Abort, Pause, and Continue actions.

Introduce two new BDD suites: budget_cleanup and budget_transitions, with six scenarios validating cleanup semantics, budget reclamation, soft-limit pacing escalation, recovery after assembly completion, and tightest dimension enforcement.

Create fixtures, steps, and scenario bindings to support the new BDD tests.

Update docs/roadmap.md to mark 8.3.6 as done and add implementation decisions to ADR 0002.

This change significantly improves test coverage for the three-tier per-connection memory budget protection system, enabling better verification of budget enforcement, back-pressure, cleanup, and pressure-level transitions without modifying production code.

Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>

* test(budget): harden connection abort assertions for budget-related errors

Refined the `assert_connection_aborted` method in both `budget_transitions.rs` and `memory_budget_hard_cap.rs` tests to specifically verify that the error kind is `InvalidData`, which is the expected error kind produced by budget enforcement, instead of accepting any server error. Updated documentation accordingly to reflect this stricter error validation.

Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>

* refactor(tests): simplify error handling and adjust payload checks in budget tests

Refactor budget transition and memory budget hard cap tests:
- Simplified the handling of send_result by flattening nested matches into combinators
- Updated assert_connection_aborted to drain payloads after server task finishes to catch payloads during teardown
- Documented that send errors in send_first_frames_for_range are intentionally ignored due to expected abort behavior
These changes improve code clarity and correctness in test fixtures.

Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>

* refactor(tests/fixtures): refactor MemoryBudgetHardCapWorld server join handling

- Added SERVER_JOIN_TIMEOUT constant for server join timeout
- Extracted join_server() method to unify server task joining and error handling
- Replaced inline server join handling in assert_connection_aborted with join_server()
- Introduced helper parse function for cleaner HardCapConfig parsing
- Moved with_runtime() impl block for better code organization
- Removed redundant error handling branch for join errors

This improves test reliability by timing out server joins and simplifies error handling.

Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>

---------

Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>

Author: leynos

docs/adr-002-streaming-requests-and-shared-message-assembly.md

@@ -373,6 +373,28 @@ Precedence is:
 - Explicit budgets always take precedence over derived defaults (ADR-002
   precedence rule: explicit > global caps > derived defaults).
 
+#### Implementation decisions (2026-03-01)
+
+- Roadmap item `8.3.6` adds comprehensive test coverage for the three-tier
+  per-connection memory budget protection model (items `8.3.1` through `8.3.5`).
+- Three `#[tokio::test]` unit tests validate the `apply_memory_pressure()`
+  async function for each `MemoryPressureAction` variant (`Abort` returns
+  `InvalidData`, `Pause` invokes the purge closure, `Continue` is a no-op).
+- A `budget_cleanup` BDD suite (3 scenarios) validates end-to-end cleanup
+  semantics: timeout purge reclaims budget headroom for subsequent frames,
+  completed assemblies reclaim budget for subsequent frames, and connection
+  close frees all partial assemblies via RAII drop of `MessageAssemblyState`.
+- A `budget_transitions` BDD suite (3 scenarios) validates pressure
+  transitions: soft-limit pacing escalates to connection termination after
+  repeated per-frame rejections, assembly completion recovers the connection
+  from soft-limit pressure, and the tightest aggregate dimension
+  (`min(bytes_per_connection, bytes_in_flight)`) controls enforcement when the
+  two dimensions differ.
+- Cleanup relies entirely on Rust's RAII semantics. When `process_stream`
+  returns, its local `message_assembly: Option<MessageAssemblyState>` drops,
+  which drops the internal `HashMap<MessageKey, PartialAssembly>`, freeing all
+  partial body and metadata buffers. No custom `Drop` implementation is needed.
+
 #### Budget enforcement
 
 - Budgets MUST cover: bytes buffered per message, bytes buffered per

docs/execplans/8-3-6-tests-for-per-connection-memory-budgets.md

@@ -299,7 +299,7 @@ and standardized per-connection memory budgets.
   `InvalidData`) behaviour.
 - [x] 8.3.5. Define derived defaults based on `buffer_capacity` when budgets
   are not set explicitly.
-- [ ] 8.3.6. Write tests for budget enforcement, back-pressure, and cleanup
+- [x] 8.3.6. Write tests for budget enforcement, back-pressure, and cleanup
   semantics.
 
 ### 8.4. Transport helper

docs/roadmap.md

@@ -336,3 +336,49 @@ fn resolve_derived_budgets_change_with_frame_budget() -> TestResult {
     }
     Ok(())
 }
+
+// ---------- apply_memory_pressure async tests ----------
+
+#[tokio::test]
+async fn apply_memory_pressure_abort_returns_invalid_data() {
+    use super::backpressure::{MemoryPressureAction, apply_memory_pressure};
+
+    let result = apply_memory_pressure(MemoryPressureAction::Abort, || {}).await;
+    let err = result.expect_err("Abort should return an error");
+    assert_eq!(
+        err.kind(),
+        io::ErrorKind::InvalidData,
+        "expected InvalidData error kind, got {:?}: {err}",
+        err.kind()
+    );
+}
+
+#[tokio::test]
+async fn apply_memory_pressure_pause_invokes_purge_closure() {
+    use super::backpressure::{MemoryPressureAction, apply_memory_pressure};
+
+    tokio::time::pause();
+
+    let mut purge_called = false;
+    let result = apply_memory_pressure(
+        MemoryPressureAction::Pause(Duration::from_millis(1)),
+        || purge_called = true,
+    )
+    .await;
+    result.expect("Pause should return Ok");
+    assert!(
+        purge_called,
+        "expected purge closure to be called during Pause action"
+    );
+}
+
+#[tokio::test]
+async fn apply_memory_pressure_continue_is_noop() {
+    use super::backpressure::{MemoryPressureAction, apply_memory_pressure};
+
+    let result = apply_memory_pressure(MemoryPressureAction::Continue, || {
+        panic!("purge closure should not be called for Continue action");
+    })
+    .await;
+    result.expect("Continue should return Ok");
+}

src/app/frame_handling/backpressure_tests.rs

@@ -0,0 +1,31 @@
+@budget_cleanup
+Feature: Budget cleanup and reclamation semantics
+  Memory budgets are reclaimed when partial assemblies time out or
+  complete, allowing new frames to arrive within the freed headroom.
+  When a connection closes, all partial assemblies are freed via RAII.
+
+  Scenario: Timeout purge reclaims budget for subsequent frames
+    Given a cleanup app configured as 200/2048/10/10
+    When a cleanup first frame for key 1 with body "aaaaaaaa" arrives
+    And cleanup virtual time advances by 201 milliseconds
+    And a cleanup first frame for key 2 with body "bbbbbbbb" arrives
+    And a cleanup final continuation for key 2 sequence 1 with body "cc" arrives
+    Then cleanup payload "bbbbbbbbcc" is eventually received
+    And no cleanup connection error is recorded
+
+  Scenario: Completed assembly reclaims budget for subsequent frames
+    Given a cleanup app configured as 200/2048/10/10
+    When a cleanup first frame for key 3 with body "cccccccc" arrives
+    And a cleanup final continuation for key 3 sequence 1 with body "d" arrives
+    Then cleanup payload "ccccccccd" is eventually received
+    When a cleanup first frame for key 4 with body "eeeeeeee" arrives
+    And a cleanup final continuation for key 4 sequence 1 with body "f" arrives
+    Then cleanup payload "eeeeeeeef" is eventually received
+    And no cleanup connection error is recorded
+
+  Scenario: Connection close frees all partial assemblies
+    Given a cleanup app configured as 200/2048/100/100
+    When a cleanup first frame for key 5 with body "aaa" arrives
+    And a cleanup first frame for key 6 with body "bbb" arrives
+    And the cleanup client disconnects
+    Then no cleanup connection error is recorded

tests/features/budget_cleanup.feature

@@ -0,0 +1,26 @@
+@budget_transitions
+Feature: Budget pressure transitions and dimension interactions
+  Memory budget protection tiers interact correctly under changing
+  pressure levels: soft-limit pacing at 80 per cent, hard-cap abort
+  at 100 per cent. The tightest aggregate dimension controls
+  enforcement when per-connection and in-flight limits differ.
+
+  Scenario: Soft pressure escalates to connection termination
+    Given a transition app configured as 200/2048/10/10
+    When transition first frames for keys 1 to 15 each with body "aa" arrive
+    Then the transition connection terminates with an error
+
+  Scenario: Recovery from soft limit after assembly completion
+    Given a transition app configured as 200/2048/10/10
+    When a transition first frame for key 20 with body "aaaaaaaa" arrives
+    And a transition final continuation for key 20 sequence 1 with body "b" arrives
+    Then transition payload "aaaaaaaab" is eventually received
+    When a transition first frame for key 21 with body "cc" arrives
+    And a transition final continuation for key 21 sequence 1 with body "d" arrives
+    Then transition payload "ccd" is eventually received
+    And no transition connection error is recorded
+
+  Scenario: Tightest aggregate dimension controls enforcement
+    Given a transition app configured as 200/2048/20/10
+    When transition first frames for keys 30 to 44 each with body "aa" arrive
+    Then the transition connection terminates with an error