fix(broker): recover delivery cursor on resume#1242
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (16)
🚧 Files skipped from review as they are similar to previous changes (12)
📝 WalkthroughWalkthroughThe broker negotiates a delivery-cursor capability, receives an authoritative ACK cursor during agent registration, keys delivery state by immutable agent identity, and seeds resumed delivery while preserving compatibility, strict gap detection, and fan-out behavior. ChangesFleet cursor recovery
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Relaycast
participant AgentRegistration
participant FleetDeliveryBook
participant AgentWorker
Relaycast->>AgentRegistration: return agent_id and delivery_ack_seq
AgentRegistration->>FleetDeliveryBook: bind identity and seed cursor
Relaycast->>AgentWorker: send resumed deliver frame
AgentWorker->>FleetDeliveryBook: observe frame
FleetDeliveryBook-->>AgentWorker: approve and report up_to_seq or reject identity/gap
Possibly related issues
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ 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 |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements a negotiated server-authoritative cursor handshake during agent registration to recover the Fleet delivery cursor after a broker restart. It introduces the relay:delivery-cursor-v1 capability, includes delivery_ack_seq in agent registration replies, and keys delivery cursors by immutable agent IDs rather than reusable names to prevent cursor inheritance issues. Feedback on the changes suggests optimizing the sequence 0 message observation path in crates/broker/src/node_control.rs by consolidating double lookups on self.agents into a single lookup.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if deliver.seq == 0 { | ||
| let up_to_seq = self.agents.get(&deliver.agent).map_or(0, |c| c.up_to_seq); | ||
| let up_to_seq = self | ||
| .agents | ||
| .get(&deliver.agent_id) | ||
| .map_or(0, |c| c.up_to_seq); | ||
| if self | ||
| .agents | ||
| .get(&deliver.agent) | ||
| .get(&deliver.agent_id) | ||
| .is_some_and(|c| c.seen_msg_ids.contains(&deliver.msg_id)) | ||
| { | ||
| return DeliveryDecision::Duplicate { up_to_seq }; | ||
| } | ||
| return DeliveryDecision::Deliver { up_to_seq }; | ||
| } |
There was a problem hiding this comment.
In the hot path of sequence 0 message observation, self.agents.get(&deliver.agent_id) is called twice. We can optimize this to a single lookup to improve performance and efficiency.
if deliver.seq == 0 {
if let Some(cursor) = self.agents.get(&deliver.agent_id) {
let up_to_seq = cursor.up_to_seq;
if cursor.seen_msg_ids.contains(&deliver.msg_id) {
return DeliveryDecision::Duplicate { up_to_seq };
}
return DeliveryDecision::Deliver { up_to_seq };
}
return DeliveryDecision::Deliver { up_to_seq: 0 };
}![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 581a4d3a04
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| let Some(cursor) = self.agents.get(&deliver.agent_id) else { | ||
| return if deliver.seq == 1 { | ||
| DeliveryDecision::Deliver { up_to_seq: 1 } |
There was a problem hiding this comment.
With the cursor map now keyed by deliver.agent_id, a late delivery for a retired identity whose reusable name has already been re-registered misses this map and seq == 1 is accepted as a fresh delivery. The surfacing path still injects by deliver.agent, so in that name-reuse/race scenario the old identity's message can be delivered to the new worker instead of being rejected as stale; please verify the active inventory's agent_id for this name before returning Deliver on a missing cursor.
Useful? React with 👍 / 👎.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
All reported issues were addressed
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/broker/src/node_control.rs (1)
683-690: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winPreserve existing progress when reseeding the same identity.
insertcan replace a higher in-memory cursor with a lower registration cursor and clearsseen_msg_ids. The next valid delivery is then treated as a gap, and seq-0 frames can produce a regressed ACK. Merge with the existing entry and preserve deduplication history.Proposed fix
+ if let Some(cursor) = self.agents.get_mut(&agent_id) { + cursor.agent_name = agent; + cursor.up_to_seq = cursor.up_to_seq.max(up_to_seq); + return; + } self.agents.insert( agent_id, AgentDeliveryCursor { agent_name: agent,Add a regression test that seeds
42, commits43, then reseeds42and still accepts44.🤖 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/broker/src/node_control.rs` around lines 683 - 690, Update the agent registration logic around the self.agents.insert call to merge with an existing AgentDeliveryCursor for the same agent_id instead of replacing it: retain the highest up_to_seq and preserve seen_msg_ids, while initializing a new cursor only when no entry exists. Add a regression test covering seed 42, commit 43, reseed 42, and successful acceptance of 44.
🤖 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.
Outside diff comments:
In `@crates/broker/src/node_control.rs`:
- Around line 683-690: Update the agent registration logic around the
self.agents.insert call to merge with an existing AgentDeliveryCursor for the
same agent_id instead of replacing it: retain the highest up_to_seq and preserve
seen_msg_ids, while initializing a new cursor only when no entry exists. Add a
regression test covering seed 42, commit 43, reseed 42, and successful
acceptance of 44.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 8ba617b4-3e6b-481a-abf3-5d9ae2b1ce8b
📒 Files selected for processing (6)
.agentworkforce/trajectories/compacted/compact_7tjqdu0n483c_2026-07-11.json.agentworkforce/trajectories/compacted/compact_7tjqdu0n483c_2026-07-11.md.agentworkforce/trajectories/compacted/compact_qkzes8r7n4br_2026-07-11.json.agentworkforce/trajectories/compacted/compact_qkzes8r7n4br_2026-07-11.mdcrates/broker/src/node_control.rscrates/broker/src/runtime/fleet.rs
✅ Files skipped from review due to trivial changes (3)
- .agentworkforce/trajectories/compacted/compact_qkzes8r7n4br_2026-07-11.md
- .agentworkforce/trajectories/compacted/compact_7tjqdu0n483c_2026-07-11.json
- .agentworkforce/trajectories/compacted/compact_7tjqdu0n483c_2026-07-11.md
willwashburn
force-pushed
the
fix/1240-delivery-cursor-restart
branch
from
c028f00 to
388f454
Compare
Resolve conflicts from the #1239 node-provider refactor while preserving the #1240 delivery-cursor recovery fix: - node_control.rs: adopt new FleetCapability { global, queue } and NodeRegister { provider } fields on the delivery-cursor capability the broker advertises; keep identity-keyed cursor seeding. - runtime/fleet.rs: keep FleetDeliveryPlan / plan_fleet_delivery and the identity-reject delivery path; drop the FleetSidecarRestartState and fleet spawn helpers that #1239 removed (along with their tests); take main's node_control import surface plus DeliveryDecision. - CHANGELOG.md: keep both Unreleased Fixed bullets alongside main's. Broker lib suite green (672 passed), clippy + fmt clean; all delivery cursor / authoritative-identity tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01F5g6BDS2YxXMQ3X3Fcq5Nf
willwashburn
force-pushed
the
fix/1240-delivery-cursor-restart
branch
from
b07c8ec to
6fec797
Compare
The merge landed the #1240 delivery-cursor bullet under the released 9.2.2 section (git matched surrounding context that main had since moved into published releases). The fix is unshipped, so move it to [Unreleased] > Fixed and restore 9.2.2 to main's version. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01F5g6BDS2YxXMQ3X3Fcq5Nf
The broker now appends the relay:delivery-cursor-v1 capacity marker to its NodeRegister (the #1240 restart-safe mailbox resume negotiation), so both nodes' aggregate capability lists include it. Update the boot/register scenario's expected capability objects to match. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01F5g6BDS2YxXMQ3X3Fcq5Nf
2 participants
Summary
Verification
Fixes #1240.
Requires AgentWorkforce/relaycast#254 to deploy first. Older engines safely omit the cursor and retain strict fresh-session behavior.