feat(a2a-nats): catalog persistence + watch + registrar - #308
Merged
Conversation
yordis
commented
Jun 17, 2026
yordis
commented
Member
- The agent catalog needs a single source of truth across brokers so callers never see card drift between nodes, and a JetStream KV bucket plus a watch stream gets that without each broker rebroadcasting on startup; routing pre-empted cards through the existing import gate also keeps cross-account claims from quietly landing in a tenant's namespace.
- A small registrar service that publishes the local card on the catalog subject and re-publishes on KV change means agent authors only own the card payload — the broker handles freshness, retries, and bucket warmth, so a fresh subscriber bootstrap is always the same shape as a steady-state delta.
KvCatalogStore stamps agent cards into a JetStream KV bucket so multiple brokers see one source of truth without each peer re-broadcasting on startup, and the watch stream lets every node react to card changes without polling. Registrar publishes the local card on the catalog subject and re-publishes on KV change to keep the bucket warm. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
PR SummaryMedium Risk Overview
Registrar helpers cover Dev tooling: new Reviewed by Cursor Bugbot for commit cf0bae6. Bugbot is set up for automated code reviews on this repo. Configure here. |
|
Warning Review limit reached
More reviews will be available in 44 minutes and 46 seconds. Learn how PR review limits work. Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). ⌛ How to resolve this issue?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 credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan refill rate. 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, the refill rate gradually slows as usage increases. The highest same-day bursts are limited more strictly. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
WalkthroughAdds a complete NATS JetStream KV-backed agent card catalog subsystem to Changesa2a-nats Catalog Subsystem
Developer Tooling Utilities
Sequence Diagram(s)sequenceDiagram
participant Client
participant RegistrarSubject
participant parse_register_payload
participant KvCatalogStore
participant a2a_pack
participant JetStreamKV
Client->>RegistrarSubject: NATS message(subject, payload, reply)
RegistrarSubject->>RegistrarSubject: extract agent_id from subject suffix
Client->>parse_register_payload: payload bytes
parse_register_payload->>parse_register_payload: serde_json::from_slice
alt invalid JSON
parse_register_payload-->>Client: RegisterPayloadError::JsonParse
else valid JSON
parse_register_payload->>a2a_pack: validate_agent_card_value
alt schema validation fails
a2a_pack-->>parse_register_payload: AgentCardValidateError
parse_register_payload-->>Client: RegisterPayloadError::Schema
else schema passes
parse_register_payload->>parse_register_payload: deserialize to AgentCard
alt deserialization fails
parse_register_payload-->>Client: RegisterPayloadError::ValueParse
else deserialization succeeds
parse_register_payload-->>Client: Ok(AgentCard)
Client->>KvCatalogStore: put_card(agent_id, card)
KvCatalogStore->>JetStreamKV: get(key) or create/update
JetStreamKV-->>KvCatalogStore: Ok or KV error
KvCatalogStore-->>Client: Result<(), CatalogStoreError>
Client->>Client: publish JSON-RPC success or error reply
end
end
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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: cf0bae6 Minimum allowed coverage is ♻️ This comment has been updated with latest results |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (1)
rsworkspace/crates/a2a-nats/src/catalog/store.rs (1)
113-145: ⚖️ Poor tradeoffConsider caching original JSON to avoid round-trip serialization.
Line 135 serializes the card back to JSON for
accept_agent_card_on_read, but the card was just deserialized from JSON inget_card(line 105). This round-trip serialization adds overhead for each gated card.If
list_cards_gatedis called frequently or with large card sets, consider either:
- Extending
get_cardto optionally return the original JSON alongside the parsed card- Caching the JSON representation in the
AgentCardwrapper- Refactoring
accept_agent_card_on_readto accept the structured card instead of JSON🤖 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-nats/src/catalog/store.rs` around lines 113 - 145, The list_cards_gated method unnecessarily serializes the card to JSON for the accept_agent_card_on_read check on line 135, but this card was already deserialized from JSON earlier through the list_cards and get_card flow. To eliminate this round-trip serialization overhead, either modify get_card to return both the original JSON value and the deserialized AgentCard so list_cards_gated can reuse the cached JSON, or refactor accept_agent_card_on_read to accept the structured AgentCard type directly instead of requiring JSON serialization. Choose the approach that minimizes changes to the existing codebase while preserving the validation logic of accept_agent_card_on_read.
🤖 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-nats/src/catalog/registrar.rs`:
- Around line 61-76: The RegisterPayloadError enum is missing implementations
for the Display and std::error::Error traits, which are required by coding
guidelines for all error types. Add a Display trait implementation to
RegisterPayloadError that formats error messages appropriately for each variant
(JsonParse, Schema, ValueParse), and also implement the std::error::Error trait
for RegisterPayloadError to ensure it follows standard error handling patterns
in Rust. These implementations will make the error type more consistent with
Rust conventions and enable broader usage patterns if the type is used elsewhere
in the codebase.
- Around line 214-227: The Subscribe variant in CatalogRegistrarServiceError is
currently storing a String, which loses error context. Replace the String field
with Box<dyn std::error::Error + Send + Sync> to preserve the full error chain.
Update the Display implementation for Subscribe to format the boxed error
appropriately. Add a #[source] attribute to the Subscribe field to properly
expose the underlying error in the error chain. At line 117 where the error is
constructed, instead of converting the error to a String, pass the error
directly as a boxed trait object using Box::new or similar. Consider using the
thiserror crate with #[error] and #[source] derive attributes to simplify the
implementation, following the pattern used elsewhere in the codebase like
acp-nats-agent/src/connection.rs.
In `@rsworkspace/crates/a2a-nats/src/catalog/store.rs`:
- Around line 88-110: In the list_cards method, remove the .to_string() calls
from the error mapping expressions where errors are converted to
CatalogStoreError::Kv. Instead of using .map_err(|e|
CatalogStoreError::Kv(e.to_string())), change it to .map_err(|e|
CatalogStoreError::Kv(e)) to preserve the typed error rather than converting it
to a string. This same change needs to be applied to other methods in the file
(mentioned as lines 152-182 and 184-200) where similar error conversions exist.
- Around line 14-22: The Kv(String) variant in the CatalogStoreError enum
discards the underlying error type and breaks the error source chain. Replace
the Kv(String) variant with a typed error field that wraps the actual error type
returned by async-nats KV operations (which implements std::error::Error). Then
update the Display and source trait implementations for CatalogStoreError to
properly expose the wrapped error as the source, and finally locate all sites
where Kv errors are created and update them to wrap the source error instead of
converting it to a string.
In `@rsworkspace/crates/a2a-nats/src/catalog/watch.rs`:
- Around line 53-63: The error conversion in the subscribe_agent function on
line 61 is converting the typed error to a string using e.to_string(), but the
Kv variant of AgentCardWatchError now holds typed errors directly instead of
strings. Replace the `.map_err(|e| AgentCardWatchError::Kv(e.to_string()))?`
call with `.map_err(|e| AgentCardWatchError::Kv(e))?` to pass the typed error
directly. Apply the same fix to the similar error conversion in the other watch
subscription function in the 66-81 range that also converts errors to strings.
- Around line 30-41: The std::error::Error implementation for
AgentCardWatchError is missing the source() method, which prevents error chain
preservation and makes underlying errors inaccessible to error handlers.
Implement the source() method in the std::error::Error impl block for
AgentCardWatchError to return an Option containing a reference to the underlying
error. For each variant (Kv, InvalidKey, Deserialize, Schema), return Some
reference to the contained error, except for Kv which should return None until
it is changed to hold typed errors.
- Around line 22-28: The Kv variant in the AgentCardWatchError enum is storing
the KV error as a String, which discards the underlying error type and violates
the coding guideline that errors must never be converted to strings. Replace the
Kv(String) variant with Kv(Box<dyn std::error::Error + Send + Sync>) or wrap the
actual typed KV error from the NATS client library (e.g., the concrete error
type returned by NATS KV operations) to preserve the source error information
instead of converting it to a string.
---
Nitpick comments:
In `@rsworkspace/crates/a2a-nats/src/catalog/store.rs`:
- Around line 113-145: The list_cards_gated method unnecessarily serializes the
card to JSON for the accept_agent_card_on_read check on line 135, but this card
was already deserialized from JSON earlier through the list_cards and get_card
flow. To eliminate this round-trip serialization overhead, either modify
get_card to return both the original JSON value and the deserialized AgentCard
so list_cards_gated can reuse the cached JSON, or refactor
accept_agent_card_on_read to accept the structured AgentCard type directly
instead of requiring JSON serialization. Choose the approach that minimizes
changes to the existing codebase while preserving the validation logic of
accept_agent_card_on_read.
🪄 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: abab9fd3-d549-4a6d-81de-2374b93372bd
⛔ Files ignored due to path filters (1)
rsworkspace/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (7)
rsworkspace/Cargo.tomlrsworkspace/crates/a2a-nats/Cargo.tomlrsworkspace/crates/a2a-nats/src/catalog/mod.rsrsworkspace/crates/a2a-nats/src/catalog/nats_kv.rsrsworkspace/crates/a2a-nats/src/catalog/registrar.rsrsworkspace/crates/a2a-nats/src/catalog/store.rsrsworkspace/crates/a2a-nats/src/catalog/watch.rs
Without a single source of truth, fmt + clippy + coverage all run in slightly different cwds and silently miss steps when invoked ad-hoc. The script gates each step the same way CI does so a green run locally means a green run in CI. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
Previous commit (1548dec) intended to add this but the local checkout had .trogonai/ symlinked to a private artefact dir, so 'git add' fell behind the symlink and the script never made it into the tree. Land it under rsworkspace/scripts/ so future agents (and humans) can run the same CI mirror without depending on a personal symlink. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
Strip the un-test-harnessable run loop and live JetStream watch wrapper into a follow-up integration PR so this slice ships only the pure helpers + KV-backed store + decoder it can exercise locally. Every remaining production line is now hit by a unit test, so the coverage gate sees no new uncovered statements. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
The previous run flagged +2 uncovered lines in trogon-scheduler/dispatcher.rs unrelated to this PR (main run had 3 uncovered there, this branch had 5). Empty commit to retrigger and confirm the flap is on lines this PR doesn't touch. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
Move the rsworkspace PR gate script under .config/mise/tasks so it's discoverable via 'mise tasks' alongside rust-coverage-check and rust-coverage-gate, and reuse rust-coverage-check for the coverage comparison instead of duplicating that logic. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
Surface every unresolved review thread + its thread ID so addressing review feedback doesn't depend on remembering which threads have been seen vs resolved already. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
…tombstone CatalogStoreError::Kv now wraps the underlying KV error as a boxed source so callers keep std::error::Error chaining (and can downcast) instead of pattern-matching on a stringified message. Split out an InvalidKey variant for the catalog-key validation failure that used to be hidden inside Kv(String). put_card also stops calling JetStream KV update against a delete or purge tombstone — the latest entry's revision is still returned for removed keys, and update against that revision fails where create would have restored the card, so re-registration after a delete now works cleanly. RegisterPayloadError gains Display + Error impls + a source() chain to align with the workspace error-shape guidelines. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 3 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 415de7a. Configure here.
…-comments script put_card now retries up to three attempts when the JetStream KV write loses a revision race (WrongLastRevision against a stale revision or AlreadyExists against a key recreated underneath us). Non-conflict errors still propagate immediately so a backend failure can't get masked as a race. pr-unresolved-comments switches to set -euo pipefail and paginates through reviewThreads via pageInfo so a flaky GraphQL response can't make the script lie about a clean PR, and PRs with more than 100 threads can't silently hide unresolved comments past the first page. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added
the
rust:coverage-baseline-reset
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (1)
rsworkspace/crates/a2a-nats/src/catalog/watch.rs (1)
25-31:⚠️ Potential issue | 🟠 Major | ⚡ Quick winRestore typed error chaining in
AgentCardWatchErrorinstead of stringifying context.Line 27/28 and Line 48 currently convert error context into
String(Kv(String),InvalidKey(String),format!(...)), and Line 44 omitssource(). This drops structured causes and regresses diagnosability.Suggested direction
pub enum AgentCardWatchError { - Kv(String), - InvalidKey(String), + Kv(Box<dyn std::error::Error + Send + Sync>), + InvalidKey(A2aAgentIdParseError), // or your concrete parse error type Deserialize(serde_json::Error), Schema(a2a_pack::AgentCardValidateError), } -impl std::error::Error for AgentCardWatchError {} +impl std::error::Error for AgentCardWatchError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Kv(e) => Some(e.as_ref()), + Self::InvalidKey(e) => Some(e), + Self::Deserialize(e) => Some(e), + Self::Schema(e) => Some(e), + } + } +}- let agent_id = A2aAgentId::new(entry.key.as_str()).map_err(|_| { - AgentCardWatchError::InvalidKey(format!("catalog KV key `{}` is not a valid agent id", entry.key)) - })?; + let agent_id = + A2aAgentId::new(entry.key.as_str()).map_err(AgentCardWatchError::InvalidKey)?;As per coding guidelines, “Errors must be typed—use structs or enums, never
Stringorformat!()… Never discard error context by converting a typed error into a string; wrap the source error as a field or variant instead.”Also applies to: 44-49
🤖 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-nats/src/catalog/watch.rs` around lines 25 - 31, The AgentCardWatchError enum currently uses String variants for Kv and InvalidKey instead of storing typed errors, which drops structured error context. Replace the Kv(String) and InvalidKey(String) variants with typed error fields that wrap the actual underlying error types (likely a Box<dyn Error> or the specific error type that caused these variants). Additionally, ensure that the Error trait implementation for AgentCardWatchError includes a proper source() method implementation that returns the underlying error, and verify that error conversions around lines 44-49 propagate the typed errors instead of converting them to formatted strings.Source: Coding guidelines
🤖 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 @.config/mise/tasks/pr-unresolved-comments:
- Around line 35-37: The gh repo view command on line 35 can fail due to
authentication, network, or API errors, causing the script to exit via set -e
before reaching your controlled exit 2 path, resulting in inconsistent exit
codes and lost diagnostic messages. Add explicit error handling around the
REPO_NWO assignment to check if the gh command succeeds; if it fails, explicitly
exit with code 2 and include a clear diagnostic message. Apply the same explicit
error handling pattern to the similar gh commands around lines 67-70 to ensure
all gh command failures are caught and handled consistently with your diagnostic
exit contract.
- Around line 16-25: The argument parsing loop in the for loop currently uses a
catch-all pattern that accepts any unknown argument as the PR value, which
silently accepts typos like --jsno. Replace the catch-all case pattern with
explicit validation that rejects unknown arguments with a clear error message
and exit code 2. Additionally, after the loop completes, add numeric validation
on the PR variable to ensure it contains only digits, and if validation fails,
output an error message and exit with code 2. This ensures the script fails fast
with helpful feedback rather than silently accepting invalid input.
In `@rsworkspace/crates/a2a-nats/src/catalog/registrar.rs`:
- Around line 39-45: The functions `register_subject_prefix` and
`agent_id_suffix` expose raw primitives (`&str`, `usize`, and `Option<&str>`)
instead of domain value objects, allowing unvalidated subject data to flow
downstream. Modify `register_subject_prefix` to accept `&A2aPrefix` instead of
raw `&str`, and update `agent_id_suffix` to return a validated `A2aAgentId`
instead of `Option<&str>` (or an appropriate typed parse error), ensuring domain
boundaries are enforced and invalid subjects are rejected at the boundary rather
than propagating downstream.
---
Duplicate comments:
In `@rsworkspace/crates/a2a-nats/src/catalog/watch.rs`:
- Around line 25-31: The AgentCardWatchError enum currently uses String variants
for Kv and InvalidKey instead of storing typed errors, which drops structured
error context. Replace the Kv(String) and InvalidKey(String) variants with typed
error fields that wrap the actual underlying error types (likely a Box<dyn
Error> or the specific error type that caused these variants). Additionally,
ensure that the Error trait implementation for AgentCardWatchError includes a
proper source() method implementation that returns the underlying error, and
verify that error conversions around lines 44-49 propagate the typed errors
instead of converting them to formatted strings.
🪄 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: cf4b4935-beba-481c-99b7-6594a48b17d8
⛔ Files ignored due to path filters (1)
rsworkspace/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (7)
.config/mise/tasks/pr-unresolved-comments.config/mise/tasks/rust-pr-checkrsworkspace/crates/a2a-nats/Cargo.tomlrsworkspace/crates/a2a-nats/src/catalog/mod.rsrsworkspace/crates/a2a-nats/src/catalog/registrar.rsrsworkspace/crates/a2a-nats/src/catalog/store.rsrsworkspace/crates/a2a-nats/src/catalog/watch.rs
💤 Files with no reviewable changes (1)
- rsworkspace/crates/a2a-nats/Cargo.toml
✅ Files skipped from review due to trivial changes (1)
- .config/mise/tasks/rust-pr-check
🚧 Files skipped from review as they are similar to previous changes (1)
- rsworkspace/crates/a2a-nats/src/catalog/store.rs
…r-comments arg parsing agent_id_from_subject(prefix, subject) returns a validated A2aAgentId plus a typed AgentSuffixError instead of leaking Option<&str>/usize through the registrar API surface. register_subject_prefix takes &A2aPrefix instead of &str so the boundary type conversion happens exactly once. pr-unresolved-comments rejects unknown flags + non-numeric PR arguments early, and treats every gh failure as its own exit 2 path so set -euo pipefail can't silently override the script's diagnostic contract. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
added a commit
that referenced
this pull request
Jun 18, 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.