Security Fix: Prevent cron output leakage via deferred injection (Fixes #498)#500
Open
Security Fix: Prevent cron output leakage via deferred injection (Fixes #498)#500
Conversation
…rred injection SECURITY FIX: Deferred messages now stay bound to original target channel Changes: - Added ActiveChannelKey struct to track channels by exact (agent_id, conversation_id) pair instead of just conversation_id, preventing cross-agent message leakage - Added DeferredMessageQueue to store messages bound to their specific target - Modified injection logic at line ~2430 to queue messages when target is inactive instead of delivering them to unrelated active channels - Added automatic delivery of deferred messages when target channel becomes active (in both idle worker resumption and new channel creation paths) - Added regression tests for the security fix Security Impact: - Prevents cron job output intended for private DMs from being delivered to public channels when the DM channel is inactive - Messages are now strictly bound to their (agent_id, conversation_id) target Fixes #498
Contributor
WalkthroughReplaced the conversation-keyed channel registry with an Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes 🚥 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 docstrings
🧪 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
src/main.rs (3)
2162-2185: RedundantActiveChannelKeycreation.The
channel_keyis created on line 2155 for theinsert(), then recreated identically on line 2163 fordrain_for(). Reuse the existing key:active_channels.insert( - ActiveChannelKey::new(agent_id.clone(), conversation_id.clone()), + channel_key.clone(), ActiveChannel { message_tx: channel_tx, _outbound_handle: outbound_handle, }, ); // Deliver any deferred messages that were waiting for this channel - let channel_key = ActiveChannelKey::new(agent_id.clone(), conversation_id.clone()); let deferred = deferred_messages.drain_for(&channel_key);The same pattern appears at lines 2349-2358.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/main.rs` around lines 2162 - 2185, You're creating a duplicate ActiveChannelKey: instead of constructing a new ActiveChannelKey for drain_for(), reuse the previously created channel_key used for insert(); replace the second ActiveChannelKey::new(...) call with the existing channel_key variable when calling deferred_messages.drain_for(&channel_key) and when looking up active_channels.get(&channel_key). Apply the same fix to the other repeated pattern that creates a second ActiveChannelKey (the block that mirrors this behavior later in the file) so both places reuse the originally created channel_key rather than recreating it.
3875-3902: Test forremove_expiredhas limited coverage.The test only verifies that fresh messages survive expiration. It doesn't verify that old messages are actually removed. Consider enhancing the test by either:
- Using a backdoor to set
deferred_atto a past time, or- Testing with
chrono::Duration::zero()to verify all messages are removed:// Zero duration removes all messages (they're at least 0 nanoseconds old) queue.remove_expired(chrono::Duration::zero()); assert_eq!(queue.len(), 0, "zero duration should remove all messages");🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/main.rs` around lines 3875 - 3902, The test deferred_message_queue_remove_expired_works currently only checks that fresh messages survive; update it to also assert that old messages are removed by either calling queue.remove_expired(chrono::Duration::zero()) and asserting queue.len() == 0, or by mutating the stored DeferredMessage.deferred_at timestamp to a past time (e.g., subtracting >expiration) and then calling DeferredMessageQueue::remove_expired(...) and asserting the message was removed; target symbols: deferred_message_queue_remove_expired_works, DeferredMessageQueue::remove_expired, DeferredMessage (deferred_at) and queue.push/len.
239-256: Consider usingretainpattern fordrain_for()efficiency.The current implementation with index tracking and reverse removal is correct but could be simplified:
- fn drain_for(&mut self, key: &ActiveChannelKey) -> Vec<spacebot::InboundMessage> { - let mut result = Vec::new(); - let mut indices_to_remove = Vec::new(); - - for (i, deferred) in self.messages.iter().enumerate() { - if &deferred.target_key == key { - result.push(deferred.message.clone()); - indices_to_remove.push(i); - } - } - - // Remove in reverse order to maintain correct indices - for i in indices_to_remove.into_iter().rev() { - self.messages.remove(i); - } - - result - } + fn drain_for(&mut self, key: &ActiveChannelKey) -> Vec<spacebot::InboundMessage> { + let mut result = Vec::new(); + self.messages.retain(|deferred| { + if &deferred.target_key == key { + result.push(deferred.message.clone()); + false + } else { + true + } + }); + result + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/main.rs` around lines 239 - 256, The drain_for method can be simplified and made more efficient by replacing manual index collection and reverse removals with a retain-based filter: in drain_for, iterate over self.messages using Vec::retain and push matching deferred.message (cloned) into the result when &deferred.target_key == key, returning false to remove that element from self.messages and true to keep others; keep using the same function name drain_for and fields self.messages, ActiveChannelKey, deferred.target_key and deferred.message so the change is localized and behavior remains identical.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/main.rs`:
- Around line 268-273: Add periodic cleanup to call
DeferredMessageQueue::remove_expired from the main event loop: create a
tokio::time::interval (e.g., cleanup_interval) and add a select! branch that
awaits cleanup_interval.tick(), then call
deferred_messages.remove_expired(max_age) with a chosen max_age (e.g.,
chrono::Duration::hours(24)), capture the length before/after to log how many
expired messages were removed, and place this branch alongside the existing main
select! loop so deferred messages targeting inactive channels don't accumulate
indefinitely.
---
Nitpick comments:
In `@src/main.rs`:
- Around line 2162-2185: You're creating a duplicate ActiveChannelKey: instead
of constructing a new ActiveChannelKey for drain_for(), reuse the previously
created channel_key used for insert(); replace the second
ActiveChannelKey::new(...) call with the existing channel_key variable when
calling deferred_messages.drain_for(&channel_key) and when looking up
active_channels.get(&channel_key). Apply the same fix to the other repeated
pattern that creates a second ActiveChannelKey (the block that mirrors this
behavior later in the file) so both places reuse the originally created
channel_key rather than recreating it.
- Around line 3875-3902: The test deferred_message_queue_remove_expired_works
currently only checks that fresh messages survive; update it to also assert that
old messages are removed by either calling
queue.remove_expired(chrono::Duration::zero()) and asserting queue.len() == 0,
or by mutating the stored DeferredMessage.deferred_at timestamp to a past time
(e.g., subtracting >expiration) and then calling
DeferredMessageQueue::remove_expired(...) and asserting the message was removed;
target symbols: deferred_message_queue_remove_expired_works,
DeferredMessageQueue::remove_expired, DeferredMessage (deferred_at) and
queue.push/len.
- Around line 239-256: The drain_for method can be simplified and made more
efficient by replacing manual index collection and reverse removals with a
retain-based filter: in drain_for, iterate over self.messages using Vec::retain
and push matching deferred.message (cloned) into the result when
&deferred.target_key == key, returning false to remove that element from
self.messages and true to keep others; keep using the same function name
drain_for and fields self.messages, ActiveChannelKey, deferred.target_key and
deferred.message so the change is localized and behavior remains identical.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
Comment on lines
+268
to
+273
| /// Remove messages older than the given duration. | ||
| fn remove_expired(&mut self, max_age: chrono::Duration) { | ||
| let now = chrono::Utc::now(); | ||
| self.messages.retain(|m| now - m.deferred_at < max_age); | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
remove_expired() is defined but never invoked.
The DeferredMessageQueue::remove_expired() method exists but is never called in the main event loop. Deferred messages targeting channels that never become active will accumulate indefinitely, causing gradual memory growth.
Consider adding a periodic cleanup, for example using a tokio::time::interval in the main select! loop:
// In the main loop, add a new branch:
_ = cleanup_interval.tick() => {
let max_age = chrono::Duration::hours(24);
let before = deferred_messages.len();
deferred_messages.remove_expired(max_age);
if before > deferred_messages.len() {
tracing::info!(
removed = before - deferred_messages.len(),
"expired deferred messages cleaned up"
);
}
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/main.rs` around lines 268 - 273, Add periodic cleanup to call
DeferredMessageQueue::remove_expired from the main event loop: create a
tokio::time::interval (e.g., cleanup_interval) and add a select! branch that
awaits cleanup_interval.tick(), then call
deferred_messages.remove_expired(max_age) with a chosen max_age (e.g.,
chrono::Duration::hours(24)), capture the length before/after to log how many
expired messages were removed, and place this branch alongside the existing main
select! loop so deferred messages targeting inactive channels don't accumulate
indefinitely.
![tembo[bot]](https://avatars.githubusercontent.com/in/1223475?s=60&v=4){kind=small}
| } | ||
|
|
||
| /// Drain and return all messages intended for the given channel key. | ||
| fn drain_for(&mut self, key: &ActiveChannelKey) -> Vec<spacebot::InboundMessage> { |
Contributor
There was a problem hiding this comment.
drain_for currently clones InboundMessage and does Vec::remove in a loop (quadratic). You can move out in one pass and rebuild the queue while preserving order.
Suggested change
| fn drain_for(&mut self, key: &ActiveChannelKey) -> Vec<spacebot::InboundMessage> { | |
| fn drain_for(&mut self, key: &ActiveChannelKey) -> Vec<spacebot::InboundMessage> { | |
| let existing = std::mem::take(&mut self.messages); | |
| let mut drained = Vec::new(); | |
| let mut kept = Vec::with_capacity(existing.len()); | |
| for deferred in existing { | |
| if &deferred.target_key == key { | |
| drained.push(deferred.message); | |
| } else { | |
| kept.push(deferred); | |
| } | |
| } | |
| self.messages = kept; | |
| drained | |
| } |
| // SECURITY FIX: Queue the message for the exact target channel | ||
| // instead of delivering it to any active channel. This prevents | ||
| // cron output from leaking to unintended channels. | ||
| deferred_messages.push(target_key, injection.message); |
Contributor
There was a problem hiding this comment.
Since we now queue injections, it’s worth expiring old entries to avoid unbounded growth if a target channel never comes back (you already have remove_expired). A lightweight approach is to do it on push.
Suggested change
| deferred_messages.push(target_key, injection.message); | |
| deferred_messages.push(target_key, injection.message); | |
| deferred_messages.remove_expired(chrono::Duration::hours(24)); |
src/main.rs
Outdated
| queue.remove_expired(chrono::Duration::seconds(60)); | ||
| assert_eq!(queue.len(), 1); | ||
|
|
||
| // Zero duration should remove all messages (they're at least 0 seconds old) |
Contributor
There was a problem hiding this comment.
This test currently stops after the comment; adding an assertion makes sure remove_expired is actually covered.
Suggested change
| // Zero duration should remove all messages (they're at least 0 seconds old) | |
| // Zero duration should remove all messages (they're at least 0 seconds old) | |
| queue.remove_expired(chrono::Duration::seconds(0)); | |
| assert_eq!(queue.len(), 0); | |
| // Note: In practice, we use reasonable expiration like 24 hours |
- Use mem::take pattern in drain_for to avoid clone + quadratic removal - Reuse ActiveChannelKey instead of creating duplicates at both insert sites - Expire deferred messages on push to prevent unbounded growth - Complete remove_expired test with zero-duration assertion - Fix test imports and type mismatches (AgentId is Arc<str>) - cargo fmt Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/main.rs (1)
2563-2587:⚠️ Potential issue | 🟠 MajorClear the broken channel entry when injected delivery errors.
This path drops the injected message on error but keeps
target_keyinactive_channels. That leaves the main loop believing the channel is still live, so later injections keep taking this branch instead of being queued for exact-target delivery.🔧 Suggested recovery pattern
- if let Some(active) = active_channels.get(&target_key) { - if let Err(error) = active.message_tx.send(injection.message).await { + if let Some(message_tx) = active_channels + .get(&target_key) + .map(|channel| channel.message_tx.clone()) + { + if let Err(error) = message_tx.send(injection.message).await { tracing::warn!( %error, conversation_id = %injection.conversation_id, agent_id = %injection.agent_id, - "failed to forward injected message to channel" + "failed to forward injected message to channel; re-queueing for exact target" ); + active_channels.remove(&target_key); + deferred_messages.push(target_key, error.0); } else {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/main.rs` around lines 2563 - 2587, When active.message_tx.send(injection.message).await returns Err, remove the stale entry from active_channels (the ActiveChannelKey entry keyed by target_key) and then enqueue the message into deferred_messages with deferred_messages.push(target_key, injection.message) instead of dropping it; update the tracing::warn log accordingly and ensure the else branch still logs successful delivery — this guarantees failed deliveries clear the broken channel so future injections will be queued for exact-target delivery.
♻️ Duplicate comments (2)
src/main.rs (2)
3931-4023:⚠️ Potential issue | 🔴 CriticalFix the new regression tests so they compile and actually assert the expiry case.
mod testsneedssuper::{ActiveChannelKey, DeferredMessageQueue}, the fixtures should useSome("agent1".into())foragent_id, and the zero-duration branch indeferred_message_queue_remove_expired_works()needs a final assertion. This matches the current CI failures in this block.🧪 Minimal fix
mod tests { - use super::wait_for_startup_warmup_tasks; + use super::{wait_for_startup_warmup_tasks, ActiveChannelKey, DeferredMessageQueue}; use std::future::pending; use std::sync::Arc; use std::time::Duration; @@ - agent_id: Some("agent1".to_string()), + agent_id: Some("agent1".into()), @@ - agent_id: Some("agent1".to_string()), + agent_id: Some("agent1".into()), @@ // Zero duration should remove all messages (they're at least 0 seconds old) - // Note: In practice, we use reasonable expiration like 24 hours + queue.remove_expired(chrono::Duration::seconds(0)); + assert_eq!(queue.len(), 0); + // Note: In practice, we use reasonable expiration like 24 hours } }Run this to confirm the missing imports in the test module and the actual type used by
InboundMessage.agent_id. Expect the first block to show onlywait_for_startup_warmup_tasksimported today, and the second block to show thatagent_idis not aStringfield.#!/bin/bash set -euo pipefail echo "=== tests block ===" sed -n '3859,4024p' src/main.rs echo echo "=== InboundMessage definition ===" rg -n -C3 'struct InboundMessage|agent_id:' src🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/main.rs` around lines 3931 - 4023, The test module is missing imports and has incorrect types and a missing assertion: add super::{ActiveChannelKey, DeferredMessageQueue} to the mod tests imports, change the InboundMessage.agent_id fields in the fixtures to Some("agent1".into()) (use Some(...) not a plain String), and in deferred_message_queue_remove_expired_works() add the final assertion after calling queue.remove_expired(chrono::Duration::zero()) to assert the queue length is 0 so the expiry branch is actually validated.
213-213:⚠️ Potential issue | 🔴 CriticalWire the deferred-queue housekeeping into production code.
deferred_at,remove_expired(),has_for(), andlen()are only used from tests right now, which is why the-D warningsbuild is failing on dead code here. Either run expiry cleanup from the main loop/on enqueue and keep the test-only helpers behind#[cfg(test)], or drop the expiration fields until that path is actually live.Also applies to: 258-272
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/main.rs` at line 213, The deferred-queue fields and helpers (deferred_at, remove_expired(), has_for(), len()) are only used in tests and cause dead-code warnings; either wire expiry cleanup into production by invoking remove_expired() from the queue’s enqueue or main event loop path so those fields/methods are exercised, or mark test-only helpers and fields with #[cfg(test)] and remove their public visibility; specifically, if you want production expiry: call the queue.remove_expired() from the main loop or from the enqueue method that adds deferred items (ensuring deferred_at remains on the struct), otherwise annotate or move deferred_at, remove_expired(), has_for(), and len() behind #[cfg(test)] and remove any unused visibility so the compiler no longer treats them as dead code.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/main.rs`:
- Around line 2215-2237: The code currently calls
deferred_messages.drain_for(&channel_key) and then attempts to send each
message, which loses the entire backlog if any send fails; change the pattern in
the ActiveChannelKey handling (the block using ActiveChannelKey::new,
deferred_messages.drain_for, deferred_count, active_channels.get, and
channel.message_tx.send) to not destructively drain up-front: instead iterate a
non-destructive view or clone of the pending messages for that channel, attempt
to send them one-by-one to channel.message_tx.send, and only remove (or drain)
messages from deferred_messages when their send succeeds; if a send fails, stop
attempting further deliveries and requeue the remaining messages (or leave them
intact) so the backlog can be retried later—apply the same fix to the other
identical block around lines 2484-2506.
---
Outside diff comments:
In `@src/main.rs`:
- Around line 2563-2587: When active.message_tx.send(injection.message).await
returns Err, remove the stale entry from active_channels (the ActiveChannelKey
entry keyed by target_key) and then enqueue the message into deferred_messages
with deferred_messages.push(target_key, injection.message) instead of dropping
it; update the tracing::warn log accordingly and ensure the else branch still
logs successful delivery — this guarantees failed deliveries clear the broken
channel so future injections will be queued for exact-target delivery.
---
Duplicate comments:
In `@src/main.rs`:
- Around line 3931-4023: The test module is missing imports and has incorrect
types and a missing assertion: add super::{ActiveChannelKey,
DeferredMessageQueue} to the mod tests imports, change the
InboundMessage.agent_id fields in the fixtures to Some("agent1".into()) (use
Some(...) not a plain String), and in
deferred_message_queue_remove_expired_works() add the final assertion after
calling queue.remove_expired(chrono::Duration::zero()) to assert the queue
length is 0 so the expiry branch is actually validated.
- Line 213: The deferred-queue fields and helpers (deferred_at,
remove_expired(), has_for(), len()) are only used in tests and cause dead-code
warnings; either wire expiry cleanup into production by invoking
remove_expired() from the queue’s enqueue or main event loop path so those
fields/methods are exercised, or mark test-only helpers and fields with
#[cfg(test)] and remove their public visibility; specifically, if you want
production expiry: call the queue.remove_expired() from the main loop or from
the enqueue method that adds deferred items (ensuring deferred_at remains on the
struct), otherwise annotate or move deferred_at, remove_expired(), has_for(),
and len() behind #[cfg(test)] and remove any unused visibility so the compiler
no longer treats them as dead code.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
Comment on lines
+2215
to
+2237
| // Deliver any deferred messages that were waiting for this channel | ||
| let channel_key = ActiveChannelKey::new(agent_id.clone(), conversation_id.clone()); | ||
| let deferred = deferred_messages.drain_for(&channel_key); | ||
| let deferred_count = deferred.len(); | ||
| if deferred_count > 0 { | ||
| if let Some(channel) = active_channels.get(&channel_key) { | ||
| for message in deferred { | ||
| if let Err(error) = channel.message_tx.send(message).await { | ||
| tracing::warn!( | ||
| %error, | ||
| conversation_id = %conversation_id, | ||
| agent_id = %agent_id, | ||
| "failed to deliver deferred message" | ||
| ); | ||
| } | ||
| } | ||
| tracing::info!( | ||
| conversation_id = %conversation_id, | ||
| agent_id = %agent_id, | ||
| count = deferred_count, | ||
| "delivered deferred messages to newly active channel" | ||
| ); | ||
| } |
Contributor
There was a problem hiding this comment.
Don't drop the backlog if deferred delivery fails.
Both blocks drain the queue before delivery starts. If one message errors here, the current item and every remaining deferred message are already out of deferred_messages, so the exact-target backlog is lost instead of being retried when that channel comes back.
🔧 Suggested recovery pattern
- if let Some(channel) = active_channels.get(&channel_key) {
- for message in deferred {
- if let Err(error) = channel.message_tx.send(message).await {
+ if let Some(message_tx) = active_channels
+ .get(&channel_key)
+ .map(|channel| channel.message_tx.clone())
+ {
+ let mut pending = deferred.into_iter();
+ while let Some(message) = pending.next() {
+ if let Err(error) = message_tx.send(message).await {
tracing::warn!(
%error,
conversation_id = %conversation_id,
agent_id = %agent_id,
"failed to deliver deferred message"
);
+ active_channels.remove(&channel_key);
+ deferred_messages.push(channel_key.clone(), error.0);
+ for message in pending {
+ deferred_messages.push(channel_key.clone(), message);
+ }
+ break;
}
}Also applies to: 2484-2506
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/main.rs` around lines 2215 - 2237, The code currently calls
deferred_messages.drain_for(&channel_key) and then attempts to send each
message, which loses the entire backlog if any send fails; change the pattern in
the ActiveChannelKey handling (the block using ActiveChannelKey::new,
deferred_messages.drain_for, deferred_count, active_channels.get, and
channel.message_tx.send) to not destructively drain up-front: instead iterate a
non-destructive view or clone of the pending messages for that channel, attempt
to send them one-by-one to channel.message_tx.send, and only remove (or drain)
messages from deferred_messages when their send succeeds; if a send fails, stop
attempting further deliveries and requeue the remaining messages (or leave them
intact) so the backlog can be retried later—apply the same fix to the other
identical block around lines 2484-2506.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
src/main.rs (1)
2217-2238:⚠️ Potential issue | 🟠 MajorDon't destructively drain before delivery succeeds.
The same failure mode is still present in both activation paths:
drain_for()removes the backlog before the firstsend().await. Ifmessage_txis already closed here, the failed item and every remaining deferred message are lost instead of being retried when that exact channel comes back. On the first send failure, remove the staleactive_channelsentry and requeue the failed + remaining messages under the sameActiveChannelKey.Also applies to: 2486-2508
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/main.rs` around lines 2217 - 2238, The code currently calls deferred_messages.drain_for(&channel_key) which destructively removes the backlog before any send succeeds; change the logic so you do not permanently remove messages up-front: iterate over a non-destructive snapshot (e.g., clone or peek the collection returned by deferred_messages.get(&channel_key) or similar) and attempt channel.message_tx.send(message).await for each; on the first Err from channel.message_tx.send, remove the stale entry from active_channels (active_channels.remove(&channel_key)) and requeue the failed message plus any remaining messages back into deferred_messages under the same ActiveChannelKey instead of losing them, only draining/deleting the deferred bucket after all sends have succeeded. Ensure the same fix is applied for the other activation path that uses drain_for (the block around lines 2486-2508).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/main.rs`:
- Around line 258-266: The methods DeferredMessageQueue::has_for and
DeferredMessageQueue::len are only used in tests and cause warnings under -D
warnings; gate them with #[cfg(test)] (or remove them) so they aren't compiled
in production. Add #[cfg(test)] immediately above the fn declarations for
has_for(&self, key: &ActiveChannelKey) and len(&self) so those methods (which
reference self.messages and ActiveChannelKey) are only compiled for tests, or
alternatively delete them if they’re unused.
---
Duplicate comments:
In `@src/main.rs`:
- Around line 2217-2238: The code currently calls
deferred_messages.drain_for(&channel_key) which destructively removes the
backlog before any send succeeds; change the logic so you do not permanently
remove messages up-front: iterate over a non-destructive snapshot (e.g., clone
or peek the collection returned by deferred_messages.get(&channel_key) or
similar) and attempt channel.message_tx.send(message).await for each; on the
first Err from channel.message_tx.send, remove the stale entry from
active_channels (active_channels.remove(&channel_key)) and requeue the failed
message plus any remaining messages back into deferred_messages under the same
ActiveChannelKey instead of losing them, only draining/deleting the deferred
bucket after all sends have succeeded. Ensure the same fix is applied for the
other activation path that uses drain_for (the block around lines 2486-2508).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
Comment on lines
+258
to
+266
| /// Check if there are any deferred messages for a specific channel. | ||
| fn has_for(&self, key: &ActiveChannelKey) -> bool { | ||
| self.messages.iter().any(|m| &m.target_key == key) | ||
| } | ||
|
|
||
| /// Get count of deferred messages. | ||
| fn len(&self) -> usize { | ||
| self.messages.len() | ||
| } |
Contributor
There was a problem hiding this comment.
Gate these helpers to tests or CI will keep failing.
cargo check --bin spacebot is currently failing under -D warnings because DeferredMessageQueue::has_for() and DeferredMessageQueue::len() are only referenced from #[cfg(test)] code. Mark them #[cfg(test)] or remove them from the production impl.
🛠️ Minimal fix
- fn has_for(&self, key: &ActiveChannelKey) -> bool {
+ #[cfg(test)]
+ fn has_for(&self, key: &ActiveChannelKey) -> bool {
self.messages.iter().any(|m| &m.target_key == key)
}
/// Get count of deferred messages.
- fn len(&self) -> usize {
+ #[cfg(test)]
+ fn len(&self) -> usize {
self.messages.len()
}🧰 Tools
🪛 GitHub Actions: CI
[error] 259-264: cargo check failed due to dead code: methods has_for and len are never used (-D warnings implies -D dead-code).
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/main.rs` around lines 258 - 266, The methods
DeferredMessageQueue::has_for and DeferredMessageQueue::len are only used in
tests and cause warnings under -D warnings; gate them with #[cfg(test)] (or
remove them) so they aren't compiled in production. Add #[cfg(test)] immediately
above the fn declarations for has_for(&self, key: &ActiveChannelKey) and
len(&self) so those methods (which reference self.messages and ActiveChannelKey)
are only compiled for tests, or alternatively delete them if they’re unused.
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.
Closes #498
Security Vulnerability Summary:
Cron job output meant for private DMs could leak to public channels when the DM channel was inactive. The system would deliver queued messages to any active channel instead of the intended target.
The Fix:
Testing:
cargo check --bin spacebot passes
Regression tests added
This is a security fix that prevents private cron output from being exposed to public channels.
Note
Implementation Summary: This security fix introduces targeted message routing to prevent output leakage across agents. The changes add two new data structures:
ActiveChannelKey(uniquely identifies channels by agent_id + conversation_id pair) andDeferredMessageQueue(queues messages for delivery to their exact target channel). The deferred injection logic now binds each queued message to its specific target, ensuring it's only delivered when that exact channel becomes active. Regression tests validate the fix. Written by Tembo for commit 31e7205.