diff --git a/docs/adr-002-streaming-requests-and-shared-message-assembly.md b/docs/adr-002-streaming-requests-and-shared-message-assembly.md
index 185c818b..4d917181 100644
--- a/docs/adr-002-streaming-requests-and-shared-message-assembly.md
+++ b/docs/adr-002-streaming-requests-and-shared-message-assembly.md
@@ -331,6 +331,25 @@ Precedence is:
   rather than an indefinite read stop, so in-flight assemblies can continue to
   make progress.
 
+#### Implementation decisions (2026-02-26)
+
+- Roadmap item `8.3.4` implements hard-cap connection abort as a
+  defence-in-depth safety net in the inbound read loop. When total buffered
+  bytes strictly exceed the aggregate cap
+  (`min(bytes_per_connection, bytes_in_flight)`), the connection is terminated
+  immediately with `InvalidData`, bypassing the `DeserFailureTracker` counter.
+- The hard-cap check is combined with the soft-limit check into a single
+  `evaluate_memory_pressure()` function in
+  `src/app/frame_handling/backpressure.rs`, returning a `MemoryPressureAction`
+  enum (`Continue`, `Pause`, or `Abort`). This keeps the inbound loop
+  (`src/app/inbound_handler.rs`) within the 400-line file limit.
+- The hard cap uses `>` (strictly exceeds) for the comparison, matching the
+  `check_aggregate_budgets` convention in `budget.rs`. The limit value itself
+  is permitted.
+- Under normal operation, per-frame budget enforcement (`8.3.2`) prevents
+  total buffered bytes from exceeding the limit. The hard cap catches the edge
+  case where this invariant is violated.
+
 #### Budget enforcement
 
 - Budgets MUST cover: bytes buffered per message, bytes buffered per
diff --git a/docs/execplans/8-3-4-hard-cap-behaviour.md b/docs/execplans/8-3-4-hard-cap-behaviour.md
new file mode 100644
index 00000000..d820e96f
--- /dev/null
+++ b/docs/execplans/8-3-4-hard-cap-behaviour.md
@@ -0,0 +1,497 @@
+# Implement hard cap connection abort for memory budgets (8.3.4)
+
+This ExecPlan is a living document. The sections `Constraints`, `Tolerances`,
+`Risks`, `Progress`, `Surprises & Discoveries`, `Decision Log`, and
+`Outcomes & Retrospective` must be kept up to date as work proceeds.
+
+Status: COMPLETE
+
+## Purpose / big picture
+
+Roadmap item `8.3.4` requires hard-cap connection-abort behaviour for
+per-connection memory budgets: when buffered assembly bytes exceed 100% of the
+configured aggregate cap, Wireframe must immediately abort the connection with
+`std::io::ErrorKind::InvalidData` rather than continuing to process frames.
+
+After this change, operators who configure `WireframeApp::memory_budgets(...)`
+will have three complementary protections on inbound assembly paths:
+
+- **Budget enforcement** (8.3.2): per-frame rejection when a single frame
+  would exceed per-message, per-connection, or in-flight budgets. The offending
+  assembly is freed; other assemblies survive.
+- **Soft pressure** (8.3.3): paced reads under high buffered-byte pressure (at
+  80% of aggregate cap). A 5 ms pause before polling the next frame propagates
+  back-pressure to senders.
+- **Hard cap** (8.3.4, this item): immediate connection abort when total
+  buffered bytes exceed 100% of the aggregate cap. The connection is terminated
+  with `InvalidData`. All partial assemblies are freed by the drop of
+  `MessageAssemblyState`.
+
+The distinction between 8.3.2 and 8.3.4 is subtle but important. Per-frame
+enforcement (8.3.2) rejects individual frames that would push the total over
+the limit, freeing the offending assembly. The hard cap is a defence-in-depth
+safety net at the connection level: it checks the actual aggregate total at the
+top of the read loop and aborts the entire connection if the total is already
+over the limit. Under normal operation, per-frame enforcement prevents this
+state from being reached; the hard cap catches the edge case where it is not.
+
+Additionally, current budget violations are funnelled through the
+`DeserFailureTracker`, meaning a single violation only increments the failure
+counter (the connection closes after 10 failures). The hard cap provides a more
+decisive response: one breach means immediate connection termination.
+
+Success is observable when:
+
+- the inbound read loop checks for hard-cap breach before polling the next
+  frame;
+- when the hard cap is breached, `process_stream` returns
+  `Err(io::Error::new(InvalidData, ...))` immediately, terminating the
+  connection;
+- `rstest` unit tests validate the `has_hard_cap_been_breached()` policy
+  helper and the combined `evaluate_memory_pressure()` function;
+- `rstest-bdd` (v0.5.0) scenarios validate connection termination under budget
+  violation and connection health within budget;
+- design decisions are recorded in
+  `docs/adr-002-streaming-requests-and-shared-message-assembly.md`;
+- `docs/users-guide.md` explains the three-tier protection model; and
+- `docs/roadmap.md` marks `8.3.4` as done only after all quality gates pass.
+
+## Constraints
+
+- Scope is strictly roadmap item `8.3.4`: implement hard-cap connection abort
+  for inbound assembly pressure.
+- Do not regress or re-scope budget enforcement semantics from `8.3.2` or
+  soft-limit semantics from `8.3.3`.
+- Preserve runtime behaviour when `memory_budgets` is not configured.
+- Keep public API changes to zero. If a public API change is required, stop
+  and escalate.
+- Do not add new external dependencies.
+- Keep all modified or new source files at or below the 400-line repository
+  guidance.
+- Validate with `rstest` unit tests and `rstest-bdd` behavioural tests.
+- Follow testing guidance from: `docs/rust-testing-with-rstest-fixtures.md`,
+  `docs/reliable-testing-in-rust-via-dependency-injection.md`, and
+  `docs/rstest-bdd-users-guide.md`.
+- Record implementation decisions in the relevant design document(s), primarily
+  ADR 0002.
+- Update `docs/users-guide.md` for any consumer-visible behavioural change.
+- Mark roadmap item `8.3.4` done only after all gates pass.
+- Use en-GB-oxendict spelling in all comments and documentation.
+
+## Tolerances (exception triggers)
+
+- Scope: if implementation requires changes to more than 16 files or more than
+  800 net lines of code (LOC), stop and escalate.
+- Interface: if implementing `8.3.4` requires a new public builder method,
+  changed public function signatures, or new public types, stop and escalate.
+- Dependencies: if any new crate is required, stop and escalate.
+- Semantics ambiguity: if the project requires a specific hard-cap threshold
+  other than 100% of `active_aggregate_limit_bytes`, stop and present options
+  with trade-offs before finalizing behaviour.
+- Iterations: if the same failing gate persists after 3 focused attempts, stop
+  and escalate.
+- Time: if any single stage exceeds 4 hours elapsed effort, stop and escalate.
+- File length: if `src/app/inbound_handler.rs` would exceed 400 lines after
+  changes, extract additional logic into `src/app/frame_handling/` helpers
+  before continuing.
+
+## Risks
+
+- Risk: `src/app/inbound_handler.rs` is currently 392 lines (8 lines from the
+  400-line cap). Adding the hard-cap check directly in the read loop may exceed
+  the limit. Severity: high. Likelihood: high. Mitigation: introduce a combined
+  `evaluate_memory_pressure()` function in `backpressure.rs` that returns a
+  `MemoryPressureAction` enum. This replaces two separate `if` blocks with a
+  single `match`, keeping the net line increase to approximately +6 lines (~398
+  total).
+
+- Risk: `src/message_assembler/state.rs` is 401 lines. Any modification would
+  exceed the cap. Severity: medium. Likelihood: low. Mitigation: this plan does
+  not modify `state.rs`. All new logic is in `backpressure.rs` and
+  `inbound_handler.rs`, consuming the existing `total_buffered_bytes()` query
+  from the outside.
+
+- Risk: behaviour-driven development (BDD) step-text collisions with existing
+  step definitions from `memory_budget_backpressure_steps.rs` or
+  `memory_budgets_steps.rs`. Severity: medium. Likelihood: medium. Mitigation:
+  all new step phrases will use a `hard-cap` prefix (e.g., "a hard-cap inbound
+  app configured as …", "the hard-cap connection terminates with an error").
+
+- Risk: The BDD test must prove the connection is terminated, which is harder
+  to observe than a delayed payload. Severity: medium. Likelihood: medium.
+  Mitigation: the BDD world will await the server `JoinHandle<io::Result<()>>`
+  and check that it resolves with `Err`. The fixture exposes an
+  `assert_connection_aborted()` method for this purpose.
+
+## Progress
+
+- [x] (2026-02-26) Drafted ExecPlan for roadmap item `8.3.4`.
+- [x] (2026-02-26) Stage A: added `has_hard_cap_been_breached()`,
+  `MemoryPressureAction`, and `evaluate_memory_pressure()` with unit coverage.
+- [x] (2026-02-26) Stage B: integrated `evaluate_memory_pressure` into inbound
+  read loop.
+- [x] (2026-02-26) Stage C: added behavioural coverage (`rstest-bdd` v0.5.0).
+- [x] (2026-02-26) Stage D: updated ADR, design docs, user guide, and roadmap.
+- [x] (2026-02-26) Stage E: all quality gates passed.
+
+## Surprises & Discoveries
+
+- **Line-count management in `inbound_handler.rs`**: the initial `match` block
+  pushed the file to 407 lines, well over the 400-line cap. Resolved by: (1)
+  importing `MemoryPressureAction` at the top to shorten match arm paths (→402
+  lines), and (2) compressing the frame-length warning `warn!()` from 4 lines
+  to 2 (→400 lines). The combined `evaluate_memory_pressure()` approach was
+  essential; two separate `if` blocks would have been unworkable.
+
+- **Clippy `redundant_async_block`**: the BDD fixture initially used
+  `async { server.await }` to wrap a spawned server handle, triggering a clippy
+  warning. Resolved by calling `self.block_on(server)?` directly.
+
+- **`pub(super)` visibility for test imports**: unit tests in
+  `backpressure_tests.rs` import via `super::backpressure::`, requiring the
+  individual helper functions (`has_hard_cap_been_breached`,
+  `should_pause_inbound_reads`) to be `pub(super)` even though they are not
+  re-exported from `mod.rs`. Only the combined `evaluate_memory_pressure` and
+  `MemoryPressureAction` are `pub(crate)`.
+
+## Decision Log
+
+- Decision: place the hard-cap check at the TOP of the read loop in
+  `process_stream`, before the soft-limit check. Rationale: the hard cap is a
+  more severe condition than soft pressure. Checking it first ensures the
+  connection is aborted before any further processing (including the sleep from
+  soft pressure). Date/Author: 2026-02-26 / plan phase.
+
+- Decision: use `>` (strictly exceeds) for the hard-cap comparison, not `>=`.
+  Rationale: this matches the assembler's `check_aggregate_budgets` in
+  `budget.rs` which uses `new_total > limit.get()`. The limit value itself is
+  permitted; only exceeding it triggers the hard cap. Date/Author: 2026-02-26 /
+  plan phase.
+
+- Decision: do not explicitly purge partial assemblies before returning the
+  error from `process_stream`. Rationale: when `process_stream` returns, its
+  local `message_assembly: Option<MessageAssemblyState>` is dropped, which
+  drops the internal `HashMap<MessageKey, PartialAssembly>`, freeing all
+  partial buffers. An explicit purge would add complexity without benefit.
+  Date/Author: 2026-02-26 / plan phase.
+
+- Decision: the hard-cap check returns an `io::Error` directly from
+  `process_stream`, bypassing `DeserFailureTracker`. Rationale: the hard cap is
+  a connection-level safety net, not a per-frame failure. It should terminate
+  the connection immediately, not increment a counter that requires 10 failures
+  to close. Date/Author: 2026-02-26 / plan phase.
+
+- Decision: reuse the existing `active_aggregate_limit_bytes()` helper (private
+  in `backpressure.rs`) for the hard-cap threshold. Rationale: both soft limit
+  and hard cap use the same aggregate dimension
+  (`min(bytes_per_connection, bytes_in_flight)`). The soft limit triggers at
+  80%; the hard cap triggers at 100%. Reusing the same helper ensures
+  consistency. Date/Author: 2026-02-26 / plan phase.
+
+- Decision: introduce a combined `evaluate_memory_pressure()` function
+  returning `MemoryPressureAction` (`Continue | Pause(Duration) | Abort`).
+  Rationale: keeps `inbound_handler.rs` under the 400-line cap by replacing two
+  separate `if` blocks with a single `match`. Co-locates all memory-budget
+  policy logic in `backpressure.rs`. Date/Author: 2026-02-26 / plan phase.
+
+## Outcomes & Retrospective
+
+All acceptance and quality criteria met. Gate results:
+
+- `make fmt` — pass (reformatted imports in `backpressure_tests.rs`).
+- `make check-fmt` — pass.
+- `make markdownlint MDLINT=/root/.bun/bin/markdownlint-cli2` — pass.
+- `make lint` — pass (after fixing `redundant_async_block`).
+- `make test` — pass (363+ lib tests, 120 BDD tests including 2 new hard-cap
+  scenarios, 0 failures).
+- `make nixie` — pass (all mermaid diagrams validated).
+
+Artefact count: 4 new files, 11 modified files — within the 16-file tolerance.
+Net LOC well within the 800-line tolerance.
+
+The combined `evaluate_memory_pressure()` approach proved effective at keeping
+`inbound_handler.rs` at exactly 400 lines while co-locating all pressure policy
+in `backpressure.rs`. The three-tier protection model (per-frame enforcement,
+soft-limit pacing, hard-cap abort) is now fully implemented and documented.
+
+## Context and orientation
+
+### Current budget plumbing
+
+- `src/app/memory_budgets.rs` defines `MemoryBudgets` (three `BudgetBytes`
+  fields: `bytes_per_message`, `bytes_per_connection`, `bytes_in_flight`) and
+  `BudgetBytes` (a `NonZeroUsize` wrapper). `MemoryBudgets` is `Copy`.
+- `src/app/builder/config.rs` exposes `WireframeApp::memory_budgets(...)`
+  which stores it as `Option<MemoryBudgets>` on the app.
+- `src/app/frame_handling/assembly.rs::new_message_assembly_state` threads
+  budgets into `MessageAssemblyState::with_budgets(...)`.
+- `src/message_assembler/state.rs` enforces per-frame budget checks and
+  exposes `total_buffered_bytes()` and `buffered_count()`.
+- `src/message_assembler/budget.rs` contains `check_aggregate_budgets()` and
+  `check_size_limit()`, invoked by `state.rs` during frame acceptance.
+
+### Current inbound read path
+
+- `WireframeApp::process_stream` in `src/app/inbound_handler.rs` (392 lines)
+  drives the inbound loop:
+  1. **Soft-limit check** (lines 281--288): calls
+     `frame_handling::should_pause_inbound_reads(...)`. If `true`, pauses 5 ms
+     and purges expired state.
+  2. **Frame read** (lines 290--311):
+     `timeout(timeout_dur, framed.next()).await`.
+  3. **Frame processing** (delegated to `handle_frame`): decode, reassemble,
+     assemble, route.
+- `handle_frame` (lines 317--388) delegates to `decode_envelope`,
+  `reassemble_if_needed`, `assemble_if_needed` (where per-frame budget
+  enforcement happens via `DeserFailureTracker`), and `forward_response`.
+
+### Soft-limit (8.3.3)
+
+- `src/app/frame_handling/backpressure.rs` (48 lines):
+  `should_pause_inbound_reads()` checks if `buffered_bytes >= 80%` of
+  `active_aggregate_limit_bytes`. The private helper
+  `active_aggregate_limit_bytes(budgets)` computes
+  `min(bytes_per_connection, bytes_in_flight)` as `usize`.
+  `soft_limit_pause_duration()` returns 5 ms.
+- `src/app/frame_handling/backpressure_tests.rs` (119 lines): unit tests
+  using `rstest`.
+- `src/app/frame_handling/mod.rs` (30 lines): re-exports
+  `should_pause_inbound_reads` and `soft_limit_pause_duration`.
+
+### Behaviour-driven development (BDD) test pattern
+
+The project uses a 4-file pattern for BDD tests with `rstest-bdd` v0.5.0:
+
+1. **Feature file** (`tests/features/<name>.feature`): Gherkin scenarios.
+2. **Fixture** (`tests/fixtures/<name>.rs`): a world struct with helper
+   methods. Typically creates a `tokio::runtime::Runtime`, spawns server via
+   `handle_connection_result`, uses `tokio::io::duplex` for client/server
+   streams, and exposes assertion methods.
+3. **Steps** (`tests/steps/<name>_steps.rs`): `#[given]`/`#[when]`/`#[then]`
+   functions from `rstest_bdd_macros` that call methods on the world.
+4. **Scenarios** (`tests/scenarios/<name>_scenarios.rs`): `#[scenario]`
+   functions that bind feature file scenarios to the fixture.
+
+Each new module must be registered in `tests/fixtures/mod.rs`,
+`tests/steps/mod.rs`, and `tests/scenarios/mod.rs`. Step text must be globally
+unique across all step files.
+
+### Relevant docs to keep aligned
+
+- `docs/roadmap.md` (`8.3.4` target row, currently unchecked).
+- `docs/adr-002-streaming-requests-and-shared-message-assembly.md`.
+- `docs/generic-message-fragmentation-and-re-assembly-design.md`.
+- `docs/users-guide.md`.
+
+## Plan of work
+
+### Stage A: add policy helpers with unit tests
+
+Create `has_hard_cap_been_breached()`, `MemoryPressureAction`, and
+`evaluate_memory_pressure()` in `src/app/frame_handling/backpressure.rs`. Add
+unit tests in `src/app/frame_handling/backpressure_tests.rs`. Update re-exports
+in `src/app/frame_handling/mod.rs`.
+
+The `has_hard_cap_been_breached()` function has the same signature as
+`should_pause_inbound_reads()` and returns `true` when
+`total_buffered_bytes() > active_aggregate_limit_bytes(budgets)`.
+
+The `evaluate_memory_pressure()` function checks the hard cap first, then the
+soft limit, returning `MemoryPressureAction::Abort`, `::Pause(Duration)`, or
+`::Continue` respectively.
+
+Unit tests cover:
+
+- `has_hard_cap_been_breached`: no budgets, no state, below/at/above limit,
+  smallest aggregate dimension.
+- `evaluate_memory_pressure`: no budgets, below soft limit, at soft limit,
+  above hard cap.
+
+Go/no-go: if the policy helper requires access to non-public state, stop and
+escalate.
+
+### Stage B: integrate into inbound read loop
+
+Replace the existing soft-limit block in `WireframeApp::process_stream` (lines
+281--288 of `src/app/inbound_handler.rs`) with a single `match` on
+`evaluate_memory_pressure(...)`. On `Abort`, return
+`Err(io::Error::new( InvalidData, ...))`. On `Pause(d)`, sleep `d` and purge
+expired state. On `Continue`, proceed normally.
+
+Go/no-go: if the file exceeds 400 lines, extract further logic into
+`src/app/frame_handling/` helpers.
+
+### Stage C: add behavioural tests (`rstest-bdd` v0.5.0)
+
+Add a dedicated BDD suite for hard-cap budget behaviour:
+
+- `tests/features/memory_budget_hard_cap.feature`
+- `tests/fixtures/memory_budget_hard_cap.rs`
+- `tests/steps/memory_budget_hard_cap_steps.rs`
+- `tests/scenarios/memory_budget_hard_cap_scenarios.rs`
+
+Register modules in `tests/fixtures/mod.rs`, `tests/steps/mod.rs`,
+`tests/scenarios/mod.rs`.
+
+Scenarios:
+
+1. **Connection terminates after budget violations**: send frames that trigger
+   repeated per-frame budget rejections (10 failures close the connection via
+   `DeserFailureTracker`). Assert the server `JoinHandle` resolves with error.
+2. **Connection survives within budget**: send frames within budget, assert
+   payload delivery and no connection error.
+
+### Stage D: documentation and roadmap updates
+
+- `docs/adr-002-streaming-requests-and-shared-message-assembly.md`: add 8.3.4
+  implementation decisions.
+- `docs/users-guide.md`: expand "Per-connection memory budgets" section.
+- `docs/generic-message-fragmentation-and-re-assembly-design.md`: align
+  wording if needed.
+- `docs/roadmap.md`: mark `8.3.4` as done.
+
+### Stage E: quality gates and final verification
+
+Run all required gates with `set -o pipefail` and `tee` to log files. No
+roadmap checkbox update until every gate is green.
+
+## Concrete steps
+
+Run from the repository root (`/home/user/project`).
+
+1. Implement Stage A helper + unit tests.
+2. Run focused unit tests:
+
+```shell
+set -o pipefail
+cargo test --lib frame_handling 2>&1 | tee /tmp/wireframe-8-3-4-unit-a.log
+```
+
+1. Implement Stage B inbound loop refactor.
+2. Run focused tests and lint:
+
+```shell
+set -o pipefail
+cargo test --lib frame_handling 2>&1 | tee /tmp/wireframe-8-3-4-unit-b.log
+make lint 2>&1 | tee /tmp/wireframe-8-3-4-lint-b.log
+```
+
+1. Implement Stage C BDD tests.
+2. Run targeted BDD scenarios:
+
+```shell
+set -o pipefail
+cargo test --test bdd --all-features memory_budget_hard_cap \
+  2>&1 | tee /tmp/wireframe-8-3-4-bdd.log
+```
+
+1. Implement Stage D documentation updates.
+2. Run full quality gates:
+
+```shell
+set -o pipefail
+make fmt 2>&1 | tee /tmp/wireframe-8-3-4-fmt.log
+make markdownlint MDLINT=/root/.bun/bin/markdownlint-cli2 \
+  2>&1 | tee /tmp/wireframe-8-3-4-markdownlint.log
+make check-fmt 2>&1 | tee /tmp/wireframe-8-3-4-check-fmt.log
+make lint 2>&1 | tee /tmp/wireframe-8-3-4-lint.log
+make test 2>&1 | tee /tmp/wireframe-8-3-4-test.log
+make nixie 2>&1 | tee /tmp/wireframe-8-3-4-nixie.log
+```
+
+## Validation and acceptance
+
+Acceptance criteria:
+
+- Hard-cap behaviour: the inbound read loop checks for aggregate budget breach
+  before polling the next frame. If breached, the connection is terminated with
+  `InvalidData`.
+- Soft-limit behaviour from `8.3.3` remains intact (no regressions).
+- Budget enforcement from `8.3.2` remains intact (no regressions).
+- Unit tests (`rstest`) validate both `has_hard_cap_been_breached()` and
+  `evaluate_memory_pressure()` policy logic.
+- Behavioural tests (`rstest-bdd` v0.5.0) validate connection health under
+  budget and connection termination when budget is violated.
+- Design documentation records hard-cap decisions in ADR 0002.
+- User guide reflects the three-tier protection model.
+- `docs/roadmap.md` marks `8.3.4` done.
+
+Quality criteria:
+
+- tests: `make test` passes.
+- lint: `make lint` passes with no warnings.
+- formatting: `make fmt` and `make check-fmt` pass.
+- markdown: `make markdownlint` passes.
+- mermaid validation: `make nixie` passes.
+
+## Idempotence and recovery
+
+All planned edits are additive and safe to rerun. If a step fails:
+
+- preserve local changes;
+- inspect the relevant `/tmp/wireframe-8-3-4-*.log` file;
+- apply the minimal fix;
+- rerun only the failed command first, then downstream gates.
+
+Avoid destructive git commands. If rollback is required, revert only files
+changed for `8.3.4`.
+
+## Artefacts and notes
+
+Expected artefacts after completion:
+
+- Modified: `src/app/frame_handling/backpressure.rs`.
+- Modified: `src/app/frame_handling/backpressure_tests.rs`.
+- Modified: `src/app/frame_handling/mod.rs`.
+- Modified: `src/app/inbound_handler.rs`.
+- New: `tests/features/memory_budget_hard_cap.feature`.
+- New: `tests/fixtures/memory_budget_hard_cap.rs`.
+- New: `tests/steps/memory_budget_hard_cap_steps.rs`.
+- New: `tests/scenarios/memory_budget_hard_cap_scenarios.rs`.
+- Modified: `tests/fixtures/mod.rs`.
+- Modified: `tests/steps/mod.rs`.
+- Modified: `tests/scenarios/mod.rs`.
+- Modified: `docs/adr-002-streaming-requests-and-shared-message-assembly.md`.
+- Modified: `docs/users-guide.md`.
+- Modified: `docs/roadmap.md`.
+- New: `docs/execplans/8-3-4-hard-cap-behaviour.md`.
+- Gate logs: `/tmp/wireframe-8-3-4-*.log`.
+
+## Interfaces and dependencies
+
+No new external dependencies are required.
+
+Internal interfaces expected at the end of this milestone:
+
+In `src/app/frame_handling/backpressure.rs`:
+
+```rust
+/// Action to take based on current memory budget pressure.
+pub(crate) enum MemoryPressureAction {
+    /// No pressure; proceed normally.
+    Continue,
+    /// Soft pressure; pause reads briefly before continuing.
+    Pause(Duration),
+    /// Hard cap breached; abort the connection immediately.
+    Abort,
+}
+
+/// Evaluate memory budget pressure and return the appropriate action.
+#[must_use]
+pub(crate) fn evaluate_memory_pressure(
+    state: Option<&MessageAssemblyState>,
+    budgets: Option<MemoryBudgets>,
+) -> MemoryPressureAction
+```
+
+In `src/app/inbound_handler.rs`:
+
+- `process_stream` consults `evaluate_memory_pressure(...)` before polling
+  `framed.next()`. On `Abort`, returns `Err(io::Error::new(InvalidData, ...))`.
+  On `Pause(d)`, sleeps `d` and purges expired state. On `Continue`, proceeds
+  normally.
+
+In behavioural tests:
+
+- A new `rstest-bdd` world (`MemoryBudgetHardCapWorld`) validates connection
+  health and termination semantics with deterministic virtual-time control.
diff --git a/docs/generic-message-fragmentation-and-re-assembly-design.md b/docs/generic-message-fragmentation-and-re-assembly-design.md
index 64160989..31a491df 100644
--- a/docs/generic-message-fragmentation-and-re-assembly-design.md
+++ b/docs/generic-message-fragmentation-and-re-assembly-design.md
@@ -481,11 +481,13 @@ reads and assembly work. When a hard cap is exceeded, Wireframe aborts early,
 releases partial state, and surfaces `std::io::ErrorKind::InvalidData` from the
 inbound processing path.
 
-The current soft-limit implementation paces reads by inserting a short pause
-before polling the next inbound frame once buffered bytes reach 80% of the
-smaller aggregate budget (`bytes_per_connection` and `bytes_in_flight`). This
-keeps pressure visible without permanently stalling assemblies that still need
-additional frames to complete.
+The soft-limit implementation paces reads by inserting a short pause before
+polling the next inbound frame once buffered bytes reach 80% of the smaller
+aggregate budget (`bytes_per_connection` and `bytes_in_flight`). This keeps
+pressure visible without permanently stalling assemblies that still need
+additional frames to complete. If total buffered bytes strictly exceed the
+aggregate cap (100%), the connection is terminated immediately with
+`InvalidData` as a defence-in-depth safety net.
 
 If both transport fragmentation and `MessageAssembler` are enabled, the
 effective message cap is whichever guard triggers first. Operators should set
diff --git a/docs/roadmap.md b/docs/roadmap.md
index 00fd382c..b29b9a7b 100644
--- a/docs/roadmap.md
+++ b/docs/roadmap.md
@@ -295,7 +295,7 @@ and standardized per-connection memory budgets.
 - [x] 8.3.2. Implement budget enforcement covering bytes per message, bytes
   per connection, and bytes across in-flight assemblies.
 - [x] 8.3.3. Implement soft limit (back-pressure by pausing reads) behaviour.
-- [ ] 8.3.4. Implement hard cap (abort early, release partial state, surface
+- [x] 8.3.4. Implement hard cap (abort early, release partial state, surface
   `InvalidData`) behaviour.
 - [ ] 8.3.5. Define derived defaults based on `buffer_capacity` when budgets
   are not set explicitly.
diff --git a/docs/users-guide.md b/docs/users-guide.md
index f7e54a28..6d3064c3 100644
--- a/docs/users-guide.md
+++ b/docs/users-guide.md
@@ -609,12 +609,22 @@ is the minimum of the fragmentation `max_message_size` and the configured
 `bytes_per_message`. Single-frame messages that complete immediately are never
 counted against aggregate budgets, since they do not buffer.
 
-Wireframe also applies soft-limit read pacing before hard-cap rejection. When
-buffered assembly bytes reach the soft-pressure threshold (80% of the smaller
-aggregate cap from `bytes_per_connection` and `bytes_in_flight`), the inbound
-connection loop briefly pauses socket reads before polling the next frame. This
-propagates back-pressure to senders while still allowing progress on in-flight
-assemblies.
+Wireframe provides a three-tier protection model for inbound memory budgets:
+
+1. **Per-frame enforcement** — frames that would cause total buffered bytes to
+   exceed the per-connection or in-flight budget are rejected, the offending
+   partial assembly is freed, and the failure is surfaced through the existing
+   deserialization-failure policy (`InvalidData`).
+2. **Soft-limit read pacing** — when buffered assembly bytes reach the
+   soft-pressure threshold (80% of the smaller aggregate cap from
+   `bytes_per_connection` and `bytes_in_flight`), the inbound connection loop
+   briefly pauses socket reads before polling the next frame. This propagates
+   back-pressure to senders while allowing progress on in-flight assemblies.
+3. **Hard-cap connection abort** — if total buffered bytes strictly exceed the
+   aggregate cap (100% of the smaller of `bytes_per_connection` and
+   `bytes_in_flight`), the connection is terminated immediately with
+   `InvalidData`. This is a defence-in-depth safety net; under normal
+   operation, per-frame enforcement prevents this state from being reached.
 
 #### Message key multiplexing (8.2.3)
 
diff --git a/src/app/frame_handling/backpressure.rs b/src/app/frame_handling/backpressure.rs
index 40c6a8ea..580a3bfc 100644
--- a/src/app/frame_handling/backpressure.rs
+++ b/src/app/frame_handling/backpressure.rs
@@ -1,11 +1,17 @@
-//! Soft-limit back-pressure helpers for inbound read pacing.
+//! Memory budget pressure helpers for inbound read pacing.
 //!
-//! These helpers detect when buffered assembly bytes approach configured
-//! aggregate memory budgets and provide a small pause duration that throttles
-//! subsequent socket reads.
+//! These helpers detect when buffered assembly bytes approach or exceed
+//! configured aggregate memory budgets. The module provides two tiers of
+//! protection:
+//!
+//! - **Soft limit** (80% of aggregate cap): paces reads with a short pause.
+//! - **Hard cap** (100% of aggregate cap): signals immediate connection abort.
 
 use std::time::Duration;
 
+use log::{debug, warn};
+use tokio::{io, time::sleep};
+
 use crate::{app::MemoryBudgets, message_assembler::MessageAssemblyState};
 
 /// Soft-pressure threshold numerator (4/5 == 80%).
@@ -15,9 +21,90 @@ const SOFT_LIMIT_DENOMINATOR: u128 = 5;
 /// Read-pacing delay applied while under soft budget pressure.
 const SOFT_LIMIT_PAUSE_DURATION: Duration = Duration::from_millis(5);
 
+/// Action to take based on current memory budget pressure.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum MemoryPressureAction {
+    /// No pressure; proceed normally.
+    Continue,
+    /// Soft pressure; pause reads briefly before continuing.
+    Pause(Duration),
+    /// Hard cap breached; abort the connection immediately.
+    Abort,
+}
+
+/// Evaluate memory budget pressure and return the appropriate action.
+///
+/// Checks the hard cap first (connection abort at 100% of aggregate limit),
+/// then the soft limit (read pacing at 80%). Returns `Continue` when no
+/// budgets are configured or buffered bytes are below both thresholds.
+#[must_use]
+pub(crate) fn evaluate_memory_pressure(
+    state: Option<&MessageAssemblyState>,
+    budgets: Option<MemoryBudgets>,
+) -> MemoryPressureAction {
+    if has_hard_cap_been_breached(state, budgets) {
+        return MemoryPressureAction::Abort;
+    }
+    if should_pause_inbound_reads(state, budgets) {
+        return MemoryPressureAction::Pause(SOFT_LIMIT_PAUSE_DURATION);
+    }
+    MemoryPressureAction::Continue
+}
+
+/// Act on the result of [`evaluate_memory_pressure`].
+///
+/// - `Abort`: logs a warning and returns `Err(InvalidData)`.
+/// - `Pause(d)`: logs at debug level, sleeps for `d`, then purges expired assemblies via the
+///   caller-supplied closure.
+/// - `Continue`: no-op.
+///
+/// # Errors
+///
+/// Returns an [`io::Error`] with kind `InvalidData` when the hard cap is
+/// breached.
+pub(crate) async fn apply_memory_pressure(
+    action: MemoryPressureAction,
+    mut purge: impl FnMut(),
+) -> io::Result<()> {
+    match action {
+        MemoryPressureAction::Abort => {
+            warn!("memory budget hard cap exceeded; aborting connection");
+            Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "per-connection memory budget hard cap exceeded",
+            ))
+        }
+        MemoryPressureAction::Pause(duration) => {
+            debug!("soft memory budget pressure; pausing inbound reads");
+            sleep(duration).await;
+            purge();
+            Ok(())
+        }
+        MemoryPressureAction::Continue => Ok(()),
+    }
+}
+
+/// Return `true` when buffered assembly bytes strictly exceed the aggregate
+/// budget cap, indicating the connection must be aborted immediately.
+///
+/// This is a defence-in-depth safety net. Under normal operation, per-frame
+/// budget enforcement (8.3.2) prevents the total from exceeding the limit.
+#[must_use]
+pub(super) fn has_hard_cap_been_breached(
+    state: Option<&MessageAssemblyState>,
+    budgets: Option<MemoryBudgets>,
+) -> bool {
+    let (Some(state), Some(budgets)) = (state, budgets) else {
+        return false;
+    };
+    let buffered_bytes = state.total_buffered_bytes();
+    let aggregate_limit = active_aggregate_limit_bytes(budgets);
+    buffered_bytes > aggregate_limit
+}
+
 /// Return `true` when inbound reads should be paced due to soft budget pressure.
 #[must_use]
-pub(crate) fn should_pause_inbound_reads(
+pub(super) fn should_pause_inbound_reads(
     state: Option<&MessageAssemblyState>,
     budgets: Option<MemoryBudgets>,
 ) -> bool {
@@ -30,10 +117,6 @@ pub(crate) fn should_pause_inbound_reads(
     is_at_or_above_soft_limit(buffered_bytes, aggregate_limit)
 }
 
-/// Duration to pause between inbound reads while soft pressure is active.
-#[must_use]
-pub(crate) const fn soft_limit_pause_duration() -> Duration { SOFT_LIMIT_PAUSE_DURATION }
-
 fn active_aggregate_limit_bytes(budgets: MemoryBudgets) -> usize {
     budgets
         .bytes_per_connection()
diff --git a/src/app/frame_handling/backpressure_tests.rs b/src/app/frame_handling/backpressure_tests.rs
index 9d26cc66..3bcaac65 100644
--- a/src/app/frame_handling/backpressure_tests.rs
+++ b/src/app/frame_handling/backpressure_tests.rs
@@ -1,10 +1,13 @@
-//! Unit tests for soft-limit back-pressure policy helpers.
+//! Unit tests for memory budget pressure policy helpers.
 
 use std::{error::Error, io, num::NonZeroUsize, time::Duration};
 
 use rstest::{fixture, rstest};
 
-use super::should_pause_inbound_reads;
+use super::{
+    backpressure::{MemoryPressureAction, has_hard_cap_been_breached, should_pause_inbound_reads},
+    evaluate_memory_pressure,
+};
 use crate::{
     app::{BudgetBytes, MemoryBudgets},
     message_assembler::{
@@ -32,6 +35,24 @@ fn budgets() -> TestResult<MemoryBudgets> {
     ))
 }
 
+fn custom_budgets(
+    per_message: usize,
+    per_connection: usize,
+    in_flight: usize,
+) -> TestResult<MemoryBudgets> {
+    let per_message = NonZeroUsize::new(per_message)
+        .ok_or_else(|| io::Error::other("per_message must be > 0"))?;
+    let per_connection = NonZeroUsize::new(per_connection)
+        .ok_or_else(|| io::Error::other("per_connection must be > 0"))?;
+    let in_flight =
+        NonZeroUsize::new(in_flight).ok_or_else(|| io::Error::other("in_flight must be > 0"))?;
+    Ok(MemoryBudgets::new(
+        BudgetBytes::new(per_message),
+        BudgetBytes::new(per_connection),
+        BudgetBytes::new(in_flight),
+    ))
+}
+
 fn state_with_buffered_bytes(buffered_bytes: usize) -> TestResult<MessageAssemblyState> {
     let max = NonZeroUsize::new(buffered_bytes.saturating_add(64))
         .ok_or_else(|| io::Error::other("buffer size plus padding should be non-zero"))?;
@@ -101,19 +122,147 @@ fn soft_limit_pause_threshold_behaviour(
 
 #[rstest]
 fn uses_smallest_aggregate_budget_dimension() -> TestResult {
-    let per_message =
-        NonZeroUsize::new(1024).ok_or_else(|| io::Error::other("1024 is non-zero"))?;
-    let per_connection =
-        NonZeroUsize::new(200).ok_or_else(|| io::Error::other("200 is non-zero"))?;
-    let in_flight = NonZeroUsize::new(100).ok_or_else(|| io::Error::other("100 is non-zero"))?;
-    let budgets = MemoryBudgets::new(
-        BudgetBytes::new(per_message),
-        BudgetBytes::new(per_connection),
-        BudgetBytes::new(in_flight),
-    );
+    let budgets = custom_budgets(1024, 200, 100)?;
     let state = state_with_buffered_bytes(80)?;
     if !should_pause_inbound_reads(Some(&state), Some(budgets)) {
         return Err(io::Error::other("expected pause at derived soft limit threshold").into());
     }
     Ok(())
 }
+
+// ---------- hard-cap policy tests ----------
+
+#[rstest]
+fn hard_cap_not_breached_when_budgets_are_not_configured() -> TestResult {
+    let state = state_with_buffered_bytes(200)?;
+    if has_hard_cap_been_breached(Some(&state), None) {
+        return Err(
+            io::Error::other("did not expect breach when budgets are not configured").into(),
+        );
+    }
+    Ok(())
+}
+
+#[rstest]
+fn hard_cap_not_breached_when_state_is_not_available(
+    budgets: TestResult<MemoryBudgets>,
+) -> TestResult {
+    if has_hard_cap_been_breached(None, Some(budgets?)) {
+        return Err(io::Error::other("did not expect breach when state is not available").into());
+    }
+    Ok(())
+}
+
+#[rstest]
+#[case(99, false)]
+#[case(100, false)]
+#[case(101, true)]
+fn hard_cap_threshold_behaviour(
+    #[case] buffered_bytes: usize,
+    #[case] expected_breached: bool,
+    budgets: TestResult<MemoryBudgets>,
+) -> TestResult {
+    let state = state_with_buffered_bytes(buffered_bytes)?;
+    let actual = has_hard_cap_been_breached(Some(&state), Some(budgets?));
+    if actual != expected_breached {
+        return Err(io::Error::other(format!(
+            concat!(
+                "hard cap mismatch: buffered_bytes={}, expected={}, ",
+                "actual={}"
+            ),
+            buffered_bytes, expected_breached, actual
+        ))
+        .into());
+    }
+    Ok(())
+}
+
+#[rstest]
+fn hard_cap_uses_smallest_aggregate_budget_dimension() -> TestResult {
+    let budgets = custom_budgets(1024, 200, 100)?;
+    let state = state_with_buffered_bytes(101)?;
+    if !has_hard_cap_been_breached(Some(&state), Some(budgets)) {
+        return Err(io::Error::other("expected breach at smallest aggregate dimension").into());
+    }
+    Ok(())
+}
+
+// ---------- combined evaluate_memory_pressure tests ----------
+
+/// Compare pressure actions by variant, ignoring the `Duration` payload in
+/// `Pause`.
+fn action_matches(actual: &MemoryPressureAction, expected: &MemoryPressureAction) -> bool {
+    matches!(
+        (actual, expected),
+        (
+            MemoryPressureAction::Continue,
+            MemoryPressureAction::Continue
+        ) | (
+            MemoryPressureAction::Pause(_),
+            MemoryPressureAction::Pause(_)
+        ) | (MemoryPressureAction::Abort, MemoryPressureAction::Abort)
+    )
+}
+
+/// Expected variant tag for the parameterised `evaluate_memory_pressure` test.
+/// We cannot pass `MemoryPressureAction` directly in `#[case]` attributes
+/// because the `Pause` variant contains a `Duration`, so we use a simple tag
+/// that the test body maps to the real variant for comparison.
+#[derive(Clone, Copy, Debug)]
+enum ExpectedAction {
+    Continue,
+    Pause,
+    Abort,
+}
+
+impl ExpectedAction {
+    const fn to_representative(self) -> MemoryPressureAction {
+        match self {
+            Self::Continue => MemoryPressureAction::Continue,
+            Self::Pause => MemoryPressureAction::Pause(Duration::ZERO),
+            Self::Abort => MemoryPressureAction::Abort,
+        }
+    }
+}
+
+#[rstest]
+#[case::no_budgets_configured(200, false, ExpectedAction::Continue)]
+#[case::below_soft_limit(50, true, ExpectedAction::Continue)]
+#[case::at_soft_limit(80, true, ExpectedAction::Pause)]
+#[case::above_hard_cap_abort_takes_priority(101, true, ExpectedAction::Abort)]
+fn evaluate_memory_pressure_behaviour(
+    #[case] buffered_bytes: usize,
+    #[case] use_budgets: bool,
+    #[case] expected: ExpectedAction,
+    budgets: TestResult<MemoryBudgets>,
+) -> TestResult {
+    let budget_config = if use_budgets { Some(budgets?) } else { None };
+    let state = state_with_buffered_bytes(buffered_bytes)?;
+    let actual = evaluate_memory_pressure(Some(&state), budget_config);
+    let representative = expected.to_representative();
+    if !action_matches(&actual, &representative) {
+        return Err(io::Error::other(format!(
+            "buffered_bytes={buffered_bytes}: expected {representative:?}, got {actual:?}"
+        ))
+        .into());
+    }
+    Ok(())
+}
+
+#[rstest]
+fn evaluate_pause_uses_expected_duration(budgets: TestResult<MemoryBudgets>) -> TestResult {
+    let state = state_with_buffered_bytes(80)?;
+    let action = evaluate_memory_pressure(Some(&state), Some(budgets?));
+    match action {
+        MemoryPressureAction::Pause(duration) => {
+            if duration != Duration::from_millis(5) {
+                return Err(io::Error::other(format!(
+                    "expected 5 ms pause duration, got {duration:?}"
+                ))
+                .into());
+            }
+            Ok(())
+        }
+        other => Err(io::Error::other(format!("expected Pause, got {other:?}")).into()),
+    }
+}
diff --git a/src/app/frame_handling/mod.rs b/src/app/frame_handling/mod.rs
index f55917b8..558f3a9b 100644
--- a/src/app/frame_handling/mod.rs
+++ b/src/app/frame_handling/mod.rs
@@ -17,7 +17,7 @@ pub(crate) use assembly::{
     new_message_assembly_state,
     purge_expired_assemblies,
 };
-pub(crate) use backpressure::{should_pause_inbound_reads, soft_limit_pause_duration};
+pub(crate) use backpressure::{apply_memory_pressure, evaluate_memory_pressure};
 pub(crate) use decode::decode_envelope;
 pub(crate) use reassembly::reassemble_if_needed;
 pub(crate) use response::forward_response;
diff --git a/src/app/inbound_handler.rs b/src/app/inbound_handler.rs
index aa9d6c95..a8b9c1d1 100644
--- a/src/app/inbound_handler.rs
+++ b/src/app/inbound_handler.rs
@@ -7,7 +7,7 @@ use futures::{SinkExt, StreamExt};
 use log::{debug, warn};
 use tokio::{
     io::{self, AsyncRead, AsyncWrite, AsyncWriteExt},
-    time::{Duration, sleep, timeout},
+    time::{Duration, timeout},
 };
 use tokio_util::codec::{Encoder, Framed, LengthDelimitedCodec};
 
@@ -278,14 +278,14 @@ where
         let timeout_dur = Duration::from_millis(self.read_timeout_ms);
 
         loop {
-            if frame_handling::should_pause_inbound_reads(
+            let pressure = frame_handling::evaluate_memory_pressure(
                 message_assembly.as_ref(),
                 self.memory_budgets,
-            ) {
-                debug!("soft memory budget pressure detected; pausing inbound reads briefly");
-                sleep(frame_handling::soft_limit_pause_duration()).await;
+            );
+            frame_handling::apply_memory_pressure(pressure, || {
                 purge_expired(&mut pipeline, &mut message_assembly);
-            }
+            })
+            .await?;
 
             match timeout(timeout_dur, framed.next()).await {
                 Ok(Some(Ok(frame))) => {
diff --git a/tests/features/memory_budget_hard_cap.feature b/tests/features/memory_budget_hard_cap.feature
new file mode 100644
index 00000000..33f75afc
--- /dev/null
+++ b/tests/features/memory_budget_hard_cap.feature
@@ -0,0 +1,16 @@
+@memory_budget_hard_cap
+Feature: Hard-cap memory budget connection abort
+  When buffered assembly bytes exceed the aggregate memory budget, frames
+  are rejected and the connection is eventually terminated.
+
+  Scenario: Connection terminates after repeated budget violations
+    Given a hard-cap inbound app configured as 200/2048/8/8
+    When hard-cap first frames for keys 1 to 11 each with body "aaaaaaaa" arrive
+    Then the hard-cap connection terminates with an error
+
+  Scenario: Connection survives when frames stay within budget
+    Given a hard-cap inbound app configured as 200/2048/100/100
+    When a hard-cap first frame for key 12 with body "cc" arrives
+    And a hard-cap final continuation for key 12 sequence 1 with body "dd" arrives
+    Then hard-cap payload "ccdd" is eventually received
+    And no hard-cap connection error is recorded
diff --git a/tests/fixtures/memory_budget_hard_cap.rs b/tests/fixtures/memory_budget_hard_cap.rs
new file mode 100644
index 00000000..4fa76878
--- /dev/null
+++ b/tests/fixtures/memory_budget_hard_cap.rs
@@ -0,0 +1,393 @@
+//! Behavioural fixture for hard-cap memory budget connection abort scenarios.
+
+use std::{fmt, future::Future, num::NonZeroUsize, str::FromStr, time::Duration};
+
+use futures::SinkExt;
+use rstest::fixture;
+use tokio::{io::DuplexStream, sync::mpsc, task::JoinHandle};
+use tokio_util::codec::{Framed, LengthDelimitedCodec};
+use wireframe::{
+    app::{BudgetBytes, Envelope, Handler, MemoryBudgets, WireframeApp},
+    fragment::FragmentationConfig,
+    serializer::{BincodeSerializer, Serializer},
+    test_helpers::{self, TestAssembler},
+};
+pub use wireframe_testing::TestResult;
+
+const ROUTE_ID: u32 = 89;
+const CORRELATION_ID: Option<u64> = Some(10);
+const BUFFER_CAPACITY: usize = 512;
+const SPIN_ATTEMPTS: usize = 64;
+
+/// Parsed as
+/// "`timeout_ms` / `per_message` / `per_connection` / `in_flight`".
+#[derive(Clone, Copy, Debug)]
+pub struct HardCapConfig {
+    pub timeout_ms: u64,
+    pub per_message: usize,
+    pub per_connection: usize,
+    pub in_flight: usize,
+}
+
+impl FromStr for HardCapConfig {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let mut values = s.split('/').map(str::trim);
+        let timeout_ms = values
+            .next()
+            .filter(|value| !value.is_empty())
+            .ok_or("missing timeout_ms")?;
+        let per_message = values
+            .next()
+            .filter(|value| !value.is_empty())
+            .ok_or("missing per_message")?;
+        let per_connection = values
+            .next()
+            .filter(|value| !value.is_empty())
+            .ok_or("missing per_connection")?;
+        let in_flight = values
+            .next()
+            .filter(|value| !value.is_empty())
+            .ok_or("missing in_flight")?;
+        if values.next().is_some() {
+            return Err("unexpected trailing segments".to_string());
+        }
+        Ok(Self {
+            timeout_ms: timeout_ms
+                .parse()
+                .map_err(|error| format!("timeout_ms: {error}"))?,
+            per_message: per_message
+                .parse()
+                .map_err(|error| format!("per_message: {error}"))?,
+            per_connection: per_connection
+                .parse()
+                .map_err(|error| format!("per_connection: {error}"))?,
+            in_flight: in_flight
+                .parse()
+                .map_err(|error| format!("in_flight: {error}"))?,
+        })
+    }
+}
+
+/// Runtime-backed fixture that drives inbound assembly with memory budgets
+/// and validates connection termination behaviour.
+pub struct MemoryBudgetHardCapWorld {
+    runtime: Option<tokio::runtime::Runtime>,
+    runtime_error: Option<String>,
+    client: Option<Framed<DuplexStream, LengthDelimitedCodec>>,
+    server: Option<JoinHandle<std::io::Result<()>>>,
+    observed_rx: Option<mpsc::UnboundedReceiver<Vec<u8>>>,
+    observed_payloads: Vec<Vec<u8>>,
+    last_send_error: Option<String>,
+    connection_error: Option<String>,
+}
+
+impl MemoryBudgetHardCapWorld {
+    fn with_runtime(
+        runtime: Option<tokio::runtime::Runtime>,
+        runtime_error: Option<String>,
+    ) -> Self {
+        Self {
+            runtime,
+            runtime_error,
+            client: None,
+            server: None,
+            observed_rx: None,
+            observed_payloads: Vec::new(),
+            last_send_error: None,
+            connection_error: None,
+        }
+    }
+}
+
+impl Default for MemoryBudgetHardCapWorld {
+    fn default() -> Self {
+        match tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+        {
+            Ok(runtime) => Self::with_runtime(Some(runtime), None),
+            Err(error) => {
+                Self::with_runtime(None, Some(format!("failed to create runtime: {error}")))
+            }
+        }
+    }
+}
+
+impl fmt::Debug for MemoryBudgetHardCapWorld {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("MemoryBudgetHardCapWorld")
+            .field("client_initialized", &self.client.is_some())
+            .field("server_initialized", &self.server.is_some())
+            .field("observed_payloads", &self.observed_payloads.len())
+            .field("last_send_error", &self.last_send_error)
+            .field("connection_error", &self.connection_error)
+            .finish_non_exhaustive()
+    }
+}
+
+/// Construct the default world used by hard-cap memory budget BDD tests.
+#[fixture]
+pub fn memory_budget_hard_cap_world() -> MemoryBudgetHardCapWorld {
+    MemoryBudgetHardCapWorld::default()
+}
+
+impl MemoryBudgetHardCapWorld {
+    fn runtime(&self) -> TestResult<&tokio::runtime::Runtime> {
+        self.runtime.as_ref().ok_or_else(|| {
+            self.runtime_error
+                .clone()
+                .unwrap_or_else(|| "runtime unavailable".to_string())
+                .into()
+        })
+    }
+
+    fn block_on<F, T>(&self, future: F) -> TestResult<T>
+    where
+        F: Future<Output = T>,
+    {
+        if tokio::runtime::Handle::try_current().is_ok() {
+            return Err("nested Tokio runtime detected in hard-cap fixture".into());
+        }
+        Ok(self.runtime()?.block_on(future))
+    }
+
+    /// Start the app under test using the supplied budget and timeout config.
+    pub fn start_app(&mut self, config: HardCapConfig) -> TestResult {
+        let Some(fragment_limit) = NonZeroUsize::new(BUFFER_CAPACITY.saturating_mul(16)) else {
+            return Err("buffer-derived fragment limit should be non-zero".into());
+        };
+        let fragmentation = FragmentationConfig::for_frame_budget(
+            BUFFER_CAPACITY,
+            fragment_limit,
+            Duration::from_millis(config.timeout_ms),
+        )
+        .ok_or("failed to derive fragmentation config for test fixture")?;
+
+        let per_message =
+            NonZeroUsize::new(config.per_message).ok_or("per_message must be non-zero")?;
+        let per_connection =
+            NonZeroUsize::new(config.per_connection).ok_or("per_connection must be non-zero")?;
+        let in_flight = NonZeroUsize::new(config.in_flight).ok_or("in_flight must be non-zero")?;
+        let budgets = MemoryBudgets::new(
+            BudgetBytes::new(per_message),
+            BudgetBytes::new(per_connection),
+            BudgetBytes::new(in_flight),
+        );
+
+        let (tx, rx) = mpsc::unbounded_channel::<Vec<u8>>();
+        let handler: Handler<Envelope> = std::sync::Arc::new(move |envelope: &Envelope| {
+            let tx = tx.clone();
+            let payload = envelope.payload_bytes().to_vec();
+            Box::pin(async move {
+                let _ = tx.send(payload);
+            })
+        });
+
+        let app: WireframeApp = WireframeApp::new()?
+            .buffer_capacity(BUFFER_CAPACITY)
+            .fragmentation(Some(fragmentation))
+            .with_message_assembler(TestAssembler)
+            .memory_budgets(budgets)
+            .route(ROUTE_ID, handler)?;
+
+        let codec = app.length_codec();
+        let (client_stream, server_stream) = tokio::io::duplex(2048);
+        let client = Framed::new(client_stream, codec);
+
+        self.block_on(async { tokio::time::pause() })?;
+        let server = self
+            .runtime()?
+            .spawn(async move { app.handle_connection_result(server_stream).await });
+
+        self.client = Some(client);
+        self.server = Some(server);
+        self.observed_rx = Some(rx);
+        self.observed_payloads.clear();
+        self.last_send_error = None;
+        self.connection_error = None;
+        Ok(())
+    }
+
+    /// Send multiple non-final first frames for keys in a range.
+    pub fn send_first_frames_for_range(&mut self, start: u64, end: u64, body: &str) -> TestResult {
+        for key in start..=end {
+            // Ignore send errors — the connection may close mid-way through
+            // and we want to observe the connection error, not the send error.
+            if self.send_first_frame(key, body).is_err() {
+                break;
+            }
+            self.spin_runtime()?;
+        }
+        Ok(())
+    }
+
+    /// Send a non-final first frame for the provided message key.
+    pub fn send_first_frame(&mut self, key: u64, body: &str) -> TestResult {
+        let payload = test_helpers::first_frame_payload(key, body.as_bytes(), false, None)?;
+        self.send_payload(payload)
+    }
+
+    /// Send a final continuation frame for the provided message key.
+    pub fn send_final_continuation_frame(
+        &mut self,
+        key: u64,
+        sequence: u32,
+        body: &str,
+    ) -> TestResult {
+        let payload =
+            test_helpers::continuation_frame_payload(key, sequence, body.as_bytes(), true)?;
+        self.send_payload(payload)
+    }
+
+    /// Assert that the connection has terminated with an error and that no
+    /// payloads were delivered before the abort.
+    pub fn assert_connection_aborted(&mut self) -> TestResult {
+        self.spin_runtime()?;
+        self.drain_ready_payloads()?;
+        if !self.observed_payloads.is_empty() {
+            return Err(format!(
+                "expected no payloads before abort, but received: {:?}",
+                self.observed_payloads
+            )
+            .into());
+        }
+        let server = self.server.take().ok_or("server not initialized")?;
+        let result = self.block_on(server)?;
+        match result {
+            Ok(Ok(())) => Err("expected connection to abort, but it completed successfully".into()),
+            Ok(Err(error)) => {
+                self.connection_error = Some(error.to_string());
+                Ok(())
+            }
+            Err(join_error) => Err(format!("server task panicked: {join_error}").into()),
+        }
+    }
+
+    /// Assert that the expected payload is eventually observed.
+    pub fn assert_payload_received(&mut self, expected: &str) -> TestResult {
+        let expected = expected.as_bytes();
+        for _ in 0..SPIN_ATTEMPTS {
+            self.drain_ready_payloads()?;
+            if self
+                .observed_payloads
+                .iter()
+                .any(|payload| payload.as_slice() == expected)
+            {
+                return Ok(());
+            }
+            self.block_on(async { tokio::task::yield_now().await })?;
+        }
+
+        Err(format!(
+            "expected payload {:?} not observed; observed={:?}",
+            expected, self.observed_payloads
+        )
+        .into())
+    }
+
+    /// Assert that no connection error has been recorded.
+    pub fn assert_no_connection_error(&self) -> TestResult {
+        if self.connection_error.is_none() {
+            return Ok(());
+        }
+        Err(format!("unexpected connection error: {:?}", self.connection_error).into())
+    }
+
+    fn send_payload(&mut self, payload: Vec<u8>) -> TestResult {
+        let envelope = Envelope::new(ROUTE_ID, CORRELATION_ID, payload);
+        let serializer = BincodeSerializer;
+        let frame = serializer.serialize(&envelope)?;
+
+        let mut client = self.client.take().ok_or("client not initialized")?;
+        let send_result = self.block_on(async {
+            client.send(frame.into()).await?;
+            client.flush().await?;
+            Ok::<(), std::io::Error>(())
+        });
+        self.client = Some(client);
+
+        match send_result {
+            Ok(Ok(())) => {
+                self.last_send_error = None;
+                Ok(())
+            }
+            Ok(Err(error)) => {
+                self.last_send_error = Some(error.to_string());
+                Err(error.into())
+            }
+            Err(error) => {
+                self.last_send_error = Some(error.to_string());
+                Err(error)
+            }
+        }
+    }
+
+    fn spin_runtime(&self) -> TestResult {
+        self.block_on(async {
+            for _ in 0..8 {
+                tokio::task::yield_now().await;
+            }
+        })?;
+        Ok(())
+    }
+
+    fn drain_ready_payloads(&mut self) -> TestResult {
+        let mut observed_rx = self.observed_rx.take().ok_or("receiver not initialized")?;
+        while let Ok(payload) = observed_rx.try_recv() {
+            self.observed_payloads.push(payload);
+        }
+        self.observed_rx = Some(observed_rx);
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod hard_cap_config_tests {
+    use std::str::FromStr;
+
+    use rstest::rstest;
+
+    use super::HardCapConfig;
+
+    #[rstest]
+    fn valid_four_segment_input() {
+        let c = HardCapConfig::from_str("200/2048/8/8").expect("valid input");
+        assert_eq!(
+            (c.timeout_ms, c.per_message, c.per_connection, c.in_flight),
+            (200, 2048, 8, 8)
+        );
+    }
+
+    #[rstest]
+    #[case::missing_first("/64/100/100")]
+    #[case::missing_second("10//100/100")]
+    #[case::missing_third("10/64//100")]
+    #[case::missing_fourth("10/64/100/")]
+    #[case::only_three("10/64/100")]
+    #[case::extra_segment("10/64/100/100/extra")]
+    #[case::alphabetic_timeout("abc/64/100/100")]
+    #[case::alphabetic_budget("10/abc/100/100")]
+    #[case::negative_value("10/64/-1/100")]
+    fn invalid_inputs_are_rejected(#[case] input: &str) {
+        assert!(
+            HardCapConfig::from_str(input).is_err(),
+            "expected Err for {input:?}"
+        );
+    }
+
+    #[rstest]
+    fn zero_timeout_is_accepted() {
+        let c = HardCapConfig::from_str("0/64/100/100").expect("zero timeout is valid");
+        assert_eq!(c.timeout_ms, 0);
+    }
+
+    // Zero budgets parse as valid `usize` values. The non-zero constraint is
+    // enforced later by `start_app()` via `NonZeroUsize::new(...)`.
+    #[rstest]
+    fn zero_budget_parses_successfully() {
+        let c = HardCapConfig::from_str("10/0/100/100").expect("zero budget parses");
+        assert_eq!(c.per_message, 0);
+    }
+}
diff --git a/tests/fixtures/mod.rs b/tests/fixtures/mod.rs
index c55ae4ad..101b8a1e 100644
--- a/tests/fixtures/mod.rs
+++ b/tests/fixtures/mod.rs
@@ -18,6 +18,7 @@ pub mod fragment;
 pub mod interleaved_push_queues;
 pub mod memory_budget_backpressure;
 mod memory_budget_backpressure_tests;
+pub mod memory_budget_hard_cap;
 pub mod memory_budgets;
 pub mod message_assembler;
 pub mod message_assembly;
diff --git a/tests/scenarios/memory_budget_hard_cap_scenarios.rs b/tests/scenarios/memory_budget_hard_cap_scenarios.rs
new file mode 100644
index 00000000..b3baeecf
--- /dev/null
+++ b/tests/scenarios/memory_budget_hard_cap_scenarios.rs
@@ -0,0 +1,23 @@
+//! Scenario test functions for hard-cap memory budget connection abort.
+
+use rstest_bdd_macros::scenario;
+
+use crate::fixtures::memory_budget_hard_cap::*;
+
+#[scenario(
+    path = "tests/features/memory_budget_hard_cap.feature",
+    name = "Connection terminates after repeated budget violations"
+)]
+fn connection_terminates_after_budget_violations(
+    memory_budget_hard_cap_world: MemoryBudgetHardCapWorld,
+) {
+    drop(memory_budget_hard_cap_world);
+}
+
+#[scenario(
+    path = "tests/features/memory_budget_hard_cap.feature",
+    name = "Connection survives when frames stay within budget"
+)]
+fn connection_survives_within_budget(memory_budget_hard_cap_world: MemoryBudgetHardCapWorld) {
+    drop(memory_budget_hard_cap_world);
+}
diff --git a/tests/scenarios/mod.rs b/tests/scenarios/mod.rs
index d7ea76c9..18c8bbfa 100644
--- a/tests/scenarios/mod.rs
+++ b/tests/scenarios/mod.rs
@@ -21,6 +21,7 @@ mod correlation_scenarios;
 mod fragment_scenarios;
 mod interleaved_push_queues_scenarios;
 mod memory_budget_backpressure_scenarios;
+mod memory_budget_hard_cap_scenarios;
 mod memory_budgets_scenarios;
 mod message_assembler_scenarios;
 mod message_assembly_inbound_scenarios;
diff --git a/tests/steps/memory_budget_hard_cap_steps.rs b/tests/steps/memory_budget_hard_cap_steps.rs
new file mode 100644
index 00000000..e108291f
--- /dev/null
+++ b/tests/steps/memory_budget_hard_cap_steps.rs
@@ -0,0 +1,73 @@
+//! Step definitions for hard-cap memory budget connection abort scenarios.
+
+use rstest_bdd_macros::{given, then, when};
+
+use crate::fixtures::memory_budget_hard_cap::{
+    HardCapConfig,
+    MemoryBudgetHardCapWorld,
+    TestResult,
+};
+
+#[given("a hard-cap inbound app configured as {config}")]
+fn given_hard_cap_app(
+    memory_budget_hard_cap_world: &mut MemoryBudgetHardCapWorld,
+    config: HardCapConfig,
+) -> TestResult {
+    memory_budget_hard_cap_world.start_app(config)
+}
+
+#[when(
+    "hard-cap first frames for keys {start:u64} to {end:u64} each with body {body:string} arrive"
+)]
+fn when_first_frames_for_range(
+    memory_budget_hard_cap_world: &mut MemoryBudgetHardCapWorld,
+    start: u64,
+    end: u64,
+    body: String,
+) -> TestResult {
+    memory_budget_hard_cap_world.send_first_frames_for_range(start, end, &body)
+}
+
+#[when("a hard-cap first frame for key {key:u64} with body {body:string} arrives")]
+fn when_hard_cap_first_frame(
+    memory_budget_hard_cap_world: &mut MemoryBudgetHardCapWorld,
+    key: u64,
+    body: String,
+) -> TestResult {
+    memory_budget_hard_cap_world.send_first_frame(key, &body)
+}
+
+#[when(
+    "a hard-cap final continuation for key {key:u64} sequence {sequence:u32} with body \
+     {body:string} arrives"
+)]
+fn when_hard_cap_final_continuation(
+    memory_budget_hard_cap_world: &mut MemoryBudgetHardCapWorld,
+    key: u64,
+    sequence: u32,
+    body: String,
+) -> TestResult {
+    memory_budget_hard_cap_world.send_final_continuation_frame(key, sequence, &body)
+}
+
+#[then("the hard-cap connection terminates with an error")]
+fn then_connection_terminates(
+    memory_budget_hard_cap_world: &mut MemoryBudgetHardCapWorld,
+) -> TestResult {
+    memory_budget_hard_cap_world.assert_connection_aborted()
+}
+
+#[then("hard-cap payload {expected:string} is eventually received")]
+fn then_hard_cap_payload_received(
+    memory_budget_hard_cap_world: &mut MemoryBudgetHardCapWorld,
+    expected: String,
+) -> TestResult {
+    memory_budget_hard_cap_world.assert_payload_received(&expected)
+}
+
+#[then("no hard-cap connection error is recorded")]
+fn then_no_hard_cap_connection_error(
+    memory_budget_hard_cap_world: &mut MemoryBudgetHardCapWorld,
+) -> TestResult {
+    memory_budget_hard_cap_world.assert_no_connection_error()
+}
diff --git a/tests/steps/mod.rs b/tests/steps/mod.rs
index 57efc70d..4acbc725 100644
--- a/tests/steps/mod.rs
+++ b/tests/steps/mod.rs
@@ -17,6 +17,7 @@ mod correlation_steps;
 mod fragment_steps;
 mod interleaved_push_queues_steps;
 mod memory_budget_backpressure_steps;
+mod memory_budget_hard_cap_steps;
 mod memory_budgets_steps;
 mod message_assembler_steps;
 mod message_assembly_inbound_steps;