Integrate inbound MessageAssembler; add assembly helper and routing

[leynos]

Summary

• Integrate inbound MessageAssembler into the connection path and add a dedicated assembly helper to drive protocol-level message assembly.
• Extend per-connection state to buffer in-flight assemblies and purge stale ones; preserve no-assembler pass-through when not configured.
• Introduce EnvelopeRouting and propagate routing data from the first frame through to the assembled message; extend FirstFrameInput and AssembledMessage accordingly.
• Refactor test harness for continuation frames to use a params object; update tests and fixtures to exercise the new assembly flow.
• Upgrade test scaffolding (rstest-bdd) and align documentation and ADRs accordingly.
• All existing public API surfaces remain stable when no assembler is configured.

Changes

Core Functionality

• Add inbound assembly helper in src/app/frame_handling/assembly.rs that:
  • consumes complete envelopes, invokes the configured MessageAssembler, and updates MessageAssemblyState;
  • enforces strict payload boundaries and frame progression (first frame vs. continuation frames);
  • returns a completed Envelope when assembly is ready for dispatch.
• Export the new helper from src/app/frame_handling/mod.rs.
• Extend inbound connection path in src/app/connection.rs to:
  • hold optional inbound message assembly state per connection;
  • invoke the assembly helper after reassemble_if_needed and before route lookup;
  • purge expired assemblies during read-timeout handling;
  • retain a no-assembler pass-through path when not configured.
• Refactor test harness for continuation frames:
  • Introduce ContinuationFrameParams and refactor send_continuation_frame_impl to accept a params object, simplifying frame construction and enabling clearer tests.
  • Update test fixtures to use the params-based approach for continuation frames.
• Update public-facing messaging types to carry envelope routing data from the first frame:
  • Introduce EnvelopeRouting to carry per-envelope routing metadata from the first frame.
  • Extend FirstFrameInput to include routing data; AssembledMessage now carries routing information for correct dispatch.
  • Exported and used types updated in src/message_assembler/* to propagate routing through assembly.

API Surface

• Maintain public API stability; no new external dependencies beyond the planned rstest-bdd upgrade.
• Update routing propagation through assembly:
  • Add EnvelopeRouting to routing data carried from first frame.
  • Extend FirstFrameInput with routing; AssembledMessage carries routing.
  • Ensure completed envelopes preserve first-frame routing metadata for correct dispatch.

Testing

• Add unit tests for the new assembly helper (interleaving, ordering, timeout purge scenarios).
• Update tests that cover inbound flow to exercise the assembly integration points.
• Upgrade behavioural test dependencies to rstest-bdd v0.5.0 and adjust scaffolding.
• Test harness changes cover continuation-frame construction via ContinuationFrameParams.

Documentation & Design

• Add ExecPlan doc entry for 8.2.5/8.2.6 (present on branch).
• Update ADR/design docs to reflect inbound integration decisions and routing propagation.
• Update docs/users-guide.md and docs/roadmap.md to reflect the new inbound behaviour and mark roadmap items 8.2.5/8.2.6 as done after gates pass.

Validation and acceptance

• Inbound runtime applies message assembly after transport fragmentation and before handler routing.
• Interleaved keyed assemblies are collected and dispatched only when complete.
• Ordering violations are rejected deterministically per existing deserialization policy.
• Stale partial assemblies are purged on idle/read-timeout events.
• Unit tests cover interleaving, ordering, and timeout purge; behavioural tests cover end-to-end inbound flow using rstest-bdd v0.5.0.
• Docs, user guide, and roadmap are updated and aligned.

Plan of work (high level)

• Stage A: upgrade test crates and ensure clean build.
• Stage B: implement inbound assembly helper and related state management.
• Stage C: wire into src/app/connection.rs and implement purging behavior.
• Stage D: add unit and behavioural tests.
• Stage E: update documentation and mark roadmap items as done after gates pass.

Artifacts

• New module: src/app/frame_handling/assembly.rs (exported via src/app/frame_handling/mod.rs).
• Updated src/app/connection.rs to support inbound assembly state and invocation.
• Updated/added unit and behavioural tests, and updated design/docs references as described above.
• ExecPlan doc under docs/execplans/8-2-5-integrate-with-the-connection-actor-inbound-path.md (present on branch).

Notes

• The PR preserves the existing no-assembler pass-through path and only activates assembly logic when an assembler is configured. This aligns with ADR expectations and existing runtime behavior.

📎 Task: https://www.devboxer.com/task/4c43c2fe-e65f-4e59-a3a2-dc176706eff0

[sourcery-ai]

Reviewer's Guide

Integrates protocol-level MessageAssembler into the inbound connection path via a new frame-handling assembly helper, centralizes envelope decoding, wires per-connection assembly state and purge logic into WireframeApp, and adds focused unit/BDD tests plus documentation and roadmap updates for the new behaviour.

Sequence diagram for inbound frame processing with message assembly

sequenceDiagram
  actor Client
  participant Connection as WireframeApp_connection
  participant FrameHandling as frame_handling
  participant Assembler as MessageAssembler
  participant AssemblyState as MessageAssemblyState
  participant Handler as HandlerService

  Client->>Connection: receive_frame()
  loop per_frame
    Connection->>FrameHandling: decode_envelope(frame)
    FrameHandling-->>Connection: Option Envelope

    alt envelope_decoded
      Connection->>FrameHandling: reassemble_if_needed(env, fragmentation)
      FrameHandling-->>Connection: Option Envelope

      alt post_fragment_envelope
        Connection->>FrameHandling: assemble_if_needed(AssemblyRuntime, deser_failures, env, MAX_DESER_FAILURES)
        activate FrameHandling
        FrameHandling->>Assembler: parse_frame_header(env.payload_bytes())
        Assembler-->>FrameHandling: ParsedFrameHeader
        FrameHandling->>AssemblyState: accept_first_frame_or_continuation(...)
        AssemblyState-->>FrameHandling: Option AssembledMessage
        deactivate FrameHandling

        alt assembly_complete
          FrameHandling-->>Connection: Some Envelope
          Connection->>Handler: forward_response(env)
          Handler-->>Connection: handle_result
        else assembly_in_progress
          FrameHandling-->>Connection: None
        end
      else no_post_fragment_envelope
        FrameHandling-->>Connection: None
      end
    else decode_failed_or_skipped
      FrameHandling-->>Connection: None
    end
  end

Loading

Sequence diagram for read-timeout purging of fragmentation and message assemblies

sequenceDiagram
  actor Client
  participant Connection as WireframeApp_connection
  participant FrameHandling as frame_handling
  participant FragState as FragmentationState
  participant AssemblyState as MessageAssemblyState

  Client->>Connection: idle
  Connection->>Connection: wait_for_frame_with_timeout()
  Connection-->>Connection: Err timeout
  Connection->>Connection: purge_expired(fragmentation, message_assembly)
  alt fragmentation_state_present
    Connection->>FragState: purge_expired()
    FragState-->>Connection: evicted_fragment_keys
  end
  Connection->>FrameHandling: purge_expired_assemblies(message_assembly)
  FrameHandling->>AssemblyState: purge_expired()
  AssemblyState-->>FrameHandling: evicted_message_keys
  FrameHandling-->>Connection: return
  Connection-->>Client: continue_waiting_for_next_frame

Loading

Updated class diagram for inbound message assembly helpers and related types

classDiagram
  class AssemblyRuntime {
    +Option~&Arc_MessageAssembler~~ assembler
    +Option~MessageAssemblyState~* state
    +new(Option~&Arc_MessageAssembler~~ assembler, Option~MessageAssemblyState~* state) AssemblyRuntime
  }

  class MessageAssemblyState {
    +new(NonZeroUsize max_message_size, Duration timeout) MessageAssemblyState
    +purge_expired() Vec~AssemblyKey~
    +accept_first_frame(FirstFrameInput input) Result~Option~AssembledMessage~, AssemblyError~
    +accept_continuation_frame(ContinuationFrameHeader header, &[u8] frame_bytes) Result~Option~AssembledMessage~, AssemblyError~
  }

  class MessageAssembler {
    <<trait>>
    +parse_frame_header(&[u8] payload) Result~ParsedFrameHeader, AssemblyError~
  }

  class Envelope {
    +u32 id
    +Option~u64~ correlation_id
    +Vec~u8~ payload
    +new(u32 id, Option~u64~ correlation_id, Vec~u8~ payload) Envelope
    +payload_bytes() &[u8]
  }

  class DeserFailureTracker {
    +new(u32* failures, u32 max_failures) DeserFailureTracker
    +record(Option~u64~ correlation_id, &str context, io::Error err) io::Result~()~
  }

  class FirstFrameHeader {
    +usize metadata_len
    +usize body_len
  }

  class ContinuationFrameHeader {
    +usize body_len
  }

  class FirstFrameInput {
    +new(FirstFrameHeader header, Vec~u8~ metadata, &[u8] body) Result~FirstFrameInput, AssemblyError~
  }

  class AssembledMessage {
    +metadata() &[u8]
    +body() &[u8]
  }

  class FrameHeader {
    <<enum>>
    First(FirstFrameHeader)
    Continuation(ContinuationFrameHeader)
  }

  class AssemblyContext {
    -MessageAssemblyState* state
    -DeserFailureTracker* failures
    -Option~u64~ correlation_id
  }

  %% Free helper functions in assembly.rs represented as a utility class
  class AssemblyHelpers {
    +new_message_assembly_state(Option~FragmentationConfig~ fragmentation, usize frame_budget) MessageAssemblyState
    +purge_expired_assemblies(Option~MessageAssemblyState~* assembly) void
    +assemble_if_needed(AssemblyRuntime runtime, u32* deser_failures, Envelope env, u32 max_deser_failures) io::Result~Option~Envelope~~
    +process_first_frame(AssemblyContext* context, FirstFrameHeader* header, &[u8] frame_bytes) io::Result~Option~AssembledMessage~~
    +process_continuation_frame(AssemblyContext* context, ContinuationFrameHeader* header, &[u8] frame_bytes) io::Result~Option~AssembledMessage~~
    +envelope_from_assembled(u32 id, Option~u64~ correlation_id, AssembledMessage* assembled) Envelope
  }

  class FragmentationConfig {
    +NonZeroUsize max_message_size
    +Duration reassembly_timeout
  }

  AssemblyRuntime --> MessageAssembler : uses
  AssemblyRuntime --> MessageAssemblyState : holds_mutably

  AssemblyHelpers ..> AssemblyRuntime : constructs
  AssemblyHelpers ..> MessageAssemblyState : creates_and_purges
  AssemblyHelpers ..> DeserFailureTracker : uses
  AssemblyHelpers ..> Envelope : creates
  AssemblyHelpers ..> FirstFrameHeader : reads
  AssemblyHelpers ..> ContinuationFrameHeader : reads
  AssemblyHelpers ..> AssembledMessage : consumes
  AssemblyHelpers ..> FragmentationConfig : reads

  DeserFailureTracker --> Envelope : correlates_with

  MessageAssembler --> FrameHeader : returns
  FrameHeader --> FirstFrameHeader : variant
  FrameHeader --> ContinuationFrameHeader : variant

  MessageAssemblyState --> FirstFrameInput : consumes
  MessageAssemblyState --> ContinuationFrameHeader : reads
  MessageAssemblyState --> AssembledMessage : returns

  AssemblyContext --> MessageAssemblyState : holds_mutably
  AssemblyContext --> DeserFailureTracker : holds_mutably
  AssemblyContext --> Envelope : correlates_via_id

  AssembledMessage --> Envelope : converted_into_via_AssemblyHelpers

  FragmentationConfig --> MessageAssemblyState : configures_new_state

Loading

File-Level Changes

Change	Details	Files
Add inbound assembly helper and runtime integration for protocol-level message assembly.	Introduce src/app/frame_handling/assembly.rs with AssemblyRuntime, assemble_if_needed, new_message_assembly_state, and purge_expired_assemblies to manage message assembly using MessageAssembler and MessageAssemblyState. Use fragmentation configuration and frame budget to derive per-connection assembly limits and timeouts, with a default timeout when fragmentation is absent. Implement strict header/payload validation for first and continuation frames, mapping all parse/continuity issues into deserialization failures via DeserFailureTracker, and reconstruct completed Envelopes from assembled metadata/body.	src/app/frame_handling/assembly.rs
Wire message assembly into the inbound connection loop and share decode/assembly helpers.	Extend WireframeApp inbound connection context to carry optional MessageAssemblyState, initialize it when a MessageAssembler is configured using new_message_assembly_state, and purge expired assemblies alongside fragmentation on read timeouts. Replace the previous decode_envelope method on WireframeApp with a shared frame_handling::decode_envelope helper, updating inbound processing and existing tests to use it while preserving failure tracking semantics. Insert assemble_if_needed after reassemble_if_needed in inbound frame handling, preserving pass-through when no assembler or assembly state is present and forwarding only fully assembled Envelopes to routing.	src/app/connection.rs src/app/connection/tests.rs src/app/frame_handling/mod.rs src/app/frame_handling/decode.rs
Add targeted unit tests and BDD scenarios for inbound assembly semantics.	Add src/app/frame_handling/assembly_tests.rs to cover interleaved keyed assembly, ordering violations, and timeout-purge behaviour using TestAssembler and MessageAssemblyState. Introduce MessageAssemblyInboundWorld fixture plus associated steps and scenarios to drive end-to-end inbound assembly via WireframeApp::handle_connection_result with a duplex codec. Add new rstest-bdd-based feature file exercising interleaving, ordering recovery, and timeout eviction, and register the new fixtures, steps, and scenarios in the tests module tree.	src/app/frame_handling/assembly_tests.rs tests/fixtures/message_assembly_inbound.rs tests/steps/message_assembly_inbound_steps.rs tests/scenarios/message_assembly_inbound_scenarios.rs tests/features/message_assembly_inbound.feature tests/fixtures/mod.rs tests/steps/mod.rs tests/scenarios/mod.rs
Upgrade BDD test dependencies and update documentation and roadmap for inbound integration.	Bump rstest-bdd and rstest-bdd-macros dev-dependencies to 0.5.0 and refresh Cargo.lock to match. Document inbound integration decisions, failure-handling policy, and runtime ordering in ADR 002, the generic fragmentation/assembly design doc, the execplan for 8.2.5/8.2.6, and the users guide. Mark roadmap items 8.2.5 and 8.2.6 complete and ensure design/docs reference the new inbound behaviour and testing approach.	Cargo.toml Cargo.lock docs/adr-002-streaming-requests-and-shared-message-assembly.md docs/generic-message-fragmentation-and-re-assembly-design.md docs/users-guide.md docs/roadmap.md docs/execplans/8-2-5-integrate-with-the-connection-actor-inbound-path.md

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

Integrate inbound message assembly into the connection path: add frame_handling decode and assembly helpers, wire MessageAssemblyState and MessageAssembler into connection runtime, propagate EnvelopeRouting through assembly types, centralise decode/assemble logic, add unit and BDD integration tests, and update docs and roadmap entries.

Changes

Cohort / File(s)	Summary
Dependencies Cargo.toml	Bumped dev-dependencies: rstest-bdd and rstest-bdd-macros 0.4.0 → 0.5.0.
Documentation docs/adr-002-streaming-requests-and-shared-message-assembly.md, docs/execplans/...8-2-5-integrate-with-the-connection-actor-inbound-path.md, docs/generic-message-fragmentation-and-re-assembly-design.md, docs/roadmap.md, docs/users-guide.md	Add execution plan and implementation decisions; mark 8.2.5/8.2.6 complete; document assembler hook placement (app instance), runtime ordering (after transport reassembly, before dispatch), timeout/purge and deserialization-failure policy; update user guide and examples.
Frame handling module src/app/frame_handling/... src/app/frame_handling/assembly.rs, src/app/frame_handling/decode.rs, src/app/frame_handling/mod.rs, src/app/frame_handling/assembly_tests.rs	Add AssemblyRuntime, new_message_assembly_state, purge_expired_assemblies, assemble_if_needed and Envelope::from_assembled; add decode_envelope helper that tracks deser failures and correlation IDs; re-export APIs; add focused unit tests for interleaving, ordering violations and timeout purge.
Connection layer src/app/connection.rs, src/app/connection/tests.rs	Wire optional MessageAssemblyState into process_stream; call purge_expired with fragmentation and assembly state; replace inline decode/assemble logic with frame_handling::decode_envelope and assemble_if_needed; reset deser-failure counter only after full inbound pipeline succeeds; update tests to call decode_envelope.
Message assembler API & state src/message_assembler/types.rs, src/message_assembler/state.rs, src/message_assembler/mod.rs, src/message_assembler/state_tests.rs, src/message_assembler/types.rs	Introduce public EnvelopeRouting (envelope_id, correlation_id); add routing field to FirstFrameInput, PartialAssembly and AssembledMessage; change FirstFrameInput::new and AssembledMessage::new signatures to accept EnvelopeRouting; update tests and re-export routing.
Test helpers & fixtures src/test_helpers.rs, tests/fixtures/*, tests/features/*, tests/scenarios/*, tests/steps/*, tests/fixtures/mod.rs, tests/steps/mod.rs	Add first_frame_payload and continuation_frame_payload; add BDD feature, step definitions, scenarios and a runtime-backed MessageAssemblyInboundWorld fixture; add scenario glue and fixtures to test modules; adjust tests to pass EnvelopeRouting::default() where required.
Test scaffolding src/app/frame_handling/assembly_tests.rs, tests/*	Add comprehensive unit tests and BDD scenarios covering interleaving, ordering violations, timeout purge and pass-through when assembler/state absent.

Sequence Diagram(s)

sequenceDiagram
    participant Client as Inbound Stream
    participant Conn as Connection Handler
    participant Decode as Frame Handling (decode)
    participant Assemble as Frame Handling (assembly)
    participant Dispatch as Handler Dispatch

    Client->>Conn: deliver frame
    Conn->>Decode: decode_envelope(parse_result, frame, &mut deser_failures, MAX_DESER_FAILURES)
    Decode->>Decode: parse envelope, derive correlation_id, track failures
    Decode-->>Conn: Ok(Some(Envelope)) / Ok(None) / Err(close)

    Conn->>Assemble: assemble_if_needed(AssemblyRuntime{assembler?, state?}, &mut deser_failures, env, MAX_DESER_FAILURES)
    alt assembler configured and frame is fragmented
        Assemble->>Assemble: parse headers, validate lengths
        alt first frame
            Assemble->>Assemble: create PartialAssembly (store routing)
            Assemble-->>Conn: Ok(None)
        else continuation
            Assemble->>Assemble: append to partial, check ordering
            alt complete
                Assemble->>Assemble: build AssembledMessage (preserve routing)
                Assemble-->>Conn: Ok(Some(Envelope))
            else awaiting more
                Assemble-->>Conn: Ok(None)
            end
        end
    else no assembler or non-fragmented
        Assemble-->>Conn: Ok(Some(Envelope)) (pass-through)
    end

    Conn->>Dispatch: dispatch assembled envelope to handler
    Dispatch->>Dispatch: invoke handler

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~50 minutes

Poem

✨ Frames mend like patchwork, piece by piece,
Keys hum, routing threads the final lease,
Decode, stitch, purge time's forgotten art,
Tests applaud each reassembled part,
Docs and roadmap crown the new start.

🚥 Pre-merge checks | ✅ 3 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Merge Conflict Detection	⚠️ Warning	❌ Merge conflicts detected (64 files): ⚔️ Cargo.lock (content) ⚔️ docs/adr-002-streaming-requests-and-shared-message-assembly.md (content) ⚔️ docs/adr-004-pluggable-protocol-codecs.md (content) ⚔️ docs/contents.md (content) ⚔️ docs/generic-message-fragmentation-and-re-assembly-design.md (content) ⚔️ docs/hardening-wireframe-a-guide-to-production-resilience.md (content) ⚔️ docs/multi-packet-and-streaming-responses-design.md (content) ⚔️ docs/roadmap.md (content) ⚔️ docs/rstest-bdd-users-guide.md (content) ⚔️ docs/the-road-to-wireframe-1-0-feature-set-philosophy-and-capability-maturity.md (content) ⚔️ docs/users-guide.md (content) ⚔️ docs/wireframe-client-design.md (content) ⚔️ src/app/builder/codec.rs (content) ⚔️ src/app/builder/config.rs (content) ⚔️ src/app/builder/core.rs (content) ⚔️ src/app/builder/state.rs (content) ⚔️ src/app/connection.rs (content) ⚔️ src/app/connection/tests.rs (content) ⚔️ src/app/fragmentation_state.rs (content) ⚔️ src/app/frame_handling/mod.rs (content) ⚔️ src/byte_order.rs (content) ⚔️ src/client/builder/mod.rs (content) ⚔️ src/client/error.rs (content) ⚔️ src/client/messaging.rs (content) ⚔️ src/client/mod.rs (content) ⚔️ src/client/runtime.rs (content) ⚔️ src/client/tests/error_handling.rs (content) ⚔️ src/client/tests/messaging.rs (content) ⚔️ src/codec.rs (content) ⚔️ src/codec/error.rs (content) ⚔️ src/codec/error_tests.rs (content) ⚔️ src/extractor/request.rs (content) ⚔️ src/fragment/error.rs (content) ⚔️ src/fragment/mod.rs (content) ⚔️ src/fragment/reassembler.rs (content) ⚔️ src/fragment/series.rs (content) ⚔️ src/fragment/tests.rs (content) ⚔️ src/frame/conversion.rs (content) ⚔️ src/lib.rs (content) ⚔️ src/message_assembler/mod.rs (content) ⚔️ src/message_assembler/state.rs (content) ⚔️ src/message_assembler/state_tests.rs (content) ⚔️ src/message_assembler/types.rs (content) ⚔️ src/test_helpers.rs (content) ⚔️ tests/app_data.rs (content) ⚔️ tests/client_runtime.rs (content) ⚔️ tests/extractor.rs (content) ⚔️ tests/features/client_lifecycle.feature (content) ⚔️ tests/features/client_runtime.feature (content) ⚔️ tests/features/fragment.feature (content) ⚔️ tests/fixtures/client_runtime.rs (content) ⚔️ tests/fixtures/codec_error/decoder_ops.rs (content) ⚔️ tests/fixtures/message_assembly.rs (content) ⚔️ tests/fixtures/mod.rs (content) ⚔️ tests/fragment_transport.rs (content) ⚔️ tests/fragment_transport/mod.rs (content) ⚔️ tests/fragment_transport/rejection.rs (content) ⚔️ tests/scenarios/client_runtime_scenarios.rs (content) ⚔️ tests/scenarios/fragment_scenarios.rs (content) ⚔️ tests/scenarios/mod.rs (content) ⚔️ tests/steps/client_lifecycle_steps.rs (content) ⚔️ tests/steps/client_runtime_steps.rs (content) ⚔️ tests/steps/fragment_steps.rs (content) ⚔️ tests/steps/mod.rs (content) These conflicts must be resolved before merging into main.	Resolve conflicts locally and push changes to this branch.

✅ Passed checks (3 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Title check	✅ Passed	The title directly summarises the main change: integrating inbound MessageAssembler and adding assembly helper with routing propagation.
Description check	✅ Passed	The description comprehensively covers the changeset: inbound assembly integration, state management, routing propagation, test refactoring, and documentation updates.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch integrate-messageassembler-inbound-kkt43c

⚔️ Resolve merge conflicts (beta)

• Auto-commit resolved conflicts to branch integrate-messageassembler-inbound-kkt43c
• Create stacked PR with resolved conflicts
• Post resolved changes as copyable diffs in a comment

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[leynos]

@coderabbitai please create a new GitHub issue for the following:

Connection module naming is ambiguous: src/connection/ is outbound-focused, while inbound routing lives in
src/app/connection.rs. This can lead to confusion in requirements, conversation and implementation planning.

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

@leynos: I'll create a GitHub issue to track this naming ambiguity concern.

---

🏁 Script executed:

#!/bin/bash
# Create GitHub issue for connection module naming ambiguity
gh issue create \
  --title "Ambiguous connection module naming causes confusion" \
  --body "## Problem

The current organisation of connection-related code creates ambiguity:

- \`src/connection/\` contains outbound-focused connection logic
- \`src/app/connection.rs\` contains inbound routing and connection handling

This naming overlap can lead to confusion during requirements gathering, technical discussions, and implementation planning, as it's not immediately clear which module handles which direction of traffic flow.

## Context

Raised during PR review: https://github.com/leynos/wireframe/pull/458

Reported by: `@leynos`

## Recommendation

Consider renaming one or both modules to make the directionality explicit, for example:
- \`src/connection/outbound/\` or \`src/outbound_connection/\`
- \`src/app/inbound_connection.rs\` or similar

This would make the separation of concerns clearer and reduce cognitive load when navigating the codebase." \
  --assignee leynos

Length of output: 46

---

@leynos: I've created GitHub issue #460 to track the connection module naming ambiguity: #460

The issue documents the confusion between src/connection/ (outbound-focused) and src/app/connection.rs (inbound routing), and suggests potential renaming approaches to make the directionality explicit.

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix:

tests/fixtures/message_assembly_inbound.rs

Comment on lines +142 to +149

    pub fn send_continuation_frame(&mut self, key: u64, sequence: u32, body: &str) -> TestResult {
        self.send_payload(continuation_frame_payload(
            key,
            sequence,
            body.as_bytes(),
            false,
        ))
    }

❌ New issue: Code Duplication
The module contains 2 functions with similar structure: MessageAssemblyInboundWorld.send_continuation_frame,MessageAssemblyInboundWorld.send_final_continuation_frame

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix:

Excess Number of Function Arguments

tests/fixtures/message_assembly_inbound.rs: MessageAssemblyInboundWorld.send_continuation_frame_impl

What lead to degradation?

MessageAssemblyInboundWorld.send_continuation_frame_impl has 5 arguments, max arguments = 4

Why does this problem occur?

Functions with many arguments indicate either a) low cohesion where the function has too many responsibilities, or b) a missing abstraction that encapsulates those arguments.
The threshold for the Rust language is 4 function arguments.

How to fix it?

Start by investigating the responsibilities of the function. Make sure it doesn't do too many things, in which case it should be split into smaller and more cohesive functions. Consider the refactoring INTRODUCE PARAMETER OBJECT to encapsulate arguments that refer to the same logical concept.

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix:

tests/fixtures/message_assembly_inbound.rs

Comment on file

//! `MessageAssemblyInboundWorld` fixture for inbound assembly integration.

❌ New issue: Primitive Obsession
In this module, 30.4% of all function arguments are primitive types, threshold = 30.0%

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix:

Code Duplication

src/app/frame_handling/assembly_tests.rs:

What lead to degradation?

The module contains 2 functions with similar structure: inbound_assembly_handles_interleaved_sequences,inbound_assembly_rejects_ordering_violations

Why does this problem occur?

Duplicated code often leads to code that's harder to change since the same logical change has to be done in multiple functions. More duplication gives lower code health.

How to fix it?

A certain degree of duplicated code might be acceptable. The problems start when it is the same behavior that is duplicated across the functions in the module, ie. a violation of the Don't Repeat Yourself (DRY) principle. DRY violations lead to code that is changed together in predictable patterns, which is both expensive and risky. DRY violations can be identified using CodeScene's X-Ray analysis to detect clusters of change coupled functions with high code similarity. Read More
Once you have identified the similarities across functions, look to extract and encapsulate the concept that varies into its own function(s). These shared abstractions can then be re-used, which minimizes the amount of duplication and simplifies change.

[coderabbitai]

Actionable comments posted: 1

🤖 Fix all issues with AI agents

In `@src/app/frame_handling/assembly_tests.rs`:
- Around line 24-30: The test fixture message_assembly_state uses
NonZeroUsize::new(1024).unwrap_or(...); replace the redundant unwrap_or fallback
with .expect("1024 is non-zero") to make the invariant explicit and provide a
clear failure message; update the call inside MessageAssemblyState::new in the
fixture (function message_assembly_state) accordingly.

[coderabbitai]

Actionable comments posted: 1

🤖 Fix all issues with AI agents

In `@tests/fixtures/message_assembly_inbound.rs`:
- Around line 239-264: Extract the hard-coded millisecond timeouts used in
assert_received_payload and assert_received_count into clearly named constants
(e.g., PAYLOAD_COLLECT_TIMEOUT_MS and COUNT_COLLECT_TIMEOUT_MS or a single
OBSERVED_COLLECT_TIMEOUT_MS if same) and replace the literal
Duration::from_millis(200) and Duration::from_millis(120) calls in those methods
with Duration::from_millis(CONSTANT). Update the constants' placement near the
top of the file or module so both functions (assert_received_payload and
assert_received_count) can reference them and ensure the intent and units
(milliseconds) are clear in the constant names.

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

@leynos: I'll verify the current resolution status by checking all the relevant files.

---

🏁 Script executed:

#!/bin/bash

echo "=== 1. Checking for centralized test helpers in src/test_helpers.rs ==="
echo "Looking for first_frame_payload and continuation_frame_payload functions:"
rg -n "pub fn (first_frame_payload|continuation_frame_payload)" src/test_helpers.rs

echo -e "\n=== 2. Checking assembly_tests.rs for pass-through tests ==="
echo "Looking for pass-through test functions:"
rg -n "fn assemble_if_needed_passes_through_when_(assembler|state)_is_none" src/app/frame_handling/assembly_tests.rs

echo -e "\n=== 3. Checking assembly.rs for fail_invalid_none helper ==="
echo "Looking for fail_invalid_none method in AssemblyContext:"
rg -n -A 3 "fn fail_invalid_none" src/app/frame_handling/assembly.rs

echo -e "\n=== 4. Checking assembly.rs for Envelope::from_assembled ==="
echo "Looking for from_assembled in Envelope impl:"
rg -n "impl Envelope" src/app/frame_handling/assembly.rs -A 15 | head -20

echo -e "\n=== 5. Checking docs/users-guide.md for 2nd-person pronoun fix ==="
echo "Looking for 'if direct access is required' or 'if you need':"
rg -n "(if direct access is required|if you need to access it directly)" docs/users-guide.md

echo -e "\n=== 6. Checking execplan for LOC expansion ==="
echo "Looking for 'lines of code (LOC)' or standalone 'LOC':"
rg -n "(lines of code \(LOC\)|more than 900 net)" docs/execplans/8-2-5-integrate-with-the-connection-actor-inbound-path.md

echo -e "\n=== 7. Checking wait_millis implementation in message_assembly_inbound.rs ==="
echo "Looking for wait_millis method implementation:"
rg -n -A 5 "pub fn wait_millis" tests/fixtures/message_assembly_inbound.rs

echo -e "\n=== 8. Checking if tests use synthetic clock ==="
echo "Looking for tokio::time::advance or pause usage in tests:"
rg -n "(time::advance|time::pause|Handle::current)" tests/fixtures/message_assembly_inbound.rs tests/steps/message_assembly_inbound_steps.rs

Length of output: 2829

---

@leynos: Yes, all review comments have now been fully resolved! Here's the updated status:

✅ All Resolved

1. Frame payload construction duplication (Overall Comment 1):

   • Helpers first_frame_payload and continuation_frame_payload are now centralised in src/test_helpers.rs (lines 123, 146)
   • Both test modules import and use these shared helpers

2. Pass-through tests (Comment 1):

   • assemble_if_needed_passes_through_when_assembler_is_none (line 258)
   • assemble_if_needed_passes_through_when_state_is_none (line 279)
   • Both tests added to src/app/frame_handling/assembly_tests.rs

3. Deduplicate failure recording (Comment 2.2):

   • AssemblyContext::fail_invalid_none helper method added (line 157 in assembly.rs)
   • Eliminates the repetitive failures.record(...); return Ok(None) pattern

4. Envelope construction (Comment 2.3):

   • Envelope::from_assembled moved to an impl block on Envelope (lines 253-268)
   • Implementation improved: extracts routing from AssembledMessage rather than requiring separate parameters

5. AssemblyContext simplification (Comment 2.1):

   • Partially resolved by adding the fail_invalid_none helper
   • Context retained for extensibility but significantly improved

6. 2nd-person pronoun (Comment 3):

   • Changed to "if direct access is required" (line 496 in docs/users-guide.md)

7. LOC acronym (Comment 4):

   • Expanded to "lines of code (LOC)" (line 49 in execplan)

8. Real-time sleeping / deterministic timeout testing (Overall Comment 2):

   • ✅ Now fully resolved! MessageAssemblyInboundWorld::wait_millis now uses tokio::time::advance(Duration::from_millis(millis)) (line 237)
   • The runtime initialises with tokio::time::pause() (line 174) to enable synthetic time control
   • Timeout-related scenarios are now deterministic and won't be flaky under load

All review comments have been successfully addressed. The codebase now has centralised test helpers, comprehensive test coverage, improved error handling ergonomics, better documentation, and deterministic time-based testing.

[coderabbitai]

Actionable comments posted: 3

🤖 Fix all issues with AI agents

Verify each finding against the current code and only fix it if needed.


In `@src/message_assembler/types.rs`:
- Around line 11-23: Replace the primitive fields in EnvelopeRouting with
newtype wrappers: change pub envelope_id: u32 to pub envelope_id: EnvelopeId and
change pub correlation_id: Option<u64> to pub correlation_id:
Option<CorrelationId>; add or reuse simple newtypes EnvelopeId(u32) and
CorrelationId(u64) (derive/impl Clone, Copy, Debug, PartialEq, Eq, Default as
needed) and update any code that constructs or reads EnvelopeRouting (e.g.,
places that build from the transport frame or match on
envelope_id/correlation_id) to convert between the primitive and newtype
(From/Into or explicit constructors) so routing metadata is type-safe without
altering the external behavior of AssembledMessage dispatch or logging.

In `@tests/fixtures/message_assembly_inbound.rs`:
- Around line 18-38: Add Rustdoc comments for all public types and methods:
document the structs MessageKey and FrameSequence with /// explaining their
purpose and wrapped value, add docs for the function
message_assembly_inbound_world describing what world it constructs/returns, and
annotate the MessageAssemblyInboundWorld type and each of its public methods
with /// describing their behavior, inputs and return values; ensure comments
appear above the type/function/method signatures and follow project style so
cargo doc includes them.
- Around line 275-301: The send_payload function currently takes self.client
before calling serializer.serialize(&envelope) which can ?-return and leave
self.client as None; fix by either performing serialization before taking the
client or, if you must take the client first, ensure you always restore it on
any early return: move the call to serializer.serialize(&envelope) to occur
before self.client.take(), or wrap the serialization in a match/result and
reassign self.client = Some(client) before propagating any Err from serialize,
then proceed with the async send using the restored client (refer to
send_payload, serializer.serialize, self.client, and send_result to locate the
change).