feat(acp-nats): add new_session handler with session.ready lifecycle#12
feat(acp-nats): add new_session handler with session.ready lifecycle#12
Conversation
PR SummaryMedium Risk Overview After a successful session creation, the bridge now asynchronously publishes an Updates Written by Cursor Bugbot for commit c79f468. This will update automatically on new commits. Configure here. |
WalkthroughAdded a New Session flow: new Changes
Sequence Diagram(s)sequenceDiagram
autonumber
actor Client
participant Agent as Bridge/Agent
participant NATS as NATS Broker
participant Metrics as Metrics
Client->>Agent: new_session(NewSessionRequest)
activate Agent
Agent->>Agent: start timer, trace context
Agent->>NATS: request -> agent.session.new (with timeout)
activate NATS
NATS-->>Agent: NewSessionResponse / NatsError
deactivate NATS
alt success
Agent->>Metrics: record success & duration
Agent->>Agent: spawn publish ext.session.ready(session_id)
Agent->>NATS: publish ext.session.ready (ExtSessionReady)
else failure
Agent->>Metrics: record_error("new_session", reason)
Agent->>Metrics: record failure & duration
end
Agent-->>Client: NewSessionResponse / Error
deactivate Agent
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 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 |
yordis
force-pushed
the
feat/acp-nats-new-session
branch
from
ff7cdac to
e6ce156
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: 3
🧹 Nitpick comments (4)
rsworkspace/crates/acp-nats/src/telemetry/metrics.rs (1)
44-56: Consider adding asession_idattribute or structured log context torecord_session_created.Currently
record_session_created()records no attributes. This is fine if you only need aggregate counts, but if you later need to correlate metrics with specific sessions, you'd need to revisit this. Given the caller already logssession_idvia tracing, this is acceptable as-is.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/telemetry/metrics.rs` around lines 44 - 56, record_session_created currently increments sessions_created with no attributes; update it to accept a session identifier and record it as an attribute so metrics can be correlated with a session. Change the signature of record_session_created (e.g., add a session_id parameter) and pass KeyValue::new("session_id", session_id) alongside other attributes to self.sessions_created.add; update all callers to provide the session_id and preserve existing behavior when caller does not have an id.rsworkspace/crates/acp-nats/src/agent/mod.rs (1)
88-93: Task list grows unbounded if sessions are created faster than tasks complete.
retain(|t| !t.is_finished())only prunes on each newspawn_session_readycall. Under sustained high session-creation rates, theVeccould grow if publishes are slow. This is unlikely in practice (publish with flush is fast), but worth keeping in mind. If this ever becomes a concern, consider a bounded channel or a periodic sweep.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/agent/mod.rs` around lines 88 - 93, The Vec stored in session_ready_publish_tasks can grow if sessions are spawned faster than publishes complete because retain(|t| !t.is_finished()) only runs on spawn; replace the Vec+manual retain by a bounded, self-pruning structure such as FuturesUnordered (store a FuturesUnordered<Task> in session_ready_publish_tasks and push the spawned task into it) or else swap the Vec for a bounded mpsc channel and have a dedicated cleanup background task that polls/completes finished tasks; update spawn_session_ready to push into that structure instead of pushing into the raw Vec and ensure the new structure is polled/cleaned regularly so finished tasks are dropped automatically.rsworkspace/crates/acp-nats/src/agent/new_session.rs (1)
25-30: Serialize error leaks internal details, inconsistent with the Deserialize branch.Line 29 includes the raw serialization error in the user-facing message (
format!("Failed to serialize...")), while the Deserialize branch on line 36 correctly returns a generic"Invalid response from agent". Consider whether exposing serialization internals is intentional.Proposed fix for consistency
NatsError::Serialize(inner) => { warn!(error = %inner, "failed to serialize new_session request"); Error::new( ErrorCode::InternalError.into(), - format!("Failed to serialize new_session request: {}", inner), + "Failed to serialize new_session request", ) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs` around lines 25 - 30, The Serialize branch currently logs the internal error but returns a user-facing message containing the raw serialization error; change NatsError::Serialize handling so it still logs the inner error with warn!(error = %inner, "failed to serialize new_session request") but construct the Error::new(ErrorCode::InternalError.into(), ...) with a non-sensitive generic message (for example the same generic used by the Deserialize branch like "Invalid response from agent" or "Failed to serialize new_session request")—do not interpolate inner into the returned string; update the code around NatsError::Serialize, warn!, and Error::new to reflect this.rsworkspace/crates/acp-nats/src/nats/subjects.rs (1)
6-12: Consistent with existing pattern; consider domain types in a future pass.The new helpers follow the same
&strparameter convention asinitialize. Per coding guidelines, domain-specific value objects (e.g.,AcpPrefix,SessionId) are preferred over primitives. Since the existing code already uses&strhere, this is fine for now but worth aligning in a dedicated refactor. As per coding guidelines: "Prefer domain-specific value objects over primitives (e.g.AcpPrefixnotString)."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/nats/subjects.rs` around lines 6 - 12, The helpers session_new and ext_session_ready currently accept primitive &str parameters; change them to take domain types (e.g., AcpPrefix for prefix and SessionId for session_id in ext_session_ready) to match coding guidelines, update the body to use the domain value (e.g., prefix.as_str() or prefix.value()) when formatting, and update all callers to pass the new types (or provide From<&str>/Into implementations for easy conversion); if you can't update callers now, add a short TODO to refactor these functions to AcpPrefix/SessionId in a dedicated pass.
🤖 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/src/agent/mod.rs`:
- Line 62: The CI failed due to formatting at the assignment "let metrics =
self.metrics.clone;" in rsworkspace/crates/acp-nats/src/agent/mod.rs; run cargo
fmt to apply the Rustfmt formatting (or manually reformat the assignment to
conform to rustfmt rules) so the declaration of the local variable metrics (from
self.metrics.clone()) matches the project's style and passes cargo fmt --check.
In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs`:
- Line 38: The CI failed due to formatting differences in
rsworkspace/crates/acp-nats/src/agent/new_session.rs; run rustfmt via `cargo
fmt` (preferably at workspace root) to reformat new_session.rs (the module/file
containing the new session implementation and its closing brace) so the file
complies with `cargo fmt --check`, then commit the formatted changes.
- Around line 46-50: The span in the acp.session.new instrument macro is missing
a declaration for session_id, so the later Span::current().record(...) call in
new_session.rs is a no-op; fix it by adding session_id to the instrument fields
list (e.g. fields(cwd = ?args.cwd, mcp_servers = args.mcp_servers.len(),
session_id)) so that the call to Span::current().record("session_id", &value) in
the new_session function actually stores the value in the span.
---
Nitpick comments:
In `@rsworkspace/crates/acp-nats/src/agent/mod.rs`:
- Around line 88-93: The Vec stored in session_ready_publish_tasks can grow if
sessions are spawned faster than publishes complete because retain(|t|
!t.is_finished()) only runs on spawn; replace the Vec+manual retain by a
bounded, self-pruning structure such as FuturesUnordered (store a
FuturesUnordered<Task> in session_ready_publish_tasks and push the spawned task
into it) or else swap the Vec for a bounded mpsc channel and have a dedicated
cleanup background task that polls/completes finished tasks; update
spawn_session_ready to push into that structure instead of pushing into the raw
Vec and ensure the new structure is polled/cleaned regularly so finished tasks
are dropped automatically.
In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs`:
- Around line 25-30: The Serialize branch currently logs the internal error but
returns a user-facing message containing the raw serialization error; change
NatsError::Serialize handling so it still logs the inner error with warn!(error
= %inner, "failed to serialize new_session request") but construct the
Error::new(ErrorCode::InternalError.into(), ...) with a non-sensitive generic
message (for example the same generic used by the Deserialize branch like
"Invalid response from agent" or "Failed to serialize new_session request")—do
not interpolate inner into the returned string; update the code around
NatsError::Serialize, warn!, and Error::new to reflect this.
In `@rsworkspace/crates/acp-nats/src/nats/subjects.rs`:
- Around line 6-12: The helpers session_new and ext_session_ready currently
accept primitive &str parameters; change them to take domain types (e.g.,
AcpPrefix for prefix and SessionId for session_id in ext_session_ready) to match
coding guidelines, update the body to use the domain value (e.g.,
prefix.as_str() or prefix.value()) when formatting, and update all callers to
pass the new types (or provide From<&str>/Into implementations for easy
conversion); if you can't update callers now, add a short TODO to refactor these
functions to AcpPrefix/SessionId in a dedicated pass.
In `@rsworkspace/crates/acp-nats/src/telemetry/metrics.rs`:
- Around line 44-56: record_session_created currently increments
sessions_created with no attributes; update it to accept a session identifier
and record it as an attribute so metrics can be correlated with a session.
Change the signature of record_session_created (e.g., add a session_id
parameter) and pass KeyValue::new("session_id", session_id) alongside other
attributes to self.sessions_created.add; update all callers to provide the
session_id and preserve existing behavior when caller does not have an id.
ℹ️ Review info
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
rsworkspace/crates/acp-nats/src/agent/mod.rsrsworkspace/crates/acp-nats/src/agent/new_session.rsrsworkspace/crates/acp-nats/src/nats/extensions.rsrsworkspace/crates/acp-nats/src/nats/mod.rsrsworkspace/crates/acp-nats/src/nats/subjects.rsrsworkspace/crates/acp-nats/src/telemetry/metrics.rs
yordis
force-pushed
the
feat/acp-nats-new-session
branch
from
e6ce156 to
c615276
Compare
There was a problem hiding this comment.
♻️ Duplicate comments (3)
rsworkspace/crates/acp-nats/src/agent/mod.rs (1)
62-62:⚠️ Potential issue | 🟡 MinorRustfmt is still required for this file.
CI reports formatting drift at Line 62 (
cargo fmt --all -- --check).🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/agent/mod.rs` at line 62, The file fails rustfmt; run cargo fmt --all (or rustfmt) to reformat mod.rs so the statement including "let metrics = self.metrics.clone();" and surrounding code conform to project formatting; if needed update/commit the formatted file and ensure rustfmt settings (rustfmt.toml) are respected so CI passes.rsworkspace/crates/acp-nats/src/agent/new_session.rs (2)
38-38:⚠️ Potential issue | 🟡 MinorRustfmt is still required for this file.
CI reports formatting drift at Line 38 (
cargo fmt --all -- --check).🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs` at line 38, Run rustfmt to fix formatting drift in rsworkspace/crates/acp-nats/src/agent/new_session.rs: reformat the file (e.g., cargo fmt --all or rustfmt on the file) so it passes CI's cargo fmt --all -- --check and commit the resulting changes; ensure the module new_session.rs is formatted according to rustfmt rules (no trailing whitespace, correct brace/indentation) before pushing.
46-50:⚠️ Potential issue | 🟡 MinorDeclare
session_idin the span fields before recording it.On Line 72,
Span::current().record("session_id", ...)is ineffective becausesession_idis not declared in the#[instrument(fields(...))]list on Lines 46-50.Proposed fix
#[instrument( name = "acp.session.new", skip(bridge, args), - fields(cwd = ?args.cwd, mcp_servers = args.mcp_servers.len()) + fields( + cwd = ?args.cwd, + mcp_servers = args.mcp_servers.len(), + session_id = tracing::field::Empty + ) )] @@ - Span::current().record("session_id", response.session_id.to_string().as_str()); + Span::current().record("session_id", tracing::field::display(&response.session_id));#!/bin/bash set -euo pipefail rg -nP --type=rust -C2 '#\[instrument\(|fields\(|Span::current\(\)\.record\("session_id"' rsworkspace/crates/acp-nats/src/agent/new_session.rsAlso applies to: 72-72
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs` around lines 46 - 50, The span needs to declare session_id in the instrument attribute so Span::current().record("session_id", ...) actually records; update the #[instrument(..., fields(...))] on the function (the #[instrument(name = "acp.session.new", skip(bridge, args), fields(...))] attribute in new_session.rs) to include session_id (e.g. add session_id = tracing::field::Empty) and then keep the existing Span::current().record("session_id", &session_id) call as-is.
🧹 Nitpick comments (2)
rsworkspace/crates/acp-nats/src/agent/new_session.rs (1)
9-44: Prefer explicitNatsErrorarms instead of a wildcard fallback.The wildcard arm catches
PublishOperation,PublishOperationExhausted, andOthervariants without explicit handling. This means futureNatsErroradditions in thetrogon-natscrate will be silently matched by the wildcard instead of producing a compile-time error, making it easy to miss required variant mappings during maintenance.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs` around lines 9 - 44, The match in map_new_session_error currently ends with a wildcard arm which hides unhandled NatsError variants; replace the final "_ =>" fallback with explicit match arms for the remaining NatsError variants (e.g., NatsError::PublishOperation, NatsError::PublishOperationExhausted, NatsError::Other) and handle each by logging (warn! with relevant fields) and returning an appropriate Error (similar pattern to existing arms, e.g., Error::new(ErrorCode::InternalError.into(), "Publish failed" / "Publish exhausted" / include other error text)), so the compiler will force you to handle any new NatsError variants in map_new_session_error.rsworkspace/crates/acp-nats/src/agent/mod.rs (1)
58-64: The method already satisfiestokio::spawnrequirements through struct-level bounds onN: RequestClient + PublishClient + FlushClient, which enforceSend + Sync + Clone + 'static. SinceConfigandMetricsboth deriveCloneand contain onlySendtypes (Duration, metric counters), no additional bounds are needed. The suggestion to add explicit method-level bounds would duplicate what the impl block already specifies, providing only marginal clarity benefit without addressing any actual constraint or behavior change.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/agent/mod.rs` around lines 58 - 64, Don't add redundant method-level Send/Sync/'static bounds to spawn_session_ready; the impl already constrains N via the impl block (N: RequestClient + PublishClient + FlushClient) and Config/Metrics are Clone and contain only Send types, so keep the signature of spawn_session_ready as-is and rely on the existing struct/impl-level bounds (spawn_session_ready, N, Config, Metrics, SessionId, self.nats).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@rsworkspace/crates/acp-nats/src/agent/mod.rs`:
- Line 62: The file fails rustfmt; run cargo fmt --all (or rustfmt) to reformat
mod.rs so the statement including "let metrics = self.metrics.clone();" and
surrounding code conform to project formatting; if needed update/commit the
formatted file and ensure rustfmt settings (rustfmt.toml) are respected so CI
passes.
In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs`:
- Line 38: Run rustfmt to fix formatting drift in
rsworkspace/crates/acp-nats/src/agent/new_session.rs: reformat the file (e.g.,
cargo fmt --all or rustfmt on the file) so it passes CI's cargo fmt --all --
--check and commit the resulting changes; ensure the module new_session.rs is
formatted according to rustfmt rules (no trailing whitespace, correct
brace/indentation) before pushing.
- Around line 46-50: The span needs to declare session_id in the instrument
attribute so Span::current().record("session_id", ...) actually records; update
the #[instrument(..., fields(...))] on the function (the #[instrument(name =
"acp.session.new", skip(bridge, args), fields(...))] attribute in
new_session.rs) to include session_id (e.g. add session_id =
tracing::field::Empty) and then keep the existing
Span::current().record("session_id", &session_id) call as-is.
---
Nitpick comments:
In `@rsworkspace/crates/acp-nats/src/agent/mod.rs`:
- Around line 58-64: Don't add redundant method-level Send/Sync/'static bounds
to spawn_session_ready; the impl already constrains N via the impl block (N:
RequestClient + PublishClient + FlushClient) and Config/Metrics are Clone and
contain only Send types, so keep the signature of spawn_session_ready as-is and
rely on the existing struct/impl-level bounds (spawn_session_ready, N, Config,
Metrics, SessionId, self.nats).
In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs`:
- Around line 9-44: The match in map_new_session_error currently ends with a
wildcard arm which hides unhandled NatsError variants; replace the final "_ =>"
fallback with explicit match arms for the remaining NatsError variants (e.g.,
NatsError::PublishOperation, NatsError::PublishOperationExhausted,
NatsError::Other) and handle each by logging (warn! with relevant fields) and
returning an appropriate Error (similar pattern to existing arms, e.g.,
Error::new(ErrorCode::InternalError.into(), "Publish failed" / "Publish
exhausted" / include other error text)), so the compiler will force you to
handle any new NatsError variants in map_new_session_error.
ℹ️ Review info
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
rsworkspace/crates/acp-nats/src/agent/mod.rsrsworkspace/crates/acp-nats/src/agent/new_session.rsrsworkspace/crates/acp-nats/src/nats/extensions.rsrsworkspace/crates/acp-nats/src/nats/mod.rsrsworkspace/crates/acp-nats/src/nats/subjects.rsrsworkspace/crates/acp-nats/src/telemetry/metrics.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- rsworkspace/crates/acp-nats/src/nats/subjects.rs
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
yordis
force-pushed
the
feat/acp-nats-new-session
branch
3 times, most recently
from
88e4d57 to
e91db52
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
rsworkspace/crates/acp-nats/src/agent/new_session.rs (2)
77-85:spawn_session_readyis fire-and-forget — no shutdown drain.The spawned task's
JoinHandleis dropped immediately (as the existing TODO at line 97 acknowledges). On graceful shutdown, in-flight session-ready publishes will be abandoned mid-flight. If the bridge needs ordered shutdown, these handles should be tracked (e.g. in aVec<JoinHandle<()>>stored onBridge) and awaited during teardown.Do you want me to draft a
session_ready_publish_taskstracking field onBridgewith the corresponding drain logic?🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs` around lines 77 - 85, spawn_session_ready is being fire-and-forget which drops its JoinHandle and abandons in-flight publishes; change spawn_session_ready (or a thin wrapper) to return a JoinHandle<()> and store that handle on Bridge (add a session_ready_publish_tasks: Vec<tokio::task::JoinHandle<()>> or similar) when you call it from new_session; ensure Bridge teardown/drain logic awaits (or aborts gracefully) all handles in session_ready_publish_tasks to guarantee ordered shutdown and completion of pending session-ready publishes. Use the Bridge methods (e.g., where other shutdown work runs) to drain awaiting those JoinHandle<()>s and remove completed handles as needed.
44-49: Consider handlingNatsError::TimeoutandNatsError::Requestvariants explicitly.The wildcard
_arm catches severalNatsErrorvariants that are not semantically addressed:Timeout,Request,PublishOperation,PublishOperationExhausted, andOther. While this mirrors the pattern ininitialize.rs, note thatinitialize.rsexplicitly handlesTimeoutandRequestvariants because they carry semantic meaning (agent availability). If these errors are relevant to session creation semantics, adding explicit arms would improve clarity and maintainability whenNatsErrorevolves.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs` around lines 44 - 49, The match in new_session currently uses a wildcard for NatsError variants; add explicit match arms for NatsError::Timeout and NatsError::Request (and any other semantically-relevant variants) instead of lumping them into the _ arm so their meanings are preserved—map Timeout and Request to a clearer error (e.g., an Unavailable/AgentUnavailable-like ErrorCode or a descriptive message) and keep the remaining variants in a final _ => Error::new(ErrorCode::InternalError.into(), "New session request failed") arm; update the warn!/error logging in each new arm to include %e and a specific message so new_session, NatsError::Timeout, and NatsError::Request are all explicitly handled and easily maintainable.rsworkspace/crates/acp-nats/src/nats/subjects.rs (1)
10-12:session_id: &strshould accept&SessionIdper domain value object guidelines.
ext_session_readyaccepts a raw&strforsession_id, which causes the call site (new_session.rsline 108) to needlessly callsession_id.to_string()to downgrade aSessionIdto&str. The function should accept&SessionIddirectly and call.to_string()internally.♻️ Suggested fix
-pub fn ext_session_ready(prefix: &str, session_id: &str) -> String { - format!("{}.agent.ext.session.ready.{}", prefix, session_id) +pub fn ext_session_ready(prefix: &str, session_id: &SessionId) -> String { + format!("{}.agent.ext.session.ready.{}", prefix, session_id) }And update the call site in
new_session.rs:-let subject = agent::ext_session_ready(&prefix, &session_id.to_string()); +let subject = agent::ext_session_ready(&prefix, &session_id);As per coding guidelines: "Prefer domain-specific value objects over primitives (e.g.
AcpPrefixnotString)."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rsworkspace/crates/acp-nats/src/nats/subjects.rs` around lines 10 - 12, Change ext_session_ready to accept session_id: &SessionId instead of &str and call session_id.to_string() inside the function (so the formatting uses the SessionId domain type). Update the call site that currently passes session_id.to_string() (in the new session creation code) to pass &session_id directly. Ensure imports/references to SessionId are added where needed and keep the function signature: pub fn ext_session_ready(prefix: &str, session_id: &SessionId) -> String.
🤖 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/src/agent/new_session.rs`:
- Around line 44-50: The catch-all arm in new_session currently fails cargo fmt
due to formatting around the Error::new(ErrorCode::InternalError.into(), "New
session request failed") call; run cargo fmt (or rustfmt) and reformat the block
so that the warn!(error = %e, "new_session NATS request failed"); and the
Error::new(...) invocation follow rustfmt's expected layout (preserving the
warn! call and the returned Error), ensuring new_session's match arm adheres to
rustfmt style and compiles cleanly.
---
Nitpick comments:
In `@rsworkspace/crates/acp-nats/src/agent/new_session.rs`:
- Around line 77-85: spawn_session_ready is being fire-and-forget which drops
its JoinHandle and abandons in-flight publishes; change spawn_session_ready (or
a thin wrapper) to return a JoinHandle<()> and store that handle on Bridge (add
a session_ready_publish_tasks: Vec<tokio::task::JoinHandle<()>> or similar) when
you call it from new_session; ensure Bridge teardown/drain logic awaits (or
aborts gracefully) all handles in session_ready_publish_tasks to guarantee
ordered shutdown and completion of pending session-ready publishes. Use the
Bridge methods (e.g., where other shutdown work runs) to drain awaiting those
JoinHandle<()>s and remove completed handles as needed.
- Around line 44-49: The match in new_session currently uses a wildcard for
NatsError variants; add explicit match arms for NatsError::Timeout and
NatsError::Request (and any other semantically-relevant variants) instead of
lumping them into the _ arm so their meanings are preserved—map Timeout and
Request to a clearer error (e.g., an Unavailable/AgentUnavailable-like ErrorCode
or a descriptive message) and keep the remaining variants in a final _ =>
Error::new(ErrorCode::InternalError.into(), "New session request failed") arm;
update the warn!/error logging in each new arm to include %e and a specific
message so new_session, NatsError::Timeout, and NatsError::Request are all
explicitly handled and easily maintainable.
In `@rsworkspace/crates/acp-nats/src/nats/subjects.rs`:
- Around line 10-12: Change ext_session_ready to accept session_id: &SessionId
instead of &str and call session_id.to_string() inside the function (so the
formatting uses the SessionId domain type). Update the call site that currently
passes session_id.to_string() (in the new session creation code) to pass
&session_id directly. Ensure imports/references to SessionId are added where
needed and keep the function signature: pub fn ext_session_ready(prefix: &str,
session_id: &SessionId) -> String.
ℹ️ Review info
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
rsworkspace/crates/acp-nats/src/agent/mod.rsrsworkspace/crates/acp-nats/src/agent/new_session.rsrsworkspace/crates/acp-nats/src/nats/extensions.rsrsworkspace/crates/acp-nats/src/nats/mod.rsrsworkspace/crates/acp-nats/src/nats/subjects.rsrsworkspace/crates/acp-nats/src/telemetry/metrics.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- rsworkspace/crates/acp-nats/src/nats/mod.rs
yordis
force-pushed
the
feat/acp-nats-new-session
branch
3 times, most recently
from
78457fe to
0358eab
Compare
- Add new_session handler aligned with initialize.rs patterns (handler-specific error mapping, single-line generics, local subject variable) - Add spawn_session_ready to publish ext.session.ready after successful session creation - Add SessionReady extension type for bridge-backend coordination - Add session_new and ext_session_ready NATS subject functions - Add record_session_created() and record_error() to Metrics - Add Bridge task tracking for session_ready publish tasks - Add comprehensive test suite mirroring initialize handler tests Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
yordis
force-pushed
the
feat/acp-nats-new-session
branch
from
0358eab to
c79f468
Compare
1 participant
Summary
new_sessionhandler aligned withinitialize.rspatterns (handler-specific error mapping viamap_new_session_error, single-line generics, local subject variable,acp_prefix()accessor)ext.session.readynotification after successful session creation viapublish_session_ready, with a 100ms delay (SESSION_READY_DELAY) to preserve the ordering guarantee documented onExtSessionReady— theAgenttrait returns the response before the transport layer sends it to the client, so without the delay the spawned task could publishsession.readybefore the client receives thesession/newresponseExtSessionReadyextension type,session_newandext_session_readyNATS subject functions, andrecord_error()metricsAcpPrefixinternal storage fromStringtoArc<str>so cloning into spawned tasks is a ref-count bump instead of a heap allocationSee upstream RFD: agentclientprotocol/agent-client-protocol#419
Test plan
cargo check -p acp-natspassescargo test -p acp-nats— all 30 tests pass (18 existing + 12 new)NatsErrorvariantsSendvalues (no borrows fromBridge)SESSION_READY_DELAYprovides sufficient margin for the transport to deliver the response beforesession.readyis published