perf: transcode Python request plane directly

[jthomson04]

Overview

Make MessagePack the configured request-plane payload default and remove the
serde_json::Value materialization from network-facing Python workers. JSON
remains available as an explicit compatibility fallback, and the payload wire
shape remains JSON-compatible.

Details

• add a statically dispatched ingress payload adapter and a Python-specific implementation that transcodes wire bytes directly to and from Python-owned objects
• use the direct adapter for unary and bidirectional Python endpoints while retaining the existing typed Python engine for health checks and in-process operations
• preserve annotated responses, backend error mapping, cancellation, health notifications, clean terminal frames, and the existing JSON-compatible payload domain
• treat missing and explicit None annotated fields identically while retaining the original nested Python payload object
• poll network-facing Python generators on demand so a generator can safely reuse and mutate a dict/list after each yield; the current item is encoded before Python is resumed for the next one
• prevent response-encoding failures from being reported as a clean, truncated stream and expose them through the work-handler serialization error metric
• document the MessagePack default plus mixed-version upgrade and rollback behavior

Python ↔ Rust boundary

PR #10437 added MessagePack request-plane framing, but the Python worker path still materialized every request and response as a JSON-shaped Rust value tree. MessagePack was therefore the wire codec, while serde_json::Value remained the in-process interchange type at the Python/Rust boundary.

Direction	Before this PR	After this PR
Frontend → Python worker	payload bytes → JSON/MessagePack deserializer → serde_json::Value tree → pythonize → Python object	payload bytes → JSON/MessagePack deserializer → serde_transcode + Pythonizer → Python object
Python worker → frontend	Python object → depythonize → serde_json::Value tree → JSON/MessagePack serializer → payload bytes	Python object → Depythonizer + serde_transcode → JSON/MessagePack serializer → payload bytes

The network-only PythonPayload(Py<PyAny>) and PythonResponseItem types keep the payload Python-owned until the selected wire serializer consumes it. No intermediate serde_json::Value, rmpv::Value, or other Rust value tree is constructed on this network path. Annotated envelopes are inspected in place, and their nested data object is passed through without rebuilding it.

This is intentionally scoped to request/response payloads. Control messages, response prologues, and framing remain JSON; the supported payload domain remains JSON-compatible; and the generic typed PythonAsyncEngine remains in place for health checks, LoRA operations, and other in-process integrations.

Where should reviewer start?

1. lib/runtime/src/pipeline/network.rs for the codec default and ingress adapter contract.
2. lib/bindings/python/rust/python_payload.rs for direct Python transcoding and annotated response handling.
3. lib/bindings/python/rust/engine.rs for the network-only Python engines and demand-driven generator polling.
4. lib/runtime/src/pipeline/network/ingress/push_handler.rs for adapter dispatch and response-stream error handling.

Related Issues

No tracking issue. This is a follow-up to PR #10437.

Compatibility

• DYN_REQUEST_PLANE_CODEC=json remains the explicit rollback and mixed-version override.
• Control messages without payload_codec still decode as JSON, so old frontends remain compatible with new workers.
• New frontends sending to Dynamo v1.2 workers must use the JSON override.
• Control messages, response prologues, and framing remain JSON.

Validation

• cargo fmt --all -- --check
• cargo check --locked for dynamo-runtime and the Python binding
• cargo clippy --locked --all-targets with warnings denied for dynamo-runtime and the Python binding
• request-plane codec unit tests: 7 passed
• Rust bidirectional default-MessagePack and explicit-JSON tests: passed
• Python direct unary/annotated/error/malformed-output/health tests under MessagePack and JSON: passed
• explicit-None annotated metadata under MessagePack and JSON: passed
• reused mutable Python response objects under unary and bidirectional MessagePack and JSON paths: passed
• deterministic response-encoding failure, incomplete-stream propagation, and serialization metric regression: passed
• direct client cancellation and metadata propagation under NATS and TCP, plus cancellation examples: passed
• launched frontend-to-Python bidirectional tests under MessagePack and JSON: 2 passed per codec
• Ruff format/check: passed
• fern check and fern docs broken-links: passed

Performance

Controlled three-run comparison

AgentX 060526 Weka trace, concurrency 512, four typed mock workers behind a network-only Python proxy, fixed jemalloc and CPU topology, 45-second load after 32 warmups, three clean process-restarted repetitions per ref/codec pair, and zero errors across all 12 runs:

Codec	main median req/s	branch median req/s	delta
MessagePack	138.374	142.336	+2.86%
JSON	137.792	142.962	+3.75%

On-CPU profile impact

Matched low-overhead MessagePack profiles used cpu-clock:u at 99 Hz with frame-pointer call graphs. Both runs completed with zero errors at essentially identical throughput (main 140.45 req/s, branch 140.08 req/s).

Stack category	main sample share	branch sample share	relative change
payload conversion	35.67%	26.69%	-25.2%
serde_json::Value	32.86%	8.76%	-73.3%
allocator	6.65%	5.51%	-17.1%
GIL-related	13.65%	14.17%	+3.8%

The profile matches the intended mechanism: removing the intermediate Rust value tree substantially reduces serde_json::Value construction/traversal and the associated allocation/deallocation work. Payload conversion remains because bytes still have to be decoded into Python objects and Python objects still have to be encoded onto the wire; the improvement is eliminating the extra Rust tree and second conversion pass. The GIL-related share is effectively flat, so the gain is primarily conversion and allocator work rather than avoiding the GIL.

The controlled matrix was captured on base fa6894c9a8 before the final conflict-free rebase to ccc835a80c; the four intervening main commits did not touch the request-plane or Python binding files changed here. Scripts, raw records, logs, and profile artifacts are retained locally under .bench/request-plane-msgpack-python/.

Summary by CodeRabbit

• New Features

  • Request-plane payloads now default to MessagePack, with an option to use JSON for compatibility.
  • Python endpoints can now pass through raw Python request/response payloads, including streamed and annotated outputs.

• Bug Fixes

  • Mixed-version request handling is clearer, including safer fallback behavior for older workers and invalid codec settings.

• Documentation

  • Updated request-plane docs with codec configuration, rollback guidance, and restart requirements.

• Tests

  • Added end-to-end coverage for both MessagePack default behavior and explicit JSON codec support.

[github-actions]

🌿 Fern Docs Preview: https://nvidia-preview-019f1f63-74e6-724a-bf5f-f252b0591e4c.docs.buildwithfern.com/dynamo/dev

[datadog-official]

🎯 Code Coverage (details)
• Patch Coverage: 14.81%
• Overall Coverage: 41.22% (-5.59%)

_{This comment will be updated automatically if new data arrives.
🔗 Commit SHA: 99b7dc6 | Docs | Give us feedback!}

[devin-ai-integration]

✅ Devin Review: No Issues Found

Devin Review analyzed this PR and found no bugs or issues to report.

[coderabbitai]

Walkthrough

This PR adds a configurable request-plane payload codec defaulting to MessagePack (JSON optional), generalizes the Ingress pipeline with pluggable request/response payload adapters, and introduces a Python-owned payload path (PythonPayload/PythonResponseItem/PythonIngressPayloadAdapter) replacing serde_json::Value-based conversion in Python bindings, with supporting docs and tests.

Changes

Request-Plane Payload Codec and Adapter Pipeline

Layer / File(s)	Summary
Codec defaulting and docs lib/runtime/src/pipeline/network.rs, docs/design-docs/request-plane.md	RequestPlanePayloadCodec becomes public; config parsing now defaults to Msgpack (was Json) for missing/invalid values; docs describe the new default, DYN_REQUEST_PLANE_CODEC=json override, and rollback behavior.
Ingress adapter traits lib/runtime/src/pipeline/network.rs	Adds EncodedResponseFrame, IngressRequestDecoder, IngressResponseEncoder, IngressPayloadAdapter traits and SerdeIngressPayloadAdapter; makes Ingress generic over an Adapter with new constructors new_with_adapter/for_engine_with_adapter.
Push handler wiring lib/runtime/src/pipeline/network/ingress/push_handler.rs	Request decoding and response streaming routed through the injected adapter; IngressDispatch/PushWorkHandler impls updated with Adapter trait bounds for unary and bidirectional paths.
Python payload module lib/bindings/python/rust/python_payload.rs, lib/bindings/python/Cargo.toml	Adds PythonPayload, PythonResponseItem, PythonIngressPayloadAdapter implementing the new ingress traits with serde/pythonize round-trips, annotated response parsing, and unit tests; adds bytes/serde-transcode dependencies.
Python engine raw frame forwarding lib/bindings/python/rust/engine.rs	PythonNetworkEngine/PythonBidirectionalEngine switch from serde_json::Value/Annotated to PythonPayload/PythonResponseItem; adds map_python_exception and spawn_raw_response_forwarder.
Endpoint ingress wiring and docs lib/bindings/python/rust/lib.rs	Replaces JSON ingress aliases with PythonServerStreamingIngress/PythonBidirectionalIngress using PythonIngressPayloadAdapter; updates docstrings for Python-owned frame semantics.
Tests lib/bindings/python/tests/test_request_plane_python_payload.py, lib/runtime/tests/bidirectional_e2e.rs, lib/runtime/tests/bidirectional_e2e_json.rs	Adds Python integration test for payload adapter streaming/error handling, and Rust tests verifying default Msgpack and explicit JSON codec selection in bidirectional echo flows.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~75 minutes

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title is concise and accurately reflects the main change: direct Python request-plane transcoding.
Docstring Coverage	✅ Passed	Docstring coverage is 87.32% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description check	✅ Passed	The PR description covers the required sections and is detailed; only the Related Issues section deviates slightly from the exact template format.

---

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

[coderabbitai]

Actionable comments posted: 3

🧹 Nitpick comments (1)

lib/bindings/python/tests/test_request_plane_python_payload.py (1)

55-95: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win

Add a timeout marker for this network-waiting test.

The test performs client.wait_for_instances() and iterates network response streams (including malformed/error cases that depend on backend error propagation). Per pytest guidelines for this path, tests with polling/network waits should carry @pytest.mark.timeout(...) to prevent CI hangs if the stream never resolves (e.g., a regression causing the error/malformed cases to hang instead of raising).

As per path instructions: "Add @pytest.mark.timeout(...) for any test expected to run >30s (or any polling/network waits/subprocess waits) to prevent CI hangs."

⏱️ Suggested fix

 `@pytest.mark.asyncio`
+@pytest.mark.timeout(30)
 `@pytest.mark.parametrize`("request_plane", ["tcp"], indirect=True)
 async def test_python_request_plane_plain_annotated_error_and_malformed_frames(

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@lib/bindings/python/tests/test_request_plane_python_payload.py` around lines
55 - 95, This test in
test_python_request_plane_plain_annotated_error_and_malformed_frames performs
network waiting and stream iteration that can hang CI, so add a pytest timeout
marker to the test. Apply `@pytest.mark.timeout` to the existing async test
alongside the current asyncio/parametrize markers, using a reasonable limit for
the request_plane_client.generate flow and the malformed/error cases so failures
surface instead of stalling.

Source: Path instructions

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@lib/bindings/python/rust/python_payload.rs`:
- Around line 255-266: Treat explicit None as absent in the Python payload
parsing logic, because PythonPayload fields like data, id, event, comment, and
error currently use PyDict::get_item()/extract_optional in a way that still
treats present None as a value and can break annotated optional handling. Update
the shared helper used by python_payload.rs to detect PyAny::is_none() and
return None for those keys, then reuse that helper for all envelope optionals so
the behavior matches missing-field semantics.

In `@lib/runtime/src/pipeline/network/ingress/push_handler.rs`:
- Around line 172-181: The response-encoding failure path in
`PushHandler::handle` / `encode_response` currently breaks out while
`send_complete_final` remains enabled, so a clean terminal frame can still be
emitted later from the final-send path. Update the `Err(err)` branch to also
disable the clean-final behavior (or otherwise record that the final frame must
be an error/omitted) before breaking, and make the final frame emission logic
that follows respect that flag so no success terminal frame is sent after an
`encode_response` failure.
- Around line 350-354: The adapter-backed ingress impls still impose the old
serde response bounds on U, which blocks non-serde response types even though
encoding now belongs to IngressResponseEncoder. Update the relevant
IngressDispatch and PushWorkHandler impls for Ingress<SingleIn<T>, ManyOut<U>,
Adapter> / related variants to remove U: Serialize + MaybeError and keep only
the bounds required by Adapter and the handler flow. Make sure the signatures in
the impacted impl blocks use the new adapter-based response contract
consistently.

---

Nitpick comments:
In `@lib/bindings/python/tests/test_request_plane_python_payload.py`:
- Around line 55-95: This test in
test_python_request_plane_plain_annotated_error_and_malformed_frames performs
network waiting and stream iteration that can hang CI, so add a pytest timeout
marker to the test. Apply `@pytest.mark.timeout` to the existing async test
alongside the current asyncio/parametrize markers, using a reasonable limit for
the request_plane_client.generate flow and the malformed/error cases so failures
surface instead of stalling.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Enterprise

Run ID: 351843c1-e077-46ec-933a-21ed55eb7dbf

📥 Commits

Reviewing files that changed from the base of the PR and between ccc835a and 963f3b3.

⛔ Files ignored due to path filters (1)

• lib/bindings/python/Cargo.lock is excluded by !**/*.lock

📒 Files selected for processing (10)

• docs/design-docs/request-plane.md
• lib/bindings/python/Cargo.toml
• lib/bindings/python/rust/engine.rs
• lib/bindings/python/rust/lib.rs
• lib/bindings/python/rust/python_payload.rs
• lib/bindings/python/tests/test_request_plane_python_payload.py
• lib/runtime/src/pipeline/network.rs
• lib/runtime/src/pipeline/network/ingress/push_handler.rs
• lib/runtime/tests/bidirectional_e2e.rs
• lib/runtime/tests/bidirectional_e2e_json.rs

[jthomson04]

Automated review (medium effort). The core refactor looks sound — I separately confirmed that the codec default asymmetry (serde #[default]=Json for absent-field old senders vs. configured()=Msgpack for new senders) is intentional and correct, that all existing Ingress call sites still compile via the defaulted type param, and that the deleted unary error-mapping is faithfully preserved in map_python_exception.

5 inline findings below, most-severe first: 2 correctness, 3 altitude/cleanup.

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[jthomson04]

/ok to test ad0db97

[GuanLuo]

left question, approve to unblock

[jthomson04]

/ok to test 99b7dc6