feat(a2a-gateway): add runtime streaming-ingress dispatch wrapper - #441
Merged
Conversation
yordis
commented
Jun 29, 2026
yordis
commented
Member
- Encodes "is this a streaming method?" exactly once at the dispatch boundary so the orchestrator can fan out to the streaming pump without re-parsing wire shapes at every call site.
- Splits the wire-parsing classifier out from the NATS-bound spawner so the typed `StreamingIngressKind` decision (and the `task_id` / `last_seq` resume cursor extraction) stays unit-testable without standing up a real broker.
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
PR SummaryMedium Risk Overview
Reviewed by Cursor Bugbot for commit 2e399cc. Bugbot is set up for automated code reviews on this repo. Configure here. |
|
Warning Review limit reached
Next review available in: 54 minutes Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable usage-based reviews in Billing to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please see our Fair Usage Limits Policy for further information, and refer to the rate limits docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
WalkthroughAdds a new ChangesStreaming Ingress Dispatch
Estimated code review effort🎯 2 (Simple) | ⏱️ ~10 minutes Possibly related PRs
Suggested labels
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 2 potential issues.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit c3ef9ed. Configure here.
| params | ||
| .get("last_seq") | ||
| .or_else(|| params.get("lastSeq")) | ||
| .and_then(Value::as_u64) |
There was a problem hiding this comment.
Resubscribe resume cursor incomplete
Medium Severity
classify_streaming_spawn derives tasks.resubscribe resume position via last_seq_from_resubscribe_params, which only reads last_seq and lastSeq. It ignores lastSequence, last_sequence, and metadata.lastEventId / metadata.last_event_id that parse_resubscribe_params already supports, so those clients resume at sequence zero and replay the full task event stream.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit c3ef9ed. Configure here.
| .get("last_seq") | ||
| .or_else(|| params.get("lastSeq")) | ||
| .and_then(Value::as_u64) | ||
| } |
There was a problem hiding this comment.
Duplicated resubscribe param parsing
Medium Severity
New task_id_from_resubscribe_params and last_seq_from_resubscribe_params reimplement a subset of the crate’s existing parse_resubscribe_params in gw_ingress_stream.rs, which already centralizes task id validation and cursor alias handling for the same wire shape.
Reviewed by Cursor Bugbot for commit c3ef9ed. Configure here.
|
Code Coverage SummaryDetailsDiff against mainResults for commit: 2e399cc Minimum allowed coverage is ♻️ This comment has been updated with latest results |
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
…verage)) Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
rsworkspace/crates/a2a-gateway/src/runtime/streaming/tests.rs (1)
82-145: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a mismatch case to lock in the dispatch-boundary contract.
These tests only cover matching or neutral payload methods, so a regression that re-parses
payload.methodinstead of trustingmethod_dotswould still pass. Since this wrapper exists specifically to classify at the dispatch boundary, add one test with conflicting values.Proposed test
+#[test] +fn classify_uses_dispatch_method_not_payload_method() { + let headers = headers_with_req_id("r-6"); + let payload = br#"{"jsonrpc":"2.0","id":"r-6","method":"tasks/resubscribe","params":{"id":"t-1","last_seq":5}}"#; + let intent = classify_streaming_spawn("message.stream", &headers, payload); + match intent { + StreamingSpawnIntent::Spawn(StreamingIngressKind::MessageStream { .. }) => {} + other => panic!("expected MessageStream kind from method_dots, got {other:?}"), + } +}🤖 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 `@rsworkspace/crates/a2a-gateway/src/runtime/streaming/tests.rs` around lines 82 - 145, Add a regression test in classify_streaming_spawn to cover a dispatch-boundary mismatch where the incoming method argument and the JSON payload method disagree, and assert the wrapper still classifies based on the method passed in rather than re-parsing payload.method. Place it alongside the existing classify_returns_not_streaming_for_unrelated_method and classify_spawns_tasks_resubscribe_kind_with_resume_cursor tests so the contract is locked in for StreamingSpawnIntent and classify_streaming_spawn.rsworkspace/crates/a2a-gateway/src/runtime/streaming.rs (1)
58-75: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winMatch on
method_dotsbefore extractingreq_id.
req_id_from_headers_or_payload(...)runs before the method is known to be streaming, so the wrapper can end up parsing payload metadata for unary requests too. Sincemethod_dotsalone decides whether a request can ever spawn the pump, move thereq_idlookup inside the two streaming arms.Suggested refactor
pub fn classify_streaming_spawn( method_dots: &str, headers: &async_nats::HeaderMap, payload: &[u8], ) -> StreamingSpawnIntent { - let Some(req_id) = req_id_from_headers_or_payload(headers, payload) else { - return StreamingSpawnIntent::NotStreaming; - }; match method_dots { - MESSAGE_STREAM_METHOD_DOTS => StreamingSpawnIntent::Spawn(StreamingIngressKind::MessageStream { req_id }), + MESSAGE_STREAM_METHOD_DOTS => { + let Some(req_id) = req_id_from_headers_or_payload(headers, payload) else { + return StreamingSpawnIntent::NotStreaming; + }; + StreamingSpawnIntent::Spawn(StreamingIngressKind::MessageStream { req_id }) + } TASKS_RESUBSCRIBE_METHOD_DOTS => { + let Some(req_id) = req_id_from_headers_or_payload(headers, payload) else { + return StreamingSpawnIntent::NotStreaming; + }; let params = json_rpc_params(payload); let Some(task_id) = task_id_from_resubscribe_params(¶ms) else { return StreamingSpawnIntent::NotStreaming; }; let last_seq = last_seq_from_resubscribe_params(¶ms).unwrap_or(0);🤖 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 `@rsworkspace/crates/a2a-gateway/src/runtime/streaming.rs` around lines 58 - 75, The streaming intent logic in streaming::spawn should not extract req_id before confirming the request is a streaming method. Move the req_id_from_headers_or_payload lookup inside the MESSAGE_STREAM_METHOD_DOTS and TASKS_RESUBSCRIBE_METHOD_DOTS arms of the match on method_dots, so unary requests never parse payload metadata unnecessarily; keep req_id, task_id_from_resubscribe_params, and last_seq handling within the streaming branches only.
🤖 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 `@rsworkspace/crates/a2a-gateway/src/runtime/streaming.rs`:
- Around line 79-101: The `task_id_from_resubscribe_params` and
`last_seq_from_resubscribe_params` helpers are probing raw `serde_json::Value`
directly instead of using a validated wire type. Introduce a
`TasksResubscribeParamsWire` boundary type, parse the params once there, and
convert into the domain representation exactly once before the spawn path
continues. Keep the public flow on typed data and remove the ad-hoc `.get(...)`
validation chains from `streaming.rs`.
---
Nitpick comments:
In `@rsworkspace/crates/a2a-gateway/src/runtime/streaming.rs`:
- Around line 58-75: The streaming intent logic in streaming::spawn should not
extract req_id before confirming the request is a streaming method. Move the
req_id_from_headers_or_payload lookup inside the MESSAGE_STREAM_METHOD_DOTS and
TASKS_RESUBSCRIBE_METHOD_DOTS arms of the match on method_dots, so unary
requests never parse payload metadata unnecessarily; keep req_id,
task_id_from_resubscribe_params, and last_seq handling within the streaming
branches only.
In `@rsworkspace/crates/a2a-gateway/src/runtime/streaming/tests.rs`:
- Around line 82-145: Add a regression test in classify_streaming_spawn to cover
a dispatch-boundary mismatch where the incoming method argument and the JSON
payload method disagree, and assert the wrapper still classifies based on the
method passed in rather than re-parsing payload.method. Place it alongside the
existing classify_returns_not_streaming_for_unrelated_method and
classify_spawns_tasks_resubscribe_kind_with_resume_cursor tests so the contract
is locked in for StreamingSpawnIntent and classify_streaming_spawn.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: ea6fbd3f-04ac-44e9-a595-ebe7947b33cc
📒 Files selected for processing (3)
rsworkspace/crates/a2a-gateway/src/runtime.rsrsworkspace/crates/a2a-gateway/src/runtime/streaming.rsrsworkspace/crates/a2a-gateway/src/runtime/streaming/tests.rs
Comment on lines
+79
to
+101
| /// Extract the resume task id from a `tasks/resubscribe` params | ||
| /// object. Accepts the three canonical field names (`id`, `task_id`, | ||
| /// `taskId`) so clients written against either the JSON-RPC contract | ||
| /// or the more verbose dispatch schema land at the same typed id. | ||
| #[must_use] | ||
| pub fn task_id_from_resubscribe_params(params: &Value) -> Option<A2aTaskId> { | ||
| params | ||
| .get("id") | ||
| .or_else(|| params.get("task_id")) | ||
| .or_else(|| params.get("taskId")) | ||
| .and_then(Value::as_str) | ||
| .and_then(|raw| A2aTaskId::new(raw).ok()) | ||
| } | ||
|
|
||
| /// Extract the resume sequence number from a `tasks/resubscribe` | ||
| /// params object. Absence means "start from the latest" -- the pump | ||
| /// treats `None` as "no resume cursor" rather than "resume at 0". | ||
| #[must_use] | ||
| pub fn last_seq_from_resubscribe_params(params: &Value) -> Option<u64> { | ||
| params | ||
| .get("last_seq") | ||
| .or_else(|| params.get("lastSeq")) | ||
| .and_then(Value::as_u64) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
Model tasks/resubscribe params as a wire type instead of probing raw JSON.
These helpers keep untrusted serde_json::Value alive across multiple public functions and split validation across ad-hoc .get(...) chains. A dedicated TasksResubscribeParamsWire parsed once and converted into a domain value would keep the spawn path on validated data only. As per coding guidelines, "Untrusted input must use distinct *Input / *Wire / *Request types. Convert those boundary types into domain types exactly once."
🤖 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 `@rsworkspace/crates/a2a-gateway/src/runtime/streaming.rs` around lines 79 -
101, The `task_id_from_resubscribe_params` and
`last_seq_from_resubscribe_params` helpers are probing raw `serde_json::Value`
directly instead of using a validated wire type. Introduce a
`TasksResubscribeParamsWire` boundary type, parse the params once there, and
convert into the domain representation exactly once before the spawn path
continues. Keep the public flow on typed data and remove the ad-hoc `.get(...)`
validation chains from `streaming.rs`.
Source: Coding guidelines
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.