feat: agent runtime protocol#5000
Conversation
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughAdds a new 🚥 Pre-merge checks | ✅ 3 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (3 passed)
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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
🤖 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 `@crates/agent_runtime_protocol/examples/mock_container.rs`:
- Around line 127-143: Update the permission handler registered by
on_receive_request so it does not automatically select AllowOnce or AllowAlways.
Default requests to RequestPermissionOutcome::Cancelled, or require explicit
user confirmation before creating SelectedPermissionOutcome; never select
AllowAlways without confirmation.
- Around line 58-75: Update the mock-container startup flow around
run_mock_container, carrier.accept, and acp_ready.notified so container task
failure is observed while waiting for either connection acceptance or readiness
notification. Select both waits against the mutable JoinHandle, propagate its
failure promptly, and preserve the existing successful connection/readiness
flow.
In `@crates/agent_runtime_protocol/SPEC.md`:
- Around line 228-230: Align the ACP delivery behavior with the specification by
updating the dispatch loops that ignore AcpRouter::route errors: add structured
TRACE logging for each failed delivery, including the message, agent, and
agent-instance identifiers. Alternatively, remove the failed-delivery TRACE
guarantee from the corresponding SPEC.md section if tracing is not intended.
In `@crates/agent_runtime_protocol/src/connection/mod.rs`:
- Around line 465-472: Update the COMMAND_METHOD branch in the request-handling
match so Command::parse_message failures produce an invalid-parameters JSON-RPC
error response using the available request_id, rather than continuing silently.
Preserve the existing behavior of ignoring commands that lack a request ID and
continue processing valid commands normally.
- Around line 160-174: Update the request flow surrounding the pending map
insertion and receiver.await so cancellation removes request_id from
self.pending via an asynchronous drop guard. Ensure the guard runs when the
caller drops or times out the future, while preserving the existing response
handling and outbound-send cleanup without removing entries prematurely on
successful responses.
- Around line 350-355: Update the close method to call
shutdown.send_replace(true) instead of shutdown.send(true), preserving the
shutdown state when no subscribers exist so later acp() calls cannot open a
channel. Add a regression test covering close before the first ACP channel is
created.
In `@crates/agent_runtime_protocol/src/schema/v0/mod.rs`:
- Around line 84-89: Update the agent-scoped event schema containing agent_id
and agent_instance_id so deserialization rejects payloads where exactly one
identifier is present, while accepting both or neither as required by SPEC.md.
Implement validation during deserialization or replace the independent options
with one optional pair, and add negative wire tests covering agentId-only and
agentInstanceId-only payloads.
In `@crates/agent_runtime_protocol/src/transport/jsonrpsee/mod.rs`:
- Around line 180-209: The connection routing flow around IncomingConnection and
accept must discard superseded generations before returning a channel. Add the
route generation to each queued IncomingConnection, then have accept validate
that the queued connection still owns the matching route generation and skip
stale or cancelled entries until an active one remains; apply the same
generation-aware handling to the later IncomingConnection construction.
- Around line 143-156: The JSON-RPC transport currently authorizes routes solely
by connection_id; update the route registration, SEND_METHOD handler, and
subscribe flow to require and validate a server-issued per-client capability
token. Bind each route to its owning client, reject mismatched tokens, and
prevent replacement from leaving stale IncomingConnection entries queued;
preserve connection_not_found behavior for invalid or missing routes.
🪄 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: Pro
Run ID: 9d64eb8e-74d5-4f7e-98ed-6164b55dbebc
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock,!**/Cargo.lock
📒 Files selected for processing (18)
Cargo.tomlcrates/agent_runtime_protocol/Cargo.tomlcrates/agent_runtime_protocol/SPEC.mdcrates/agent_runtime_protocol/examples/acp.rscrates/agent_runtime_protocol/examples/commands_and_events.rscrates/agent_runtime_protocol/examples/mock_container.rscrates/agent_runtime_protocol/examples/websocket.rscrates/agent_runtime_protocol/src/connection/mod.rscrates/agent_runtime_protocol/src/connection/test.rscrates/agent_runtime_protocol/src/lib.rscrates/agent_runtime_protocol/src/schema/mod.rscrates/agent_runtime_protocol/src/schema/v0/mod.rscrates/agent_runtime_protocol/src/schema/v0/test.rscrates/agent_runtime_protocol/src/testing/mod.rscrates/agent_runtime_protocol/src/testing/test.rscrates/agent_runtime_protocol/src/transport/jsonrpsee/mod.rscrates/agent_runtime_protocol/src/transport/jsonrpsee/test.rscrates/agent_runtime_protocol/src/transport/mod.rs
| let mut container = tokio::spawn(run_mock_container( | ||
| format!("ws://{address}"), | ||
| Arc::clone(&acp_ready), | ||
| )); | ||
| let incoming = carrier | ||
| .accept() | ||
| .await | ||
| .ok_or("mock container disconnected before subscribing")?; | ||
| println!( | ||
| "service accepted runtime connection {}", | ||
| incoming.connection_id() | ||
| ); | ||
|
|
||
| let service = ServerConnection::connect(incoming.into_channel(), ()); | ||
| acp_ready.notified().await; | ||
|
|
||
| let target = AgentTarget::new(AGENT_ID, AGENT_INSTANCE_ID); | ||
| let acp = service.acp(target)?; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Observe container failure while waiting for readiness.
If run_mock_container fails before subscribing or calling notify_one, main blocks forever on accept() or notified(). Select both readiness waits against &mut container, or use a oneshot readiness channel carrying Result.
🤖 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 `@crates/agent_runtime_protocol/examples/mock_container.rs` around lines 58 -
75, Update the mock-container startup flow around run_mock_container,
carrier.accept, and acp_ready.notified so container task failure is observed
while waiting for either connection acceptance or readiness notification. Select
both waits against the mutable JoinHandle, propagate its failure promptly, and
preserve the existing successful connection/readiness flow.
| .on_receive_request( | ||
| async move |request: RequestPermissionRequest, responder, _connection| { | ||
| println!("\n[ACP permission request] {request:#?}"); | ||
| let selected = request.options.iter().find(|option| { | ||
| matches!( | ||
| option.kind, | ||
| PermissionOptionKind::AllowOnce | PermissionOptionKind::AllowAlways | ||
| ) | ||
| }); | ||
| let outcome = selected.map_or(RequestPermissionOutcome::Cancelled, |option| { | ||
| println!("[ACP permission selected] {}", option.name); | ||
| RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new( | ||
| option.option_id.clone(), | ||
| )) | ||
| }); | ||
| responder.respond(RequestPermissionResponse::new(outcome)) | ||
| }, |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Do not automatically approve ACP permission requests.
This selects AllowOnce or AllowAlways without user confirmation, allowing the adapter to perform privileged actions on the host. Default to Cancelled or require explicit confirmation; never automatically choose AllowAlways.
🤖 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 `@crates/agent_runtime_protocol/examples/mock_container.rs` around lines 127 -
143, Update the permission handler registered by on_receive_request so it does
not automatically select AllowOnce or AllowAlways. Default requests to
RequestPermissionOutcome::Cancelled, or require explicit user confirmation
before creating SelectedPermissionOutcome; never select AllowAlways without
confirmation.
| Unknown targets are discarded locally because the outer message is a | ||
| notification and has no response. Each failed delivery is traced at `TRACE` | ||
| level with its message, agent, and agent-instance identifiers. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Align the failed ACP-delivery guarantee with the implementation.
The dispatch loops discard AcpRouter::route errors without tracing, so the promised TRACE entry is never emitted. Add structured tracing at the ignored error sites or remove this guarantee from the specification.
🤖 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 `@crates/agent_runtime_protocol/SPEC.md` around lines 228 - 230, Align the ACP
delivery behavior with the specification by updating the dispatch loops that
ignore AcpRouter::route errors: add structured TRACE logging for each failed
delivery, including the message, agent, and agent-instance identifiers.
Alternatively, remove the failed-delivery TRACE guarantee from the corresponding
SPEC.md section if tracing is not intended.
| let (sender, receiver) = oneshot::channel(); | ||
| self.pending | ||
| .lock() | ||
| .map_err(|_| ConnectionError::Closed)? | ||
| .insert(request_id, sender); | ||
|
|
||
| if self.outbound.unbounded_send(Ok(message)).is_err() { | ||
| self.pending | ||
| .lock() | ||
| .map_err(|_| ConnectionError::Closed)? | ||
| .remove(&request_id); | ||
| return Err(ConnectionError::Closed); | ||
| } | ||
|
|
||
| receiver.await.unwrap_or(Err(ConnectionError::Closed)) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Remove cancelled commands from pending.
If the caller drops or times out this future after insertion, the receiver disappears but the map entry remains until a response or connection shutdown. A non-responsive peer therefore causes unbounded growth. Use a drop guard that removes request_id on cancellation.
🤖 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 `@crates/agent_runtime_protocol/src/connection/mod.rs` around lines 160 - 174,
Update the request flow surrounding the pending map insertion and receiver.await
so cancellation removes request_id from self.pending via an asynchronous drop
guard. Ensure the guard runs when the caller drops or times out the future,
while preserving the existing response handling and outbound-send cleanup
without removing entries prematurely on successful responses.
| fn close(&self) { | ||
| let _ = self.shutdown.send(true); | ||
| if let Ok(mut channels) = self.inbound.lock() { | ||
| channels.clear(); | ||
| } | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== mod.rs outline ==\n'
ast-grep outline crates/agent_runtime_protocol/src/connection/mod.rs --view expanded | sed -n '1,220p'
printf '\n== relevant lines around close and acp ==\n'
sed -n '1,260p' crates/agent_runtime_protocol/src/connection/mod.rs | cat -n | sed -n '1,260p'
printf '\n== search for shutdown and acp usages ==\n'
rg -n "shutdown|fn acp|send_replace|watch::" crates/agent_runtime_protocol/src/connection/mod.rs crates/agent_runtime_protocol/src -g '!target'Repository: macro-inc/macro
Length of output: 15136
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== AcpRouter methods ==\n'
sed -n '260,360p' crates/agent_runtime_protocol/src/connection/mod.rs | cat -n
printf '\n== connection tests ==\n'
sed -n '1,520p' crates/agent_runtime_protocol/src/connection/test.rs | cat -n | sed -n '1,520p'
printf '\n== search for close-before-open or shutdown-related tests ==\n'
rg -n "close|shutdown|acp\(" crates/agent_runtime_protocol/src/connection/test.rs crates/agent_runtime_protocol/src -g '!target'Repository: macro-inc/macro
Length of output: 29607
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== run_server and run_runtime cleanup paths ==\n'
sed -n '360,540p' crates/agent_runtime_protocol/src/connection/mod.rs | cat -nRepository: macro-inc/macro
Length of output: 5975
🌐 Web query:
tokio watch Sender send no receivers current value retained send_replace semantics
💡 Result:
In Tokio's watch channel, the behavior when there are no receivers depends on which send method is used [1][2]. The standard send method, Sender::send, fails if the channel has no active receivers [1][3]. When it fails, the provided value is not stored in the channel and is instead returned to the caller within a SendError [1][4]. This prevents the value from being available to any future receivers that might be created later [1]. In contrast, methods like Sender::send_replace (as well as send_modify and send_if_modified) permit sending a value even when no receivers exist [1][3]. When you call send_replace, the new value is successfully stored as the current value of the channel, replacing the previous one, and is immediately available to any receiver created via Sender::subscribe [1][2]. The method returns the previous value held by the channel [1][2]. Because of this distinction, send_replace is often considered a safer or more robust default when the intention is to ensure the channel always holds the most recent state, regardless of whether a consumer is currently connected [4]. While the channel is technically "closed" when all receivers are dropped, calling subscribe on the Sender will "reopen" it, allowing new receivers to access the current value stored by previous send_replace operations [1][5].
Citations:
- 1: https://docs.rs/tokio/latest/tokio/sync/watch/struct.Sender.html
- 2: https://docs.rs/tokio/1.48.0/tokio/sync/watch/struct.Sender.html
- 3: https://docs.rs/tokio/latest/src/tokio/sync/watch.rs.html
- 4:
watch::Sender::sendis a error-prone default tokio-rs/tokio#7228 - 5: Documentation of watch::Sender implies that channel is closed forever when all receivers are dropped tokio-rs/tokio#4957
Persist the shutdown state even when no ACP channel is open. send(true) drops the closed state when there are no subscribers, so a later acp() can still open a channel after the connection has already been shut down. Use send_replace(true) here and add a regression test for close-before-first-ACP.
🧰 Tools
🪛 ast-grep (0.44.1)
[hint] 250-355: mod.rs should only declare submodules and re-exports; move logic into its own module file.
Context: impl AcpRouter {
fn new(outbound: LogicalSender) -> Self {
let (shutdown, _) = watch::channel(false);
Self {
outbound,
inbound: Arc::new(Mutex::new(HashMap::new())),
next_message_id: Arc::new(AtomicU64::new(1)),
shutdown,
}
}
fn open(&self, target: AgentTarget) -> Result<Channel, ConnectionError> {
if *self.shutdown.borrow() {
return Err(ConnectionError::Closed);
}
let (peer, bridge) = Channel::duplex();
let Channel { mut rx, tx } = bridge;
let mut inbound = self.inbound.lock().map_err(|_| ConnectionError::Closed)?;
if inbound.contains_key(&target) {
return Err(ConnectionError::InvalidMessage(format!(
"ACP channel already open for {}/{}",
target.agent_id, target.agent_instance_id
)));
}
inbound.insert(target.clone(), tx);
drop(inbound);
let outbound = self.outbound.clone();
let inbound = Arc::clone(&self.inbound);
let next_message_id = Arc::clone(&self.next_message_id);
let mut shutdown = self.shutdown.subscribe();
tokio::spawn(async move {
loop {
tokio::select! {
message = rx.next() => {
let Some(Ok(message)) = message else {
break;
};
let message_id = next_message_id.fetch_add(1, Ordering::Relaxed);
let delivery = AcpMessage::new(
format!("acp-{message_id}"),
target.agent_id.clone(),
target.agent_instance_id.clone(),
message,
);
let Ok(params) = serde_json::to_value(delivery) else {
break;
};
let Ok(message) =
RawJsonRpcMessage::notification(ACP_METHOD.to_owned(), params)
else {
break;
};
if outbound.unbounded_send(Ok(message)).is_err() {
break;
}
}
_ = shutdown.changed() => break,
}
}
if let Ok(mut channels) = inbound.lock() {
channels.remove(&target);
}
});
Ok(peer)
}
#[tracing::instrument(
name = "route_acp_message",
level = "trace",
skip(self, delivery),
fields(
message_id = %delivery.message_id,
agent_id = %delivery.agent_id,
agent_instance_id = %delivery.agent_instance_id,
),
err(level = "trace")
)]
fn route(&self, delivery: AcpMessage) -> Result<(), ConnectionError> {
let target = AgentTarget::new(delivery.agent_id, delivery.agent_instance_id);
let sender = self
.inbound
.lock()
.map_err(|_| ConnectionError::Closed)?
.get(&target)
.cloned()
.ok_or_else(|| {
ConnectionError::InvalidMessage(format!(
"no ACP channel for {}/{}",
target.agent_id, target.agent_instance_id
))
})?;
sender
.unbounded_send(Ok(delivery.message))
.map_err(|_| ConnectionError::Closed)
}
fn close(&self) {
let _ = self.shutdown.send(true);
if let Ok(mut channels) = self.inbound.lock() {
channels.clear();
}
}
}
Note: Rule CS-25 in docs/STYLE_GUIDE.md, mined from PR #4175.
Define functions, types, and impls in a named module file and re-export
from mod.rs if needed. lib.rs is deliberately exempt: a small crate
living entirely in lib.rs is fine.
Aspirational: ~246 existing mod.rs files violate this; apply it to new
modules and to mod.rs files you are already refactoring.
(rust-mod-rs-declarations-only)
🤖 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 `@crates/agent_runtime_protocol/src/connection/mod.rs` around lines 350 - 355,
Update the close method to call shutdown.send_replace(true) instead of
shutdown.send(true), preserving the shutdown state when no subscribers exist so
later acp() calls cannot open a channel. Add a regression test covering close
before the first ACP channel is created.
| match method { | ||
| COMMAND_METHOD => { | ||
| let Some(request_id) = request_id else { | ||
| continue; | ||
| }; | ||
| let Ok(command) = Command::parse_message(method, ¶ms) else { | ||
| continue; | ||
| }; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Respond to malformed command requests instead of dropping them.
A command with a valid request ID but invalid parameters is silently ignored, leaving its caller waiting indefinitely. Return an invalid-parameters JSON-RPC response using the same request ID.
🤖 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 `@crates/agent_runtime_protocol/src/connection/mod.rs` around lines 465 - 472,
Update the COMMAND_METHOD branch in the request-handling match so
Command::parse_message failures produce an invalid-parameters JSON-RPC error
response using the available request_id, rather than continuing silently.
Preserve the existing behavior of ignoring commands that lack a request ID and
continue processing valid commands normally.
| /// Stable logical agent identifier for an agent-scoped event. | ||
| #[serde(skip_serializing_if = "Option::is_none")] | ||
| pub agent_id: Option<String>, | ||
| /// Current agent-process identifier for an agent-scoped event. | ||
| #[serde(skip_serializing_if = "Option::is_none")] | ||
| pub agent_instance_id: Option<String>, |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Reject partially scoped system events.
These independent optional fields accept an event containing only agentId or only agentInstanceId, despite SPEC.md Lines 160-161 requiring both or neither. Such events parse successfully and reach the system-event handler. Add deserialization validation or model the target as one optional pair, with negative wire tests.
🤖 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 `@crates/agent_runtime_protocol/src/schema/v0/mod.rs` around lines 84 - 89,
Update the agent-scoped event schema containing agent_id and agent_instance_id
so deserialization rejects payloads where exactly one identifier is present,
while accepting both or neither as required by SPEC.md. Implement validation
during deserialization or replace the independent options with one optional
pair, and add negative wire tests covering agentId-only and agentInstanceId-only
payloads.
| module.register_async_method(SEND_METHOD, move |params, _, _| { | ||
| let state = state.clone(); | ||
| async move { | ||
| let delivery = params.parse::<RuntimeMessage>()?; | ||
| let inbound = state | ||
| .routes | ||
| .lock() | ||
| .map_err(|_| internal_error("connection registry is unavailable"))? | ||
| .get(&delivery.connection_id) | ||
| .map(|route| route.inbound.clone()) | ||
| .ok_or_else(|| connection_not_found(&delivery.connection_id))?; | ||
| inbound | ||
| .unbounded_send(Ok(delivery.message)) | ||
| .map_err(|_| connection_not_found(&delivery.connection_id))?; |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the target module and inspect relevant symbols.
ast-grep outline crates/agent_runtime_protocol/src/transport/jsonrpsee/mod.rs --view expanded || true
echo
echo "=== Relevant lines in mod.rs ==="
sed -n '1,260p' crates/agent_runtime_protocol/src/transport/jsonrpsee/mod.rs | cat -n | sed -n '1,260p'
echo
echo "=== Search for connection_id / route ownership / accept queue ==="
rg -n "connection_id|IncomingConnection|register_async_method\\(|SEND_METHOD|accept\\(" crates/agent_runtime_protocol/src/transport/jsonrpsee -S
echo
echo "=== Search broader transport/protocol references ==="
rg -n "IncomingConnection|OutgoingConnection|connection_id|routes\\.lock|unbounded_send|connection_not_found" crates/agent_runtime_protocol/src -SRepository: macro-inc/macro
Length of output: 24322
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== connect_runtime and helpers ==="
sed -n '260,380p' crates/agent_runtime_protocol/src/transport/jsonrpsee/mod.rs | cat -n
echo
echo "=== jsonrpsee transport tests ==="
sed -n '1,360p' crates/agent_runtime_protocol/src/transport/jsonrpsee/test.rs | cat -n
echo
echo "=== call sites / construction of ServerTransport ==="
rg -n "ServerTransport::new\\(|rpc_module\\(|connect_runtime\\(" crates/agent_runtime_protocol/src -SRepository: macro-inc/macro
Length of output: 17857
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== connect_runtime call sites across repo ==="
rg -n "connect_runtime\\(" -S .
echo
echo "=== higher-level transport wiring around jsonrpsee ==="
rg -n "jsonrpsee|ServerTransport|SUBSCRIBE_METHOD|SEND_METHOD" crates/agent_runtime_protocol/src -SRepository: macro-inc/macro
Length of output: 4753
Bind routes to the owning client, not just connectionId.
RpcModule<()> has no caller identity, so any websocket client can send to or replace another client's route by guessing its connectionId. A replacement before accept() can also leave a stale IncomingConnection in the queue. Use an authenticated per-connection token or other server-issued capability for both send and subscribe.
🤖 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 `@crates/agent_runtime_protocol/src/transport/jsonrpsee/mod.rs` around lines
143 - 156, The JSON-RPC transport currently authorizes routes solely by
connection_id; update the route registration, SEND_METHOD handler, and subscribe
flow to require and validate a server-issued per-client capability token. Bind
each route to its owning client, reject mismatched tokens, and prevent
replacement from leaving stale IncomingConnection entries queued; preserve
connection_not_found behavior for invalid or missing routes.
| let generation = state.generation.fetch_add(1, Ordering::Relaxed); | ||
| let (connection, carrier) = Channel::duplex(); | ||
| let Channel { | ||
| mut rx, | ||
| tx: inbound, | ||
| } = carrier; | ||
| let (cancel, mut cancelled) = oneshot::channel(); | ||
| let previous = state.routes.lock().ok().and_then(|mut routes| { | ||
| routes.insert( | ||
| subscribe.connection_id.clone(), | ||
| Route { | ||
| generation, | ||
| inbound, | ||
| cancel, | ||
| }, | ||
| ) | ||
| }); | ||
| if let Some(previous) = previous { | ||
| let _ = previous.cancel.send(()); | ||
| } | ||
|
|
||
| if state | ||
| .incoming | ||
| .send(IncomingConnection { | ||
| connection_id: subscribe.connection_id.clone(), | ||
| channel: connection, | ||
| }) | ||
| .is_err() | ||
| { | ||
| remove_route(&state, &subscribe.connection_id, generation); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Discard superseded connections before returning from accept().
If a replacement subscribes before the service drains incoming, the cancelled connection remains queued ahead of the active generation. accept() then returns a closed, stale channel first. Carry the generation in IncomingConnection and skip entries that no longer own the route.
Also applies to: 240-243
🤖 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 `@crates/agent_runtime_protocol/src/transport/jsonrpsee/mod.rs` around lines
180 - 209, The connection routing flow around IncomingConnection and accept must
discard superseded generations before returning a channel. Add the route
generation to each queued IncomingConnection, then have accept validate that the
queued connection still owns the matching route generation and skip stale or
cancelled entries until an active one remains; apply the same generation-aware
handling to the later IncomingConnection construction.
b1aa510 to
4ae06da
Compare
4ae06da to
e448b80
Compare
whutchinson98
left a comment
whutchinson98
left a comment
There was a problem hiding this comment.
@seanaye do you think the agent_runtime_protocol could be hexified?
I feel like it would be useful to do this to make it easy to test/mock for testing.
cc @ehayes2000 please update this
2 participants
No description provided.