feat(trogon-nats, acp-nats): add JetStream integration foundation - #65
Merged
Conversation
PR SummaryMedium Risk Overview
Written by Cursor Bugbot for commit d6a3590. This will update automatically on new commits. Configure here. |
|
Warning Rate limit exceeded
Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 13 minutes and 10 seconds. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (32)
WalkthroughThis pull request adds JetStream support to the NATS-based agent system. It enables the Changes
Sequence Diagram(s)sequenceDiagram
participant Agent as Agent<br/>(Bridge)
participant Provisioner as provision_streams()
participant JsContext as JetStreamContext
participant NatsJs as async-nats
Agent->>Provisioner: provision_streams(js, prefix)
Provisioner->>Provisioner: Get all configs<br/>(COMMANDS, RESPONSES,<br/>CLIENT_OPS, NOTIFICATIONS)
loop for each stream config
Provisioner->>JsContext: get_or_create_stream(config)
JsContext->>NatsJs: fetch or create stream
NatsJs-->>JsContext: success/error
JsContext-->>Provisioner: Result<()>
alt Stream created successfully
Provisioner->>Provisioner: Log info event
else Stream creation failed
Provisioner->>Provisioner: Map error with stream name
end
end
Provisioner-->>Agent: Result<(), ProvisionError>
sequenceDiagram
participant Connection as NATS Connection
participant Dispatcher as dispatch_js_message()
participant Handler as Agent Handler<br/>(e.g., prompt)
participant Bridge as Bridge<N,C,J>
Connection->>Dispatcher: JetStream message received
Dispatcher->>Dispatcher: Parse subject to method
alt Subject parse successful
Dispatcher->>Dispatcher: Deserialize request body
alt Deserialization successful
Dispatcher->>Handler: Call handler(bridge, args)
Handler->>Bridge: Access NATS/JetStream clients
Bridge-->>Handler: Response/Result
Handler-->>Dispatcher: Result<Response>
alt Handler successful
Dispatcher->>Dispatcher: ack() message
else Handler failed
Dispatcher->>Dispatcher: Send error reply
Dispatcher->>Dispatcher: ack() message
end
else Deserialization failed
Dispatcher->>Dispatcher: term() message
end
else Subject parse failed
Dispatcher->>Dispatcher: term() message
end
Estimated Code Review Effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly Related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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 |
|
Code Coverage SummaryDetailsDiff against mainResults for commit: d6a3590 Minimum allowed coverage is ♻️ This comment has been updated with latest results |
![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 and found 4 potential issues.
Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
yordis
force-pushed
the
feat/jetstream-integration
branch
from
March 28, 2026 02:16
ac72017 to
9a1e4ed
Compare
![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 (4)
rsworkspace/crates/acp-nats-agent/src/connection.rs (1)
326-412: Extract the routing table once.
dispatch_js_messagenow copies the fullAgentMethodmatch fromdispatch_message. The next method addition or dispatch-policy tweak will need two edits and these paths will drift quickly. Pull the shared routing into one helper that returnsResult<(), DispatchError>, then keep only the post-dispatch ack policy separate here.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats-agent/src/connection.rs` around lines 326 - 412, dispatch_js_message duplicates the full AgentMethod routing logic from dispatch_message; extract that shared routing into a single helper (e.g., route_agent_method or dispatch_agent_method) which takes (&JsMessage, &A, &N) or the parsed subject and returns Result<(), DispatchError> (where DispatchError models the per-dispatch failures), move all handle_request/handle_notification match arms into that helper, and have dispatch_js_message call the helper then perform only the post-dispatch ACK/term policy (i.e., keep the js_msg.term()/ack handling in dispatch_js_message). Update both dispatch_js_message and dispatch_message to call the new helper so future method additions only change the single routing function.rsworkspace/crates/trogon-nats/src/jetstream/mocks.rs (1)
287-313:messages()always fails — document this trait implementation quirk.The
JetStreamConsumertrait requiresmessages(), butMockJetStreamConsumeralways returns an error, requiring tests to useraw_messages()instead. While the comment explains this is becauseMockJsMessagecannot convert toJsMessage, this means code using the trait abstraction cannot be tested with this mock.Consider adding a doc comment on
MockJetStreamConsumerexplaining this limitation and the intended testing pattern (useraw_messages()directly rather than the trait method).📝 Suggested documentation
+/// Mock JetStream consumer for testing. +/// +/// **Note:** The [`JetStreamConsumer::messages()`] trait method always returns an error +/// because `MockJsMessage` cannot be converted to `JsMessage` without a real JetStream message. +/// Use [`raw_messages()`](Self::raw_messages) directly in tests to receive `MockJsMessage` values +/// and assert on signal recordings. pub struct MockJetStreamConsumer { rx: Mutex<Option<mpsc::UnboundedReceiver<MockJsMessage>>>, }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-nats/src/jetstream/mocks.rs` around lines 287 - 313, Add a doc comment to the MockJetStreamConsumer type explaining that its JetStreamConsumer::messages() implementation always returns an error because MockJsMessage cannot be converted into a real jetstream::Message, and that tests should call MockJetStreamConsumer::raw_messages() (which returns the underlying MockJsMessage stream) for signal assertions; reference the methods messages(), raw_messages(), the type MockJetStreamConsumer and MockJsMessage in the comment so callers know this mock cannot be used via the JetStreamConsumer trait for message-based tests.rsworkspace/crates/acp-nats/src/jetstream/provision.rs (1)
79-85: Test name doesn't match actual idempotency semantics.The test asserts
ctx.created_streams().len() == 8after twoprovision_streamscalls, which verifies the mock records every call. However, true idempotent behavior would mean calling the function multiple times produces the same end state (4 streams, not 8). The underlyingget_or_create_streamis idempotent in production, but this test verifies mock call recording rather than idempotency.Consider renaming to
provision_calls_create_for_all_streams_each_timeor adjusting the assertion/comment to clarify intent.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/jetstream/provision.rs` around lines 79 - 85, The test named provision_is_idempotent is misleading because it calls provision_streams twice but asserts ctx.created_streams().len() == 8 (recording both calls) instead of verifying idempotent end-state; either rename the test to something like provision_calls_create_for_all_streams_each_time to reflect that the mock records every call, or change the assertion to verify idempotency by asserting the final stream set size (e.g., 4) after two provision_streams calls and/or assert against MockJetStreamContext behavior consistent with get_or_create_stream so the test truly validates idempotent behavior of provision_streams.rsworkspace/crates/trogon-nats/src/jetstream/client.rs (1)
84-101: Consider documenting stream existence precondition.
create_consumercallsget_streamwhich will fail if the stream hasn't been provisioned. SinceJetStreamConsumerFactoryandJetStreamContextare separate traits (correctly following one-trait-per-operation), there's no compile-time enforcement thatprovision_streamsruns beforecreate_consumer.Adding a doc comment clarifying this runtime precondition would help users avoid subtle initialization ordering bugs.
📝 Suggested documentation
impl JetStreamConsumerFactory for NatsJetStreamClient { type Error = JetStreamError; type Consumer = NatsJetStreamConsumer; + /// Creates a pull consumer for the specified stream. + /// + /// # Errors + /// Returns an error if the stream does not exist. Ensure streams are + /// provisioned via [`provision_streams`] before creating consumers. async fn create_consumer( &self, stream_name: &str,🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/trogon-nats/src/jetstream/client.rs` around lines 84 - 101, Document that create_consumer has a runtime precondition: it calls self.context.get_stream and will return an error if the named stream does not exist, so callers must ensure streams are provisioned (e.g., via JetStreamConsumerFactory::provision_streams or the JetStreamContext provisioning routine) before invoking create_consumer; add this note as a doc comment on the create_consumer method (mentioning get_stream, JetStreamConsumerFactory, JetStreamContext, and provision_streams) so users know the required initialization ordering.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rsworkspace/crates/acp-nats-agent/src/connection.rs`:
- Around line 420-434: The catch-all Err(_) currently ack()s all non-handled
errors and thus also acknowledges DispatchError::Reply, dropping requests when
reply publishing fails; change the match to explicitly handle
DispatchError::Reply (match Err(DispatchError::Reply)) and call
js_msg.term().await (with the existing warn logging on error) instead of ack(),
leaving the remaining Err(_) arm to ack() only true handler errors; update the
match arms referencing DispatchError::Reply, js_msg.term(), and js_msg.ack()
accordingly.
---
Nitpick comments:
In `@rsworkspace/crates/acp-nats-agent/src/connection.rs`:
- Around line 326-412: dispatch_js_message duplicates the full AgentMethod
routing logic from dispatch_message; extract that shared routing into a single
helper (e.g., route_agent_method or dispatch_agent_method) which takes
(&JsMessage, &A, &N) or the parsed subject and returns Result<(), DispatchError>
(where DispatchError models the per-dispatch failures), move all
handle_request/handle_notification match arms into that helper, and have
dispatch_js_message call the helper then perform only the post-dispatch ACK/term
policy (i.e., keep the js_msg.term()/ack handling in dispatch_js_message).
Update both dispatch_js_message and dispatch_message to call the new helper so
future method additions only change the single routing function.
In `@rsworkspace/crates/acp-nats/src/jetstream/provision.rs`:
- Around line 79-85: The test named provision_is_idempotent is misleading
because it calls provision_streams twice but asserts ctx.created_streams().len()
== 8 (recording both calls) instead of verifying idempotent end-state; either
rename the test to something like
provision_calls_create_for_all_streams_each_time to reflect that the mock
records every call, or change the assertion to verify idempotency by asserting
the final stream set size (e.g., 4) after two provision_streams calls and/or
assert against MockJetStreamContext behavior consistent with
get_or_create_stream so the test truly validates idempotent behavior of
provision_streams.
In `@rsworkspace/crates/trogon-nats/src/jetstream/client.rs`:
- Around line 84-101: Document that create_consumer has a runtime precondition:
it calls self.context.get_stream and will return an error if the named stream
does not exist, so callers must ensure streams are provisioned (e.g., via
JetStreamConsumerFactory::provision_streams or the JetStreamContext provisioning
routine) before invoking create_consumer; add this note as a doc comment on the
create_consumer method (mentioning get_stream, JetStreamConsumerFactory,
JetStreamContext, and provision_streams) so users know the required
initialization ordering.
In `@rsworkspace/crates/trogon-nats/src/jetstream/mocks.rs`:
- Around line 287-313: Add a doc comment to the MockJetStreamConsumer type
explaining that its JetStreamConsumer::messages() implementation always returns
an error because MockJsMessage cannot be converted into a real
jetstream::Message, and that tests should call
MockJetStreamConsumer::raw_messages() (which returns the underlying
MockJsMessage stream) for signal assertions; reference the methods messages(),
raw_messages(), the type MockJetStreamConsumer and MockJsMessage in the comment
so callers know this mock cannot be used via the JetStreamConsumer trait for
message-based tests.
🪄 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: 9bf89aee-1948-4012-9f7e-7d6a26d5ca92
⛔ Files ignored due to path filters (1)
rsworkspace/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (32)
rsworkspace/crates/acp-nats-agent/Cargo.tomlrsworkspace/crates/acp-nats-agent/src/connection.rsrsworkspace/crates/acp-nats/Cargo.tomlrsworkspace/crates/acp-nats/src/agent/authenticate.rsrsworkspace/crates/acp-nats/src/agent/bridge.rsrsworkspace/crates/acp-nats/src/agent/cancel.rsrsworkspace/crates/acp-nats/src/agent/close_session.rsrsworkspace/crates/acp-nats/src/agent/ext_method.rsrsworkspace/crates/acp-nats/src/agent/ext_notification.rsrsworkspace/crates/acp-nats/src/agent/fork_session.rsrsworkspace/crates/acp-nats/src/agent/initialize.rsrsworkspace/crates/acp-nats/src/agent/list_sessions.rsrsworkspace/crates/acp-nats/src/agent/load_session.rsrsworkspace/crates/acp-nats/src/agent/new_session.rsrsworkspace/crates/acp-nats/src/agent/prompt.rsrsworkspace/crates/acp-nats/src/agent/resume_session.rsrsworkspace/crates/acp-nats/src/agent/set_session_config_option.rsrsworkspace/crates/acp-nats/src/agent/set_session_mode.rsrsworkspace/crates/acp-nats/src/agent/set_session_model.rsrsworkspace/crates/acp-nats/src/constants.rsrsworkspace/crates/acp-nats/src/jetstream/consumers.rsrsworkspace/crates/acp-nats/src/jetstream/mod.rsrsworkspace/crates/acp-nats/src/jetstream/provision.rsrsworkspace/crates/acp-nats/src/jetstream/streams.rsrsworkspace/crates/acp-nats/src/lib.rsrsworkspace/crates/trogon-nats/Cargo.tomlrsworkspace/crates/trogon-nats/src/jetstream/client.rsrsworkspace/crates/trogon-nats/src/jetstream/message.rsrsworkspace/crates/trogon-nats/src/jetstream/mocks.rsrsworkspace/crates/trogon-nats/src/jetstream/mod.rsrsworkspace/crates/trogon-nats/src/jetstream/traits.rsrsworkspace/crates/trogon-nats/src/lib.rs
yordis
force-pushed
the
feat/jetstream-integration
branch
from
March 28, 2026 02:22
9a1e4ed to
08cd5b0
Compare
Enable the async-nats `jetstream` feature and introduce the foundational abstractions for an immutable, logged-everything NATS JetStream system. trogon-nats: - JetStream traits (JetStreamContext, JetStreamPublisher, JetStreamConsumer) - JsMessage wrapper exposing full ack/nak/term/in_progress signal surface - Production impls wrapping async_nats::jetstream - Mock impls with signal recording for testing acp-nats: - 4-stream topology (COMMANDS, RESPONSES, CLIENT_OPS, NOTIFICATIONS) - Stream provisioning with idempotent create-or-update - Consumer config factories for prompt and runner flows - Bridge<N, C, J> generic with J=() default for backward compatibility - New headers: X-Session-Id, X-Causation-Id, ACP_JETSTREAM_ENABLED acp-nats-agent: - dispatch_js_message with ack/term signal handling per outcome Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
force-pushed
the
feat/jetstream-integration
branch
from
March 28, 2026 02:27
08cd5b0 to
d6a3590
Compare
This was referenced Mar 28, 2026
This was referenced Apr 5, 2026
This was referenced Apr 23, 2026
yordis
added a commit
that referenced
this pull request
Jun 16, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 22, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 22, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 23, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 23, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 23, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 25, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 25, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 26, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 26, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 26, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 26, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 26, 2026
Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
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.
Summary
JetStreamContext,JetStreamPublisher,JetStreamConsumer) andJsMessagewrapper with full ack/nak/term/in_progress signal surface totrogon-natsacp-natsBridge<N, C>toBridge<N, C, J=()>for optional JetStream support with zero blast radius on existing codedispatch_js_messagetoacp-nats-agentwith ack/term signal handling per dispatch outcomeDesign
Immutable, logged-everything approach: every ACP interaction is persisted as a fact in JetStream streams. Only high-frequency streaming content chunks use memory-backed ephemeral storage.
Handlers receive the full
JsMessageand own the complete message lifecycle — no abstraction hides JetStream's ack/nak/term/in_progress signals.{PREFIX}_COMMANDS{PREFIX}_RESPONSES{PREFIX}_CLIENT_OPS{PREFIX}_NOTIFICATIONSTest plan
J=()default preserves backward compatibilityMockJsMessage,MockJetStreamContext,MockJetStreamPublisher,MockJetStreamConsumer) with signal recording