From 581a4d3a0401c4171f8c409e2b20bdddd9d34ad7 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Fri, 10 Jul 2026 10:15:16 -0400 Subject: [PATCH 1/3] fix(broker): recover delivery cursor on resume (#1240) --- .../2026-07/traj_sm3f7yqrhz8z/summary.md | 32 ++++ .../2026-07/traj_sm3f7yqrhz8z/trajectory.json | 57 +++++++ CHANGELOG.md | 1 + crates/broker/src/fleet_wire.rs | 18 ++- crates/broker/src/node_control.rs | 145 +++++++++++++++--- crates/broker/src/runtime/api.rs | 1 + crates/broker/src/runtime/fleet.rs | 46 +++++- crates/broker/src/runtime/relaycast_events.rs | 2 + .../fleet-wire/reply.agent_register.json | 1 + crates/broker/tests/fleet_wire_fixtures.rs | 1 + specs/fleet-delivery.md | 20 ++- 11 files changed, 291 insertions(+), 33 deletions(-) create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/summary.md create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/trajectory.json diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/summary.md new file mode 100644 index 000000000..a72755a90 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/summary.md @@ -0,0 +1,32 @@ +# Trajectory: Recover Fleet delivery cursor after broker restart + +> **Status:** ✅ Completed +> **Task:** #1240 +> **Confidence:** 94% +> **Started:** July 10, 2026 at 10:14 AM +> **Completed:** July 10, 2026 at 10:14 AM + +--- + +## Summary + +Negotiated Relaycast delivery cursors during agent registration, keyed broker cursors by immutable agent identity, retained strict gap detection, and added restart/compatibility coverage. + +**Approach:** Standard approach + +--- + +## Key Decisions + +### Use a negotiated server-authoritative cursor handshake +- **Chose:** Use a negotiated server-authoritative cursor handshake +- **Reasoning:** Inferring a cursor from the first replay could skip a genuine gap, while broker-local persistence can become stale if the engine or identity changes. Relaycast returns delivery_ack_seq only after the broker advertises relay:delivery-cursor-v1; Relay keys it to agent_id and retains cursor+1 validation. + +--- + +## Chapters + +### 1. Work +*Agent: default* + +- Use a negotiated server-authoritative cursor handshake: Use a negotiated server-authoritative cursor handshake diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/trajectory.json new file mode 100644 index 000000000..1e4a9903c --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/trajectory.json @@ -0,0 +1,57 @@ +{ + "id": "traj_sm3f7yqrhz8z", + "version": 1, + "task": { + "title": "Recover Fleet delivery cursor after broker restart", + "source": { + "system": "plain", + "id": "#1240" + } + }, + "status": "completed", + "startedAt": "2026-07-10T14:14:10.835Z", + "completedAt": "2026-07-10T14:14:38.759Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-07-10T14:14:16.598Z" + } + ], + "chapters": [ + { + "id": "chap_gd2bddm2qzjw", + "title": "Work", + "agentName": "default", + "startedAt": "2026-07-10T14:14:16.598Z", + "endedAt": "2026-07-10T14:14:38.759Z", + "events": [ + { + "ts": 1783692856599, + "type": "decision", + "content": "Use a negotiated server-authoritative cursor handshake: Use a negotiated server-authoritative cursor handshake", + "raw": { + "question": "Use a negotiated server-authoritative cursor handshake", + "chosen": "Use a negotiated server-authoritative cursor handshake", + "alternatives": [], + "reasoning": "Inferring a cursor from the first replay could skip a genuine gap, while broker-local persistence can become stale if the engine or identity changes. Relaycast returns delivery_ack_seq only after the broker advertises relay:delivery-cursor-v1; Relay keys it to agent_id and retains cursor+1 validation." + }, + "significance": "high" + } + ] + } + ], + "retrospective": { + "summary": "Negotiated Relaycast delivery cursors during agent registration, keyed broker cursors by immutable agent identity, retained strict gap detection, and added restart/compatibility coverage.", + "approach": "Standard approach", + "confidence": 0.94 + }, + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "58ec87cc2699877fe5bf3b8c54a38be13504e0be", + "endRef": "58ec87cc2699877fe5bf3b8c54a38be13504e0be" + } +} \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 7b5818afc..9aa647503 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- `agent-relay-broker` resumes Relaycast mailbox delivery at the server's authoritative per-agent ACK cursor after a broker restart, preserving strict gap detection and legacy node compatibility. - Swift SDK: depending on this repository by git URL no longer fails with `no such module 'Relaycast'` — the root `Package.swift` now declares the `relaycast` dependency `AgentRelaySDK` imports. - `HarnessDriverClient.spawn()` now polls the broker's startup handshake for the full `startupTimeoutMs` budget (default 45s) instead of a fixed ~10s, so a slow-but-healthy Relaycast handshake that keeps answering `503` while warming up is no longer misreported as a spawn failure. - `agent-relay integration subscribe` now resolves provider-native `--resource` values through relayfile before binding, so Slack channel names, GitHub repos, Linear team keys, and Telegram chats bind to matching relayfile VFS globs while explicit `/`-prefixed globs still work. diff --git a/crates/broker/src/fleet_wire.rs b/crates/broker/src/fleet_wire.rs index b23b20a26..d5c857516 100644 --- a/crates/broker/src/fleet_wire.rs +++ b/crates/broker/src/fleet_wire.rs @@ -7,6 +7,10 @@ use serde::{ use serde_json::Value; pub const FLEET_WIRE_VERSION: FleetWireVersion = FleetWireVersion; +/// Node capability that negotiates `delivery_ack_seq` in `agent.register` +/// replies. It is declared as capacity so older engines never materialize it +/// as an invokable action. +pub const DELIVERY_CURSOR_CAPABILITY: &str = "relay:delivery-cursor-v1"; #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub struct FleetWireVersion; @@ -453,6 +457,15 @@ pub struct AgentRegisterReplyData { skip_serializing_if = "Option::is_none" )] pub name: Option, + /// Relaycast's authoritative cumulative delivery cursor for this agent. + /// Present only when the node advertised the cursor-handshake capability; + /// absent keeps replies compatible with older engines. + #[serde( + default, + deserialize_with = "deserialize_optional_presence", + skip_serializing_if = "Option::is_none" + )] + pub delivery_ack_seq: Option, } pub fn validate_agent_register_reply_data( @@ -850,7 +863,8 @@ mod tests { "data": { "agent_id": "agt_1", "token": "at_live_1", - "name": "codex-builder-1" + "name": "codex-builder-1", + "delivery_ack_seq": 42 } })) .unwrap(); @@ -859,6 +873,7 @@ mod tests { assert_eq!(data.agent_id, "agt_1"); assert_eq!(data.token, "at_live_1"); assert_eq!(data.name.as_deref(), Some("codex-builder-1")); + assert_eq!(data.delivery_ack_seq, Some(42)); let without_name = validate_agent_register_reply_data(&json!({ "agent_id": "agt_1", @@ -866,6 +881,7 @@ mod tests { })) .unwrap(); assert_eq!(without_name.name, None); + assert_eq!(without_name.delivery_ack_seq, None); let missing_token = json!({ "agent_id": "agt_1", diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index 98f5f9093..27be49c42 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -471,6 +471,7 @@ pub(crate) struct AgentRegistrationToken { pub(crate) name: String, pub(crate) agent_id: String, pub(crate) token: String, + pub(crate) delivery_ack_seq: Option, } #[derive(Debug)] @@ -548,6 +549,7 @@ impl SeenMsgIds { #[derive(Debug, Default, Clone)] struct AgentDeliveryCursor { + agent_name: String, up_to_seq: u64, seen_msg_ids: SeenMsgIds, } @@ -558,9 +560,26 @@ pub(crate) struct FleetDeliveryBook { } impl FleetDeliveryBook { - #[cfg(test)] - pub(crate) fn seed_ack(&mut self, agent: impl Into, up_to_seq: u64) { - self.agents.entry(agent.into()).or_default().up_to_seq = up_to_seq; + /// Seed a cursor returned by Relaycast for this exact agent identity. + /// + /// The immutable `agent_id` is the key: a later agent reusing the same name + /// must not inherit the old identity's cumulative ACK position. + pub(crate) fn seed_authoritative_cursor( + &mut self, + agent: impl Into, + agent_id: impl Into, + up_to_seq: u64, + ) { + let agent = agent.into(); + self.remove_agent(&agent); + self.agents.insert( + agent_id.into(), + AgentDeliveryCursor { + agent_name: agent, + up_to_seq, + seen_msg_ids: SeenMsgIds::default(), + }, + ); } pub(crate) fn observe(&self, deliver: &Deliver) -> DeliveryDecision { @@ -571,17 +590,20 @@ impl FleetDeliveryBook { // suppression. The cumulative ack reports the current cursor (the engine // ack is monotonic, so re-acking up_to_seq for a seq-0 frame is a no-op). 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 }; } - let Some(cursor) = self.agents.get(&deliver.agent) else { + let Some(cursor) = self.agents.get(&deliver.agent_id) else { return if deliver.seq == 1 { DeliveryDecision::Deliver { up_to_seq: 1 } } else { @@ -612,7 +634,14 @@ impl FleetDeliveryBook { } pub(crate) fn commit_delivered(&mut self, deliver: &Deliver) -> u64 { - let cursor = self.agents.entry(deliver.agent.clone()).or_default(); + let cursor = self + .agents + .entry(deliver.agent_id.clone()) + .or_insert_with(|| AgentDeliveryCursor { + agent_name: deliver.agent.clone(), + ..AgentDeliveryCursor::default() + }); + cursor.agent_name.clone_from(&deliver.agent); // seq:0 fan-out frames never advance the sequence cursor; they are // deduped purely by msg_id. if deliver.seq == 0 { @@ -627,7 +656,7 @@ impl FleetDeliveryBook { } pub(crate) fn remove_agent(&mut self, agent: &str) { - self.agents.remove(agent); + self.agents.retain(|_, cursor| cursor.agent_name != agent); } } @@ -775,6 +804,27 @@ pub(crate) fn build_node_register( default_version: &str, resume_cursor: Option, ) -> NodeRegister { + let mut capabilities = manifest + .capabilities + .iter() + .filter(|capability| capability.name != crate::fleet_wire::DELIVERY_CURSOR_CAPABILITY) + .map(|capability| FleetCapability { + name: capability.name.clone(), + kind: capability.kind.clone(), + metadata: capability.metadata.as_ref().map(|metadata| { + metadata + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect::>() + }), + }) + .collect::>(); + capabilities.push(FleetCapability { + name: crate::fleet_wire::DELIVERY_CURSOR_CAPABILITY.to_string(), + kind: Some("capacity".to_string()), + metadata: None, + }); + NodeRegister { v: FLEET_WIRE_VERSION, id: None, @@ -787,20 +837,7 @@ pub(crate) fn build_node_register( .and_then(non_empty) .unwrap_or(default_node_id) .to_string(), - capabilities: manifest - .capabilities - .iter() - .map(|capability| FleetCapability { - name: capability.name.clone(), - kind: capability.kind.clone(), - metadata: capability.metadata.as_ref().map(|metadata| { - metadata - .iter() - .map(|(key, value)| (key.clone(), value.clone())) - .collect::>() - }), - }) - .collect(), + capabilities, max_agents: manifest.max_agents.unwrap_or(0), tags: manifest.tags.clone().unwrap_or_default(), version: manifest @@ -1484,6 +1521,7 @@ where name: data.name.unwrap_or_else(|| pending.name.clone()), agent_id: data.agent_id, token: data.token, + delivery_ack_seq: data.delivery_ack_seq, }; match pending.reply.send(Ok(token.clone())) { Ok(()) => true, @@ -1828,7 +1866,7 @@ mod tests { #[test] fn delivery_book_allows_seeded_resume_cursor() { let mut book = FleetDeliveryBook::default(); - book.seed_ack("agent-a", 42); + book.seed_authoritative_cursor("agent-a", "agent-a-id", 42); let deliver = Deliver { v: FLEET_WIRE_VERSION, agent: "agent-a".to_string(), @@ -1847,6 +1885,52 @@ mod tests { assert_eq!(book.commit_delivered(&deliver), 43); } + #[test] + fn delivery_book_scopes_authoritative_cursors_to_agent_identity() { + let mut book = FleetDeliveryBook::default(); + book.seed_authoritative_cursor("shared-name", "agent-old", 42); + book.seed_authoritative_cursor("other", "agent-other", 7); + + let resumed = Deliver { + v: FLEET_WIRE_VERSION, + agent: "shared-name".to_string(), + agent_id: "agent-old".to_string(), + delivery_id: "delivery-43".to_string(), + msg_id: "msg-43".to_string(), + seq: 43, + mode: DeliveryMode::Wait, + payload: json!({"text": "resumed"}), + }; + assert_eq!( + book.observe(&resumed), + DeliveryDecision::Deliver { up_to_seq: 43 } + ); + + let reused_name = Deliver { + agent_id: "agent-new".to_string(), + ..resumed.clone() + }; + assert_eq!( + book.observe(&reused_name), + DeliveryDecision::Gap { up_to_seq: 0 }, + "a new identity must not inherit the previous agent's cursor" + ); + + let other = Deliver { + agent: "other".to_string(), + agent_id: "agent-other".to_string(), + delivery_id: "delivery-8".to_string(), + msg_id: "msg-8".to_string(), + seq: 8, + ..resumed + }; + assert_eq!( + book.observe(&other), + DeliveryDecision::Deliver { up_to_seq: 8 }, + "resumed agents recover independently" + ); + } + #[test] fn delivery_book_retries_until_delivery_is_committed() { let mut book = FleetDeliveryBook::default(); @@ -1906,7 +1990,7 @@ mod tests { #[test] fn delivery_book_surfaces_seq_zero_fanout_without_advancing_cursor() { let mut book = FleetDeliveryBook::default(); - book.seed_ack("agent-a", 5); + book.seed_authoritative_cursor("agent-a", "agent-a-id", 5); // A seq:0 fan-out frame (e.g. action.completed) is always surfaced, // bypassing the monotonic-sequence gate, and acks the current cursor. @@ -2069,6 +2153,7 @@ mod tests { name: "agent-a".to_string(), agent_id: "agt-1".to_string(), token: "at_test".to_string(), + delivery_ack_seq: None, } ); } @@ -2324,6 +2409,14 @@ mod tests { Value::String("codex".to_string()) )])) ); + assert_eq!( + register.capabilities.last(), + Some(&FleetCapability { + name: crate::fleet_wire::DELIVERY_CURSOR_CAPABILITY.to_string(), + kind: Some("capacity".to_string()), + metadata: None, + }) + ); } #[tokio::test] @@ -2487,7 +2580,8 @@ mod tests { data: json!({ "name": "agent-a", "agent_id": "agt-1", - "token": "at_test" + "token": "at_test", + "delivery_ack_seq": 42 }), })) .unwrap(), @@ -2542,6 +2636,7 @@ mod tests { name: "agent-a".to_string(), agent_id: "agt-1".to_string(), token: "at_test".to_string(), + delivery_ack_seq: Some(42), } ); command_tx diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 9529f0bf8..33aeb5ba8 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -149,6 +149,7 @@ impl BrokerRuntime { let session_ref = super::fleet::fleet_initial_session_ref(&spec); match super::fleet::register_node_agent_token( fleet_control_tx, + fleet_delivery_book, name.as_str(), None, session_ref, diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 30afcdd0b..82794415e 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -697,6 +697,7 @@ impl BrokerRuntime { &mut self.dedup, &mut self.agent_spawn_count, &self.fleet_control_tx, + &mut self.fleet_delivery_book, &self.fleet_node_name, Some(invoke.invocation_id.clone()), session_ref, @@ -853,6 +854,7 @@ impl BrokerRuntime { ) -> Result { register_node_agent_token( &self.fleet_control_tx, + &mut self.fleet_delivery_book, spec.name.as_str(), invocation_id, session_ref, @@ -1025,6 +1027,7 @@ impl BrokerRuntime { /// the worker MCP never re-registers over HTTP. pub(super) async fn register_node_agent_token( fleet_control_tx: &mpsc::Sender, + fleet_delivery_book: &mut FleetDeliveryBook, name: &str, invocation_id: Option, session_ref: Option, @@ -1044,10 +1047,18 @@ pub(super) async fn register_node_agent_token( }) .await .map_err(|_| "fleet_control_unavailable".to_string())?; - tokio::time::timeout(FLEET_AGENT_REGISTER_TIMEOUT, reply_rx) + let token = tokio::time::timeout(FLEET_AGENT_REGISTER_TIMEOUT, reply_rx) .await .map_err(|_| "agent_register_timeout".to_string())? - .map_err(|_| "agent_register_reply_dropped".to_string())? + .map_err(|_| "agent_register_reply_dropped".to_string())??; + if let Some(up_to_seq) = token.delivery_ack_seq { + fleet_delivery_book.seed_authoritative_cursor( + token.name.clone(), + token.agent_id.clone(), + up_to_seq, + ); + } + Ok(token) } pub(super) async fn publish_fleet_load_snapshot( @@ -1740,7 +1751,7 @@ mod tests { } #[tokio::test] - async fn action_invoke_spawn_forwards_session_ref_and_invocation_into_agent_register() { + async fn action_invoke_spawn_seeds_authoritative_cursor_before_resumed_delivery() { // An `action.invoke` spawn carrying `harnessConfig.session_id` must // forward a non-None session_ref (and the invocation id) into the node // `agent.register` it emits, so the spawn resumes the session and the @@ -1766,7 +1777,16 @@ mod tests { // emitted AgentRegister to confirm both fields are threaded through. let (tx, mut rx) = mpsc::channel::(4); let register_handle = tokio::spawn(async move { - register_node_agent_token(&tx, "agent-a", Some("inv-42".to_string()), session_ref).await + let mut delivery_book = FleetDeliveryBook::default(); + let token = register_node_agent_token( + &tx, + &mut delivery_book, + "agent-a", + Some("inv-42".to_string()), + session_ref, + ) + .await?; + Ok::<_, String>((token, delivery_book)) }); let command = rx.recv().await.expect("register command emitted"); @@ -1784,10 +1804,26 @@ mod tests { name: "agent-a".to_string(), agent_id: "agent-a-id".to_string(), token: "at_test".to_string(), + delivery_ack_seq: Some(42), })) .unwrap(); - let token = register_handle.await.unwrap().unwrap(); + let (token, delivery_book) = register_handle.await.unwrap().unwrap(); assert_eq!(token.token, "at_test"); + + let resumed = Deliver { + v: FLEET_WIRE_VERSION, + agent: "agent-a".to_string(), + agent_id: "agent-a-id".to_string(), + delivery_id: "delivery-43".to_string(), + msg_id: "msg-43".to_string(), + seq: 43, + mode: DeliveryMode::Wait, + payload: json!({"text": "after restart"}), + }; + assert_eq!( + delivery_book.observe(&resumed), + crate::node_control::DeliveryDecision::Deliver { up_to_seq: 43 } + ); } #[test] diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index dade3af3f..55069fe77 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -280,6 +280,7 @@ pub(super) async fn spawn_worker_from_request( dedup: &mut DedupCache, agent_spawn_count: &mut u32, fleet_control_tx: &mpsc::Sender, + fleet_delivery_book: &mut FleetDeliveryBook, node_name: &str, invocation_id: Option, session_ref: Option, @@ -400,6 +401,7 @@ pub(super) async fn spawn_worker_from_request( } else { match super::fleet::register_node_agent_token( fleet_control_tx, + fleet_delivery_book, name.as_str(), invocation_id.clone(), session_ref.clone(), diff --git a/crates/broker/tests/fixtures/fleet-wire/reply.agent_register.json b/crates/broker/tests/fixtures/fleet-wire/reply.agent_register.json index 2132dcfc9..93c237231 100644 --- a/crates/broker/tests/fixtures/fleet-wire/reply.agent_register.json +++ b/crates/broker/tests/fixtures/fleet-wire/reply.agent_register.json @@ -5,6 +5,7 @@ "ok": true, "data": { "agent_id": "agt_01J7FLEET000000000000101", + "delivery_ack_seq": 42, "token": "at_live_0123456789abcdef01234567", "name": "codex-builder-1" } diff --git a/crates/broker/tests/fleet_wire_fixtures.rs b/crates/broker/tests/fleet_wire_fixtures.rs index c9bdd52d0..813647cb1 100644 --- a/crates/broker/tests/fleet_wire_fixtures.rs +++ b/crates/broker/tests/fleet_wire_fixtures.rs @@ -113,6 +113,7 @@ fn fleet_wire_fixtures_round_trip_semantically() { assert_eq!(agent_register.agent_id, "agt_01J7FLEET000000000000101"); assert_eq!(agent_register.token, "at_live_0123456789abcdef01234567"); assert_eq!(agent_register.name.as_deref(), Some("codex-builder-1")); + assert_eq!(agent_register.delivery_ack_seq, Some(42)); } } diff --git a/specs/fleet-delivery.md b/specs/fleet-delivery.md index 356fb4a71..bb9219aa4 100644 --- a/specs/fleet-delivery.md +++ b/specs/fleet-delivery.md @@ -139,7 +139,23 @@ Consequence: persistence _across process death_ is a **resumable-only** property - **Relaycast** holds all durable state (source of truth): mailboxes, agent records (`resumable`, `session_ref`, origin node), locations, node registry. Must survive Relaycast restarts. - **Broker** keeps only in-memory per-session state: `seq` cursor, dedup set, local pending-injection queue. **No disk needed for delivery durability.** - Uplink blip, broker alive → cursor/dedup survive → clean replay, no duplicates. - - Broker process dies → its child agents die too → they respawn and _want_ redelivery → redelivery is correct, not duplicate. + - Broker process dies → its child agents die too. When a resumable identity re-registers, Relaycast returns that identity's authoritative cumulative ACK cursor before replaying pending delivery, so the fresh broker can continue at `cursor + 1` without accepting an arbitrary first sequence. + +#### Cursor recovery handshake + +A broker that understands cursor recovery advertises the reserved node capability +`relay:delivery-cursor-v1` (with `kind: "capacity"`) on every `node.register`. For +that connection only, Relaycast adds `delivery_ack_seq` to a successful +`agent.register` reply. The value belongs to the returned immutable `agent_id`, +not merely its reusable name. Relaycast sends this reply before draining the +agent's pending mailbox; WebSocket ordering therefore establishes the cursor +before the first resumed `deliver` frame. + +The broker accepts a positive sequence only when it is exactly one greater than +the returned cursor. If the capability or reply field is absent, the broker uses +the legacy fresh-session rule (`seq == 1`) and continues rejecting gaps. Relaycast +omits the field for nodes that did not advertise the capability so older brokers +with strict reply decoders remain compatible. ### 8.5 One durable store @@ -157,7 +173,7 @@ The per-agent mailbox **subsumes** any per-node replay buffer: node-disconnect r A node's broker holds one control connection to Relaycast, serving two roles: **compute provider** (advertises capabilities, receives spawn/release action invocations, reports results) and **delivery relay** (receives inbound for the PTY agents located on it, injects, acks). -- **Register** (on connect): node name, capabilities, version, `max_agents`, tags, and a resume cursor for replay. +- **Register** (on connect): node name, capabilities (including negotiated protocol capabilities), version, `max_agents`, tags, and a resume cursor for replay. - **Heartbeat** (~10–15s): `load`, `active_agents`. Relaycast TTL marks offline → stop placing there; mark its located agents unreachable. - **Reconnect inventory sync:** after register, the broker re-announces its full live agent inventory (`agent_id`, name, `invocationId`, `session_ref`). Relaycast reconciles **locations** and open **invocations** (§7) from it. - **Deregister:** graceful on shutdown; else liveness TTL. From 50ae9dc2b53ec2c283eb3e86fa22114b2d43af90 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Fri, 10 Jul 2026 10:13:01 -0400 Subject: [PATCH 2/3] fix(broker): defer manual delivery ACKs (#1241) --- .../2026-07/traj_cbx77hjzvdaz/summary.md | 37 ++++ .../2026-07/traj_cbx77hjzvdaz/trajectory.json | 69 +++++++ CHANGELOG.md | 1 + crates/broker/src/listen_api.rs | 10 +- crates/broker/src/node_control.rs | 188 ++++++++++++++++-- crates/broker/src/runtime/api.rs | 99 +++++---- crates/broker/src/runtime/delivery.rs | 82 +++++--- crates/broker/src/runtime/fleet.rs | 131 +++++++++++- crates/broker/src/runtime/mod.rs | 2 +- crates/broker/src/runtime/tests.rs | 164 +++++++++++++-- crates/broker/src/types.rs | 18 +- specs/fleet-delivery.md | 10 +- 12 files changed, 696 insertions(+), 115 deletions(-) create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/trajectory.json diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md new file mode 100644 index 000000000..4fdc2b516 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md @@ -0,0 +1,37 @@ +# Trajectory: Fix manual-flush Relaycast ACK durability + +> **Status:** ✅ Completed +> **Task:** #1241 +> **Confidence:** 93% +> **Started:** July 10, 2026 at 10:15 AM +> **Completed:** July 10, 2026 at 10:15 AM + +--- + +## Summary + +Fixed #1241 by separating received and ACKed delivery cursors, retaining Relaycast receipts through manual flush, rejecting full queues without eviction, and ACKing only successfully injected FIFO prefixes. Verified focused cursor, replay, overflow, success, and failure regressions; cargo fmt check passed; full broker library suite passed 672 tests with 4 ignored. + +**Approach:** Standard approach + +--- + +## Key Decisions + +### Split delivery progress into received and ACKed cursors +- **Chose:** Split delivery progress into received and ACKed cursors +- **Reasoning:** Manual-flush must accept multiple contiguous Relaycast sequences without cumulatively acknowledging volatile queue entries; duplicates and gaps therefore report only the ACKed cursor, while successful enqueue advances received state. + +### Reject the newest delivery when the manual queue is full +- **Chose:** Reject the newest delivery when the manual queue is full +- **Reasoning:** Evicting an already-held delivery could discard the only actionable copy after a later cumulative ACK. Atomic rejection leaves the FIFO unchanged, does not advance received or ACKed state, and preserves Relaycast ownership for replay. + +--- + +## Chapters + +### 1. Work +*Agent: default* + +- Split delivery progress into received and ACKed cursors: Split delivery progress into received and ACKed cursors +- Reject the newest delivery when the manual queue is full: Reject the newest delivery when the manual queue is full diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/trajectory.json new file mode 100644 index 000000000..19416c81b --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/trajectory.json @@ -0,0 +1,69 @@ +{ + "id": "traj_cbx77hjzvdaz", + "version": 1, + "task": { + "title": "Fix manual-flush Relaycast ACK durability", + "source": { + "system": "plain", + "id": "#1241" + } + }, + "status": "completed", + "startedAt": "2026-07-10T14:15:30.303Z", + "completedAt": "2026-07-10T14:15:39.949Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-07-10T14:15:35.758Z" + } + ], + "chapters": [ + { + "id": "chap_xdlfc8rczkfx", + "title": "Work", + "agentName": "default", + "startedAt": "2026-07-10T14:15:35.758Z", + "endedAt": "2026-07-10T14:15:39.949Z", + "events": [ + { + "ts": 1783692935759, + "type": "decision", + "content": "Split delivery progress into received and ACKed cursors: Split delivery progress into received and ACKed cursors", + "raw": { + "question": "Split delivery progress into received and ACKed cursors", + "chosen": "Split delivery progress into received and ACKed cursors", + "alternatives": [], + "reasoning": "Manual-flush must accept multiple contiguous Relaycast sequences without cumulatively acknowledging volatile queue entries; duplicates and gaps therefore report only the ACKed cursor, while successful enqueue advances received state." + }, + "significance": "high" + }, + { + "ts": 1783692935760, + "type": "decision", + "content": "Reject the newest delivery when the manual queue is full: Reject the newest delivery when the manual queue is full", + "raw": { + "question": "Reject the newest delivery when the manual queue is full", + "chosen": "Reject the newest delivery when the manual queue is full", + "alternatives": [], + "reasoning": "Evicting an already-held delivery could discard the only actionable copy after a later cumulative ACK. Atomic rejection leaves the FIFO unchanged, does not advance received or ACKed state, and preserves Relaycast ownership for replay." + }, + "significance": "high" + } + ] + } + ], + "retrospective": { + "summary": "Fixed #1241 by separating received and ACKed delivery cursors, retaining Relaycast receipts through manual flush, rejecting full queues without eviction, and ACKing only successfully injected FIFO prefixes. Verified focused cursor, replay, overflow, success, and failure regressions; cargo fmt check passed; full broker library suite passed 672 tests with 4 ignored.", + "approach": "Standard approach", + "confidence": 0.93 + }, + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "3cfe0bf69bb24ad966b58ed288813bb3a7ef1a73", + "endRef": "3cfe0bf69bb24ad966b58ed288813bb3a7ef1a73" + } +} diff --git a/CHANGELOG.md b/CHANGELOG.md index 9aa647503..a96287f0d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - `agent-relay-broker` resumes Relaycast mailbox delivery at the server's authoritative per-agent ACK cursor after a broker restart, preserving strict gap detection and legacy node compatibility. +- `agent-relay-broker` no longer acknowledges Relaycast `manual_flush` deliveries while they exist only in volatile memory; flushes ACK only an injected FIFO prefix, and full queues reject new deliveries without evicting held messages. - Swift SDK: depending on this repository by git URL no longer fails with `no such module 'Relaycast'` — the root `Package.swift` now declares the `relaycast` dependency `AgentRelaySDK` imports. - `HarnessDriverClient.spawn()` now polls the broker's startup handshake for the full `startupTimeoutMs` budget (default 45s) instead of a fixed ~10s, so a slow-but-healthy Relaycast handshake that keeps answering `503` while warming up is no longer misreported as a spawn failure. - `agent-relay integration subscribe` now resolves provider-native `--resource` values through relayfile before binding, so Slack channel names, GitHub repos, Linear team keys, and Telegram chats bind to matching relayfile VFS globs while explicit `/`-prefixed globs still work. diff --git a/crates/broker/src/listen_api.rs b/crates/broker/src/listen_api.rs index bf9ac81e0..8efa63170 100644 --- a/crates/broker/src/listen_api.rs +++ b/crates/broker/src/listen_api.rs @@ -1988,10 +1988,10 @@ async fn listen_api_get_pending( /// `POST /api/spawned/{name}/flush` → `{ "flushed": N }`. /// -/// Drains the queue and injects each message into the worker in order -/// using the existing fire-and-forget inject path. The inbound delivery mode is -/// *not* changed — a caller still in `manual_flush` delivery mode will continue -/// to queue newly-arriving messages. +/// Injects queued messages into the worker in FIFO order and stops at the first +/// failure, retaining that message and its suffix for a later attempt. The +/// inbound delivery mode is *not* changed — a caller still in `manual_flush` +/// delivery mode will continue to queue newly-arriving messages. async fn listen_api_flush_pending( axum::extract::State(state): axum::extract::State, axum::extract::Path(name): axum::extract::Path, @@ -4763,6 +4763,7 @@ mod auth_tests { mode: MessageInjectionMode::Steer, queued_at_ms: 100, event_id: Some(EventId::new("evt_1")), + relaycast_receipt: None, }, PendingRelayMessage { from: "Bob".to_string(), @@ -4775,6 +4776,7 @@ mod auth_tests { mode: MessageInjectionMode::Wait, queued_at_ms: 200, event_id: None, + relaycast_receipt: None, }, ])); } diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index 27be49c42..12ce9fa54 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -23,6 +23,7 @@ use crate::{ RelaycastToBroker, FLEET_WIRE_VERSION, }, protocol::{HandlerResult, HandlerResultPayload, NodeManifest}, + types::RelaycastDeliveryReceipt, }; const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(12); @@ -550,7 +551,8 @@ impl SeenMsgIds { #[derive(Debug, Default, Clone)] struct AgentDeliveryCursor { agent_name: String, - up_to_seq: u64, + acked_up_to_seq: u64, + received_up_to_seq: u64, seen_msg_ids: SeenMsgIds, } @@ -576,7 +578,8 @@ impl FleetDeliveryBook { agent_id.into(), AgentDeliveryCursor { agent_name: agent, - up_to_seq, + acked_up_to_seq: up_to_seq, + received_up_to_seq: up_to_seq, seen_msg_ids: SeenMsgIds::default(), }, ); @@ -593,7 +596,7 @@ impl FleetDeliveryBook { let up_to_seq = self .agents .get(&deliver.agent_id) - .map_or(0, |c| c.up_to_seq); + .map_or(0, |c| c.acked_up_to_seq); if self .agents .get(&deliver.agent_id) @@ -612,19 +615,19 @@ impl FleetDeliveryBook { }; if cursor.seen_msg_ids.contains(&deliver.msg_id) { return DeliveryDecision::Duplicate { - up_to_seq: cursor.up_to_seq, + up_to_seq: cursor.acked_up_to_seq, }; } - if deliver.seq <= cursor.up_to_seq { + if deliver.seq <= cursor.received_up_to_seq { return DeliveryDecision::Stale { - up_to_seq: cursor.up_to_seq, + up_to_seq: cursor.acked_up_to_seq, }; } - if deliver.seq != cursor.up_to_seq.saturating_add(1) { + if deliver.seq != cursor.received_up_to_seq.saturating_add(1) { return DeliveryDecision::Gap { - up_to_seq: cursor.up_to_seq, + up_to_seq: cursor.acked_up_to_seq, }; } @@ -633,7 +636,7 @@ impl FleetDeliveryBook { } } - pub(crate) fn commit_delivered(&mut self, deliver: &Deliver) -> u64 { + pub(crate) fn commit_received(&mut self, deliver: &Deliver) -> u64 { let cursor = self .agents .entry(deliver.agent_id.clone()) @@ -642,17 +645,71 @@ impl FleetDeliveryBook { ..AgentDeliveryCursor::default() }); cursor.agent_name.clone_from(&deliver.agent); - // seq:0 fan-out frames never advance the sequence cursor; they are + // seq:0 fan-out frames never advance either sequence cursor; they are // deduped purely by msg_id. if deliver.seq == 0 { cursor.seen_msg_ids.insert(&deliver.msg_id); - return cursor.up_to_seq; + return cursor.received_up_to_seq; } - if deliver.seq == cursor.up_to_seq.saturating_add(1) { + if deliver.seq == cursor.received_up_to_seq.saturating_add(1) { cursor.seen_msg_ids.insert(&deliver.msg_id); - cursor.up_to_seq = deliver.seq; + cursor.received_up_to_seq = deliver.seq; + } + cursor.received_up_to_seq + } + + pub(crate) fn commit_acked_receipt( + &mut self, + receipt: &RelaycastDeliveryReceipt, + ) -> Option { + let cursor = self.agents.get_mut(receipt.agent_id.as_str())?; + cursor.agent_name = receipt.agent.to_string(); + if receipt.seq == 0 { + cursor.seen_msg_ids.insert(receipt.msg_id.as_str()); + return Some(cursor.acked_up_to_seq); } - cursor.up_to_seq + if receipt.seq != cursor.acked_up_to_seq.saturating_add(1) + || receipt.seq > cursor.received_up_to_seq + { + return None; + } + cursor.acked_up_to_seq = receipt.seq; + Some(cursor.acked_up_to_seq) + } + + pub(crate) fn can_ack_receipt(&self, receipt: &RelaycastDeliveryReceipt) -> bool { + let Some(cursor) = self.agents.get(receipt.agent_id.as_str()) else { + return false; + }; + receipt.seq == 0 + || (receipt.seq == cursor.acked_up_to_seq.saturating_add(1) + && receipt.seq <= cursor.received_up_to_seq) + } + + pub(crate) fn commit_delivered(&mut self, deliver: &Deliver) -> u64 { + self.commit_received(deliver); + let receipt = RelaycastDeliveryReceipt { + agent: deliver.agent.clone().into(), + agent_id: deliver.agent_id.clone().into(), + delivery_id: deliver.delivery_id.clone().into(), + msg_id: deliver.msg_id.clone().into(), + seq: deliver.seq, + }; + self.commit_acked_receipt(&receipt) + .unwrap_or_else(|| self.acked_up_to_seq(&deliver.agent_id)) + } + + pub(crate) fn acked_up_to_seq(&self, agent_id: &str) -> u64 { + self.agents + .get(agent_id) + .map_or(0, |cursor| cursor.acked_up_to_seq) + } + + #[cfg(test)] + pub(crate) fn received_up_to_seq(&self, agent_id: &str) -> u64 { + self.agents + .get(agent_id) + .map_or(0, |cursor| cursor.received_up_to_seq) } pub(crate) fn remove_agent(&mut self, agent: &str) { @@ -1961,6 +2018,109 @@ mod tests { ); } + #[test] + fn delivery_book_receives_multiple_sequences_without_acknowledging_them() { + let mut book = FleetDeliveryBook::default(); + let first = Deliver { + v: FLEET_WIRE_VERSION, + agent: "agent-a".to_string(), + agent_id: "agent-a-id".to_string(), + delivery_id: "delivery-1".to_string(), + msg_id: "msg-1".to_string(), + seq: 1, + mode: DeliveryMode::Wait, + payload: json!({"text": "one"}), + }; + let second = Deliver { + delivery_id: "delivery-2".to_string(), + msg_id: "msg-2".to_string(), + seq: 2, + payload: json!({"text": "two"}), + ..first.clone() + }; + + assert_eq!(book.commit_received(&first), 1); + assert_eq!(book.acked_up_to_seq("agent-a-id"), 0); + assert_eq!( + book.observe(&first), + DeliveryDecision::Duplicate { up_to_seq: 0 }, + "a replayed held frame must not be queued twice or ACKed" + ); + assert_eq!( + book.observe(&second), + DeliveryDecision::Deliver { up_to_seq: 2 } + ); + assert_eq!(book.commit_received(&second), 2); + assert_eq!(book.received_up_to_seq("agent-a-id"), 2); + assert_eq!(book.acked_up_to_seq("agent-a-id"), 0); + + let second_receipt = RelaycastDeliveryReceipt { + agent: "agent-a".into(), + agent_id: "agent-a-id".into(), + delivery_id: "delivery-2".into(), + msg_id: "msg-2".into(), + seq: 2, + }; + assert_eq!( + book.commit_acked_receipt(&second_receipt), + None, + "cumulative ACK cannot skip the held first sequence" + ); + + let first_receipt = RelaycastDeliveryReceipt { + agent: "agent-a".into(), + agent_id: "agent-a-id".into(), + delivery_id: "delivery-1".into(), + msg_id: "msg-1".into(), + seq: 1, + }; + assert_eq!(book.commit_acked_receipt(&first_receipt), Some(1)); + assert_eq!(book.commit_acked_receipt(&second_receipt), Some(2)); + } + + #[test] + fn delivery_book_replays_unacked_manual_sequences_after_restart_baseline() { + let mut before_restart = FleetDeliveryBook::default(); + before_restart.seed_authoritative_cursor("agent-a", "agent-a-id", 42); + let first = Deliver { + v: FLEET_WIRE_VERSION, + agent: "agent-a".to_string(), + agent_id: "agent-a-id".to_string(), + delivery_id: "delivery-43".to_string(), + msg_id: "msg-43".to_string(), + seq: 43, + mode: DeliveryMode::Wait, + payload: json!({"text": "held before restart"}), + }; + let second = Deliver { + delivery_id: "delivery-44".to_string(), + msg_id: "msg-44".to_string(), + seq: 44, + ..first.clone() + }; + before_restart.commit_received(&first); + before_restart.commit_received(&second); + assert_eq!(before_restart.received_up_to_seq("agent-a-id"), 44); + assert_eq!(before_restart.acked_up_to_seq("agent-a-id"), 42); + + // Issue #1240 supplies this persisted ACK baseline after a real broker + // restart. Received-only state is intentionally volatile so Relaycast + // can replay every unacknowledged manual delivery. + let mut after_restart = FleetDeliveryBook::default(); + after_restart.seed_authoritative_cursor("agent-a", "agent-a-id", 42); + assert_eq!( + after_restart.observe(&first), + DeliveryDecision::Deliver { up_to_seq: 43 } + ); + after_restart.commit_received(&first); + assert_eq!( + after_restart.observe(&second), + DeliveryDecision::Deliver { up_to_seq: 44 } + ); + after_restart.commit_received(&second); + assert_eq!(after_restart.acked_up_to_seq("agent-a-id"), 42); + } + #[test] fn delivery_book_remove_agent_prunes_cursor_and_msg_ids() { let mut book = FleetDeliveryBook::default(); diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 33aeb5ba8..4c472e8a7 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -1429,52 +1429,68 @@ impl BrokerRuntime { if !workers.has_worker(&name) { let _ = reply.send(Err(DeliveryRouteError::WorkerNotFound(name))); } else { - let entry = delivery_states.entry(name.clone()).or_default(); - let previous = entry.mode; - entry.mode = mode; - let to_flush: Vec = if previous - == InboundDeliveryMode::ManualFlush - && mode == InboundDeliveryMode::AutoInject - { - entry.drain_pending() - } else { - Vec::new() - }; - let flushed = to_flush.len(); - if !to_flush.is_empty() { + let previous = delivery_states.entry(name.clone()).or_default().mode; + let transition_requires_flush = previous == InboundDeliveryMode::ManualFlush + && mode == InboundDeliveryMode::AutoInject; + let flush_result = if transition_requires_flush { tracing::info!( target = "agent_relay::broker", worker = %name, - drained = flushed, "draining pending queue on manual_flush → auto_inject transition" ); - } - for queued in to_flush { - inject_pending_relay_message( + super::fleet::flush_pending_relay_messages( + delivery_states, workers, - pending_deliveries, + fleet_delivery_book, + fleet_control_tx, &name, - &queued, delivery_retry_interval, ) - .await; + .await + } else { + super::fleet::FlushPendingRelayResult::default() + }; + let flushed = flush_result.flushed; + if let Some(error) = flush_result.failure.as_deref() { + tracing::warn!( + target = "agent_relay::broker", + worker = %name, + flushed, + error, + "stopped delivery-mode transition at failed pending message" + ); + } + let actual_mode = if transition_requires_flush && flush_result.failure.is_some() + { + InboundDeliveryMode::ManualFlush + } else { + delivery_states.entry(name.clone()).or_default().mode = mode; + mode + }; + if flushed > 0 { + tracing::info!( + target = "agent_relay::broker", + worker = %name, + drained = flushed, + "drained pending queue on delivery-mode transition" + ); } tracing::info!( target = "agent_relay::broker", worker = %name, previous_mode = previous.as_wire_str(), - mode = mode.as_wire_str(), + mode = actual_mode.as_wire_str(), flushed, "inbound delivery mode updated" ); - if previous != mode { + if previous != actual_mode { let _ = send_event( sdk_out_tx, json!({ "kind":"agent_inbound_delivery_mode_changed", "name":&name, "previous_mode":previous.as_wire_str(), - "mode":mode.as_wire_str(), + "mode":actual_mode.as_wire_str(), }), ) .await; @@ -1491,7 +1507,10 @@ impl BrokerRuntime { ) .await; } - let _ = reply.send(Ok(SetInboundDeliveryModeOk { mode, flushed })); + let _ = reply.send(Ok(SetInboundDeliveryModeOk { + mode: actual_mode, + flushed, + })); } } ListenApiRequest::GetPending { name, reply } => { @@ -1509,11 +1528,16 @@ impl BrokerRuntime { if !workers.has_worker(&name) { let _ = reply.send(Err(DeliveryRouteError::WorkerNotFound(name))); } else { - let to_flush: Vec = delivery_states - .get_mut(&name) - .map(|state| state.drain_pending()) - .unwrap_or_default(); - let flushed = to_flush.len(); + let flush_result = super::fleet::flush_pending_relay_messages( + delivery_states, + workers, + fleet_delivery_book, + fleet_control_tx, + &name, + delivery_retry_interval, + ) + .await; + let flushed = flush_result.flushed; if flushed > 0 { tracing::info!( target = "agent_relay::broker", @@ -1522,15 +1546,14 @@ impl BrokerRuntime { "flushing pending queue on explicit /flush" ); } - for queued in to_flush { - inject_pending_relay_message( - workers, - pending_deliveries, - &name, - &queued, - delivery_retry_interval, - ) - .await; + if let Some(error) = flush_result.failure.as_deref() { + tracing::warn!( + target = "agent_relay::broker", + worker = %name, + flushed, + error, + "stopped explicit flush at failed pending message" + ); } if flushed > 0 { let _ = send_event( diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index 3dc4c4c44..37545fc08 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -400,6 +400,7 @@ fn emit_delivery_read_ack_telemetry( pub(crate) enum InboundQueueOutcome { Queued, DrainNow(Vec), + RejectedFull, WorkerMissing, } @@ -446,6 +447,7 @@ pub(crate) struct InboundContext<'a> { pub(super) priority: u8, pub(super) mode: MessageInjectionMode, pub(super) event_id: Option<&'a str>, + pub(super) relaycast_receipt: Option, } /// Queue an inbound relay message through the per-worker [`InboundDeliveryMode`]. @@ -475,6 +477,21 @@ pub(crate) fn queue_inbound_for_delivery_mode( let state = delivery_states .entry(WorkerName::from(worker_name)) .or_default(); + if state.pending.len() >= crate::types::MAX_PENDING_PER_WORKER { + tracing::warn!( + target = "agent_relay::broker", + worker = %worker_name, + from = %ctx.from, + mode = state.mode.as_wire_str(), + queue_len = state.pending.len(), + max_pending = crate::types::MAX_PENDING_PER_WORKER, + "pending queue full - rejecting newest message" + ); + return InboundQueueResult { + outcome: InboundQueueOutcome::RejectedFull, + evicted_from: None, + }; + } let should_drain = state.should_drain_immediately(); let queued_at_ms = chrono::Utc::now().timestamp_millis().max(0) as u64; let msg = PendingRelayMessage { @@ -488,6 +505,7 @@ pub(crate) fn queue_inbound_for_delivery_mode( mode: ctx.mode, queued_at_ms, event_id: ctx.event_id.map(EventId::from), + relaycast_receipt: ctx.relaycast_receipt, }; let evicted_from = match state.accept_inbound(msg) { InboundDeliveryDispatch::Queued { queue_len } => { @@ -579,38 +597,46 @@ pub(crate) async fn try_inject_pending_relay_message( } } -/// Inject a previously-queued pending relay message into the worker via -/// the existing `queue_and_try_delivery_raw` path. Used by the -/// `/api/spawned/{name}/flush` handler and by the auto-drain on a -/// `manual_flush → auto_inject` transition. Failures are logged but not -/// propagated — the broker treats `flush` as best-effort fire-and-forget -/// the same way `/api/send` does for individual targets. -pub(crate) async fn inject_pending_relay_message( +/// Attempt one PTY injection without transferring ownership to the broker's +/// retry queue. Manual-flush callers keep the original message at the head of +/// their FIFO on failure, along with its Relaycast receipt, so retrying cannot +/// race a second broker-owned copy of the same delivery. +pub(crate) async fn try_inject_pending_relay_message_once( workers: &mut WorkerRegistry, - pending_deliveries: &mut HashMap, worker_name: &str, msg: &PendingRelayMessage, retry_interval: Duration, -) { - let event_id = msg.event_id.as_deref().unwrap_or(""); - if let Err(error) = try_inject_pending_relay_message( - workers, - pending_deliveries, - worker_name, - msg, - retry_interval, - ) - .await - { - tracing::warn!( - target = "agent_relay::broker", - worker = %worker_name, - from = %msg.from, - event_id = %event_id, - error = %error, - "failed to inject pending relay message during flush" - ); - } +) -> Result<()> { + let event_id = msg + .event_id + .clone() + .unwrap_or_else(|| EventId::new(format!("flush_{}", Uuid::new_v4().simple()))); + let delivery_id = msg + .relaycast_receipt + .as_ref() + .map(|receipt| receipt.delivery_id.clone()) + .unwrap_or_else(|| DeliveryId::new(format!("del_{}", Uuid::new_v4().simple()))); + let delivery = RelayDelivery { + delivery_id, + event_id, + workspace_id: msg.workspace_id.clone(), + workspace_alias: msg.workspace_alias.clone(), + from: msg.from.clone(), + target: msg.target.clone(), + body: msg.body.clone(), + thread_id: msg.thread_id.clone(), + priority: Some(msg.priority), + injection_mode: msg.mode.clone(), + }; + + timeout(retry_interval, workers.deliver(worker_name, delivery)) + .await + .map_err(|_| { + anyhow::anyhow!( + "pending relay delivery timed out after {}ms", + retry_interval.as_millis() + ) + })? } #[allow(clippy::too_many_arguments)] diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 82794415e..7cdad9fbc 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -10,6 +10,13 @@ use crate::{ }; const FLEET_AGENT_REGISTER_TIMEOUT: Duration = Duration::from_secs(30); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FleetDeliverySurfaceOutcome { + Acknowledge, + HoldForManualFlush, +} + #[derive(Debug, Clone, Default)] pub(super) struct FleetSidecarRestartState { policy: RestartPolicy, @@ -348,7 +355,13 @@ impl BrokerRuntime { let up_to_seq = match decision { crate::node_control::DeliveryDecision::Deliver { up_to_seq: _ } => { match self.surface_fleet_deliver(&deliver).await { - Ok(()) => self.fleet_delivery_book.commit_delivered(&deliver), + Ok(FleetDeliverySurfaceOutcome::Acknowledge) => { + self.fleet_delivery_book.commit_delivered(&deliver) + } + Ok(FleetDeliverySurfaceOutcome::HoldForManualFlush) => { + self.fleet_delivery_book.commit_received(&deliver); + return; + } Err(error) => { tracing::warn!( target = "relay_broker::fleet", @@ -378,10 +391,13 @@ impl BrokerRuntime { /// Surface a node `deliver` frame by branching on its payload `type`: /// message-class events inject into the recipient worker's PTY; reaction / /// read receipts are acked with a tracing log only (PTY surfacing deferred). - /// An `Ok` return means the delivery may be committed and acked; an `Err` - /// means injection failed and the ack must be withheld so the engine - /// redelivers. - async fn surface_fleet_deliver(&mut self, deliver: &Deliver) -> Result<(), anyhow::Error> { + /// `Acknowledge` means the delivery crossed the PTY injection boundary; + /// `HoldForManualFlush` means it was received into the volatile FIFO but + /// remains owned by Relaycast until a later successful flush. + async fn surface_fleet_deliver( + &mut self, + deliver: &Deliver, + ) -> Result { let payload_type = deliver .payload .get("type") @@ -446,6 +462,13 @@ impl BrokerRuntime { priority, mode: injection_mode, event_id: Some(&deliver.msg_id), + relaycast_receipt: Some(RelaycastDeliveryReceipt { + agent: WorkerName::from(&deliver.agent), + agent_id: AgentId::from(&deliver.agent_id), + delivery_id: DeliveryId::from(&deliver.delivery_id), + msg_id: EventId::from(&deliver.msg_id), + seq: deliver.seq, + }), }, ); if let Some(dropped_from) = &queue_result.evicted_from { @@ -483,7 +506,7 @@ impl BrokerRuntime { }), ) .await; - Ok(()) + Ok(FleetDeliverySurfaceOutcome::HoldForManualFlush) } InboundQueueOutcome::DrainNow(to_drain) => { // Mirrors the HTTP send path: drain may surface older @@ -519,11 +542,18 @@ impl BrokerRuntime { } } } - current_result + current_result.map(|()| FleetDeliverySurfaceOutcome::Acknowledge) } + InboundQueueOutcome::RejectedFull => anyhow::bail!( + "manual delivery queue is full for '{}'; retaining Relaycast ownership", + deliver.agent + ), InboundQueueOutcome::WorkerMissing => { let relay_delivery = self.fleet_relay_delivery(deliver); - self.workers.deliver(&deliver.agent, relay_delivery).await + self.workers + .deliver(&deliver.agent, relay_delivery) + .await + .map(|()| FleetDeliverySurfaceOutcome::Acknowledge) } } } @@ -536,7 +566,7 @@ impl BrokerRuntime { payload_type = %payload_type, "acking node receipt/reaction delivery without PTY surfacing (deferred)" ); - Ok(()) + Ok(FleetDeliverySurfaceOutcome::Acknowledge) } FleetDeliverySurfacing::AckUnknown => { tracing::warn!( @@ -546,7 +576,7 @@ impl BrokerRuntime { payload_type = %payload_type, "acking unrecognized node delivery payload type without surfacing" ); - Ok(()) + Ok(FleetDeliverySurfaceOutcome::Acknowledge) } } } @@ -1018,6 +1048,87 @@ impl BrokerRuntime { } } +#[derive(Debug, Default, PartialEq, Eq)] +pub(super) struct FlushPendingRelayResult { + pub(super) flushed: usize, + pub(super) failure: Option, +} + +/// Inject a worker's held queue in FIFO order. A failed item and every item +/// behind it remain queued. Relaycast ACKs advance only after the corresponding +/// PTY write succeeds, so the emitted cursor is always an injected prefix. +pub(super) async fn flush_pending_relay_messages( + delivery_states: &mut HashMap, + workers: &mut WorkerRegistry, + fleet_delivery_book: &mut FleetDeliveryBook, + fleet_control_tx: &mpsc::Sender, + worker_name: &WorkerName, + retry_interval: Duration, +) -> FlushPendingRelayResult { + let mut result = FlushPendingRelayResult::default(); + + loop { + let next = delivery_states + .get(worker_name) + .and_then(|state| state.pending.front()) + .cloned(); + let Some(queued) = next else { + break; + }; + + if let Some(receipt) = queued.relaycast_receipt.as_ref() { + if !fleet_delivery_book.can_ack_receipt(receipt) { + result.failure = Some(format!( + "delivery sequence {} for '{}' is not the next ACKable receipt", + receipt.seq, receipt.agent + )); + break; + } + } + + if let Err(error) = + try_inject_pending_relay_message_once(workers, worker_name, &queued, retry_interval) + .await + { + result.failure = Some(error.to_string()); + break; + } + + if let Some(receipt) = queued.relaycast_receipt.as_ref() { + let Some(up_to_seq) = fleet_delivery_book.commit_acked_receipt(receipt) else { + result.failure = Some(format!( + "delivery sequence {} for '{}' could not advance the ACK cursor", + receipt.seq, receipt.agent + )); + break; + }; + if let Err(error) = fleet_control_tx + .send(FleetControlCommand::Send(delivery_ack( + receipt.agent.to_string(), + up_to_seq, + ))) + .await + { + tracing::warn!( + target = "relay_broker::fleet", + agent = %receipt.agent, + up_to_seq, + error = %error, + "failed to enqueue delivery ACK after manual flush" + ); + } + } + + let removed = delivery_states + .get_mut(worker_name) + .and_then(|state| state.pending.pop_front()); + debug_assert_eq!(removed.as_ref(), Some(&queued)); + result.flushed += 1; + } + + result +} + /// Bind an agent to this node by sending node-control `agent.register` and /// awaiting the engine reply with the minted agent token. This is the single /// "register agent via node" step both the `/api/spawn` path and the node diff --git a/crates/broker/src/runtime/mod.rs b/crates/broker/src/runtime/mod.rs index 4375bda50..c52c99338 100644 --- a/crates/broker/src/runtime/mod.rs +++ b/crates/broker/src/runtime/mod.rs @@ -48,7 +48,7 @@ use crate::{ telemetry::{ActionSource, TelemetryClient, TelemetryEvent}, types::{ AgentResultMcpConfig, InboundDeliveryDispatch, InboundDeliveryMode, InboundDeliveryState, - PendingRelayMessage, + PendingRelayMessage, RelaycastDeliveryReceipt, }, }; diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 2b6fc4273..abef64454 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -6,10 +6,12 @@ use std::{ time::{Duration, Instant}, }; +use crate::fleet_wire::{BrokerToRelaycast, Deliver, DeliveryMode, FLEET_WIRE_VERSION}; use crate::ids::{ AgentId, ChannelName, DeliveryId, EventId, MessageTarget, WorkerName, WorkspaceAlias, WorkspaceId, }; +use crate::node_control::{FleetControlCommand, FleetDeliveryBook}; use crate::protocol::{ AgentSpec, BrokerEvent, DeliveryReadAckStatus, HarnessReleasePolicy, HeadlessHarnessConfig, HeadlessHarnessDriver, MessageInjectionMode, RelayDelivery, ResolvedHarnessConfig, @@ -51,7 +53,9 @@ use crate::dedup::DedupCache; use crate::relaycast::{ format_worker_preregistration_error, RelaycastHttpClient, RelaycastRegistrationError, WsControl, }; -use crate::types::{InboundDeliveryMode, InboundDeliveryState}; +use crate::types::{ + InboundDeliveryMode, InboundDeliveryState, PendingRelayMessage, RelaycastDeliveryReceipt, +}; use relaycast::ObserverScope; fn env_test_lock() -> &'static Mutex<()> { @@ -126,6 +130,42 @@ fn inbound_ctx<'a>(event_id: &'a str) -> InboundContext<'a> { priority: 1, mode: MessageInjectionMode::Steer, event_id: Some(event_id), + relaycast_receipt: None, + } +} + +fn fleet_deliver(seq: u64) -> Deliver { + Deliver { + v: FLEET_WIRE_VERSION, + agent: "worker-a".to_string(), + agent_id: "agent-worker-a".to_string(), + delivery_id: format!("delivery-{seq}"), + msg_id: format!("message-{seq}"), + seq, + mode: DeliveryMode::Wait, + payload: json!({"type": "message.created", "text": format!("message {seq}")}), + } +} + +fn held_fleet_message(deliver: &Deliver) -> PendingRelayMessage { + PendingRelayMessage { + from: "Alice".to_string(), + body: format!("message {}", deliver.seq), + target: MessageTarget::new("worker-a"), + thread_id: None, + workspace_id: Some(WorkspaceId::new("ws_demo")), + workspace_alias: Some(WorkspaceAlias::new("Demo")), + priority: 2, + mode: MessageInjectionMode::Wait, + queued_at_ms: super::unix_timestamp_millis(), + event_id: Some(EventId::from(&deliver.msg_id)), + relaycast_receipt: Some(RelaycastDeliveryReceipt { + agent: WorkerName::from(&deliver.agent), + agent_id: AgentId::from(&deliver.agent_id), + delivery_id: DeliveryId::from(&deliver.delivery_id), + msg_id: EventId::from(&deliver.msg_id), + seq: deliver.seq, + }), } } @@ -246,7 +286,7 @@ async fn inbound_queue_worker_missing_does_not_create_state() { } #[tokio::test] -async fn inbound_queue_eviction_surfaces_dropped_message() { +async fn inbound_queue_rejects_overflow_without_evicting_held_message() { let worker_name = "worker-a"; let workers = make_worker_registry_with_worker(worker_name).await; let mut delivery_states = HashMap::from([( @@ -264,32 +304,120 @@ async fn inbound_queue_eviction_surfaces_dropped_message() { assert_eq!(result.evicted_from, None); } - let result = queue_inbound_for_delivery_mode( - &mut delivery_states, - &workers, - worker_name, - inbound_ctx("evt_overflow"), - ); + let before = delivery_states + .get(worker_name) + .expect("state should exist") + .pending_snapshot(); + let rejected_deliver = fleet_deliver(1); + let mut rejected_ctx = inbound_ctx("message-1"); + rejected_ctx.relaycast_receipt = held_fleet_message(&rejected_deliver).relaycast_receipt; + let result = + queue_inbound_for_delivery_mode(&mut delivery_states, &workers, worker_name, rejected_ctx); - assert_eq!(result.outcome, InboundQueueOutcome::Queued); - assert_eq!( - result.evicted_from.as_deref(), - Some("Alice"), - "hitting the per-worker cap must surface the evicted sender so callers can emit delivery_dropped" - ); + assert_eq!(result.outcome, InboundQueueOutcome::RejectedFull); + assert_eq!(result.evicted_from, None); assert_eq!( delivery_states .get(worker_name) .expect("state should exist") - .pending_snapshot() - .len(), - crate::types::MAX_PENDING_PER_WORKER, - "queue stays at the cap after eviction" + .pending_snapshot(), + before, + "a full queue must remain byte-for-byte unchanged" ); + let delivery_book = FleetDeliveryBook::default(); + assert_eq!(delivery_book.received_up_to_seq("agent-worker-a"), 0); + assert_eq!(delivery_book.acked_up_to_seq("agent-worker-a"), 0); cleanup_worker_registry(workers).await; } +#[tokio::test] +async fn manual_flush_injects_and_acks_multiple_sequences_in_fifo_order() { + let worker_name = WorkerName::from("worker-a"); + let mut workers = make_worker_registry_with_worker(&worker_name).await; + let first = fleet_deliver(1); + let second = fleet_deliver(2); + let first_message = held_fleet_message(&first); + let second_message = held_fleet_message(&second); + let mut state = InboundDeliveryState::new(InboundDeliveryMode::ManualFlush); + state.accept_inbound(first_message); + state.accept_inbound(second_message); + let mut delivery_states = HashMap::from([(worker_name.clone(), state)]); + let mut delivery_book = FleetDeliveryBook::default(); + delivery_book.commit_received(&first); + delivery_book.commit_received(&second); + let (fleet_control_tx, mut fleet_control_rx) = mpsc::channel(4); + + let result = super::fleet::flush_pending_relay_messages( + &mut delivery_states, + &mut workers, + &mut delivery_book, + &fleet_control_tx, + &worker_name, + Duration::from_secs(1), + ) + .await; + + assert_eq!(result.flushed, 2); + assert_eq!(result.failure, None); + assert!(delivery_states[&worker_name].pending.is_empty()); + assert_eq!(delivery_book.received_up_to_seq("agent-worker-a"), 2); + assert_eq!(delivery_book.acked_up_to_seq("agent-worker-a"), 2); + for expected_seq in [1, 2] { + match fleet_control_rx.recv().await { + Some(FleetControlCommand::Send(BrokerToRelaycast::DeliveryAck(ack))) => { + assert_eq!(ack.agent, worker_name); + assert_eq!(ack.up_to_seq, expected_seq); + } + other => panic!("expected delivery ACK {expected_seq}, got {other:?}"), + } + } + assert!(fleet_control_rx.try_recv().is_err()); + + cleanup_worker_registry(workers).await; +} + +#[tokio::test] +async fn manual_flush_failure_retains_failed_message_and_suffix_without_ack() { + let worker_name = WorkerName::from("worker-a"); + let (worker_event_tx, _worker_event_rx) = mpsc::channel::(4); + let mut workers = WorkerRegistry::new( + worker_event_tx, + Vec::new(), + PathBuf::from("/tmp/agent-relay-broker-tests"), + Instant::now(), + ); + let first = fleet_deliver(1); + let second = fleet_deliver(2); + let expected = vec![held_fleet_message(&first), held_fleet_message(&second)]; + let mut state = InboundDeliveryState::new(InboundDeliveryMode::ManualFlush); + for message in expected.iter().cloned() { + state.accept_inbound(message); + } + let mut delivery_states = HashMap::from([(worker_name.clone(), state)]); + let mut delivery_book = FleetDeliveryBook::default(); + delivery_book.commit_received(&first); + delivery_book.commit_received(&second); + let (fleet_control_tx, mut fleet_control_rx) = mpsc::channel(4); + + let result = super::fleet::flush_pending_relay_messages( + &mut delivery_states, + &mut workers, + &mut delivery_book, + &fleet_control_tx, + &worker_name, + Duration::from_millis(50), + ) + .await; + + assert_eq!(result.flushed, 0); + assert!(result.failure.is_some()); + assert_eq!(delivery_states[&worker_name].pending_snapshot(), expected); + assert_eq!(delivery_book.received_up_to_seq("agent-worker-a"), 2); + assert_eq!(delivery_book.acked_up_to_seq("agent-worker-a"), 0); + assert!(fleet_control_rx.try_recv().is_err()); +} + fn make_pending_delivery(delivery_id: &str, worker: &str) -> PendingDelivery { PendingDelivery { worker_name: WorkerName::from(worker), diff --git a/crates/broker/src/types.rs b/crates/broker/src/types.rs index 5b57b0a98..3a16699a3 100644 --- a/crates/broker/src/types.rs +++ b/crates/broker/src/types.rs @@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use crate::ids::{ - AgentId, EventId, MessageTarget, ThreadId, WorkerName, WorkspaceAlias, WorkspaceId, + AgentId, DeliveryId, EventId, MessageTarget, ThreadId, WorkerName, WorkspaceAlias, WorkspaceId, }; use crate::protocol::MessageInjectionMode; @@ -93,6 +93,20 @@ pub struct PendingRelayMessage { /// telemetry / dedup parity with the auto-inject path. #[serde(default, skip_serializing_if = "Option::is_none")] pub event_id: Option, + /// Relaycast delivery metadata for messages received over node control. + /// Manual-flush messages retain this receipt until PTY injection so the + /// broker can advance the cumulative ACK only across an injected prefix. + #[serde(skip)] + pub relaycast_receipt: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RelaycastDeliveryReceipt { + pub agent: WorkerName, + pub agent_id: AgentId, + pub delivery_id: DeliveryId, + pub msg_id: EventId, + pub seq: u64, } fn default_priority() -> u8 { @@ -320,6 +334,7 @@ mod inbound_delivery_tests { mode: MessageInjectionMode::Wait, queued_at_ms: 0, event_id: None, + relaycast_receipt: None, } } @@ -363,6 +378,7 @@ mod inbound_delivery_tests { mode: MessageInjectionMode::Steer, queued_at_ms: 123_456, event_id: Some(EventId::new("evt_xyz")), + relaycast_receipt: None, }; let mut state = InboundDeliveryState::new(InboundDeliveryMode::ManualFlush); state.accept_inbound(queued.clone()); diff --git a/specs/fleet-delivery.md b/specs/fleet-delivery.md index bb9219aa4..0a21f82e6 100644 --- a/specs/fleet-delivery.md +++ b/specs/fleet-delivery.md @@ -137,7 +137,7 @@ Consequence: persistence _across process death_ is a **resumable-only** property ### 8.4 Where state lives - **Relaycast** holds all durable state (source of truth): mailboxes, agent records (`resumable`, `session_ref`, origin node), locations, node registry. Must survive Relaycast restarts. -- **Broker** keeps only in-memory per-session state: `seq` cursor, dedup set, local pending-injection queue. **No disk needed for delivery durability.** +- **Broker** keeps only in-memory per-session state: separate contiguous `received` and cumulative `acked` sequence cursors, a dedup set, and the local pending-injection queue. **No disk needed for delivery durability.** - Uplink blip, broker alive → cursor/dedup survive → clean replay, no duplicates. - Broker process dies → its child agents die too. When a resumable identity re-registers, Relaycast returns that identity's authoritative cumulative ACK cursor before replaying pending delivery, so the fresh broker can continue at `cursor + 1` without accepting an arbitrary first sequence. @@ -157,6 +157,14 @@ the legacy fresh-session rule (`seq == 1`) and continues rejecting gaps. Relayca omits the field for nodes that did not advertise the capability so older brokers with strict reply decoders remain compatible. +The authoritative cursor initializes `received == acked`. Auto-inject advances +both only after PTY injection. `manual_flush` may advance `received` as it accepts +multiple contiguous frames into its bounded FIFO, but leaves `acked` unchanged; +duplicates and gaps report only the `acked` cursor. A flush advances `acked` only +across the successfully injected FIFO prefix, and a full FIFO rejects the newest +frame without advancing either cursor. If the broker dies, received-only state is +discarded and Relaycast replays from its authoritative ACK cursor on registration. + ### 8.5 One durable store The per-agent mailbox **subsumes** any per-node replay buffer: node-disconnect replay is just "redeliver this node's agents' unacked mail on reconnect." The per-location `seq` + ack is the at-least-once transport on top of the mailbox. From f3f032a8e7f7ba360764e4b75685682d72167ebd Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Fri, 10 Jul 2026 14:45:07 +0000 Subject: [PATCH 3/3] style: auto-format with Prettier --- .../completed/2026-07/traj_cbx77hjzvdaz/summary.md | 5 ++++- .../completed/2026-07/traj_sm3f7yqrhz8z/summary.md | 4 +++- .../completed/2026-07/traj_sm3f7yqrhz8z/trajectory.json | 2 +- 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md index 4fdc2b516..17b30d27a 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md +++ b/.agentworkforce/trajectories/completed/2026-07/traj_cbx77hjzvdaz/summary.md @@ -19,10 +19,12 @@ Fixed #1241 by separating received and ACKed delivery cursors, retaining Relayca ## Key Decisions ### Split delivery progress into received and ACKed cursors + - **Chose:** Split delivery progress into received and ACKed cursors - **Reasoning:** Manual-flush must accept multiple contiguous Relaycast sequences without cumulatively acknowledging volatile queue entries; duplicates and gaps therefore report only the ACKed cursor, while successful enqueue advances received state. ### Reject the newest delivery when the manual queue is full + - **Chose:** Reject the newest delivery when the manual queue is full - **Reasoning:** Evicting an already-held delivery could discard the only actionable copy after a later cumulative ACK. Atomic rejection leaves the FIFO unchanged, does not advance received or ACKed state, and preserves Relaycast ownership for replay. @@ -31,7 +33,8 @@ Fixed #1241 by separating received and ACKed delivery cursors, retaining Relayca ## Chapters ### 1. Work -*Agent: default* + +_Agent: default_ - Split delivery progress into received and ACKed cursors: Split delivery progress into received and ACKed cursors - Reject the newest delivery when the manual queue is full: Reject the newest delivery when the manual queue is full diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/summary.md index a72755a90..47b920082 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/summary.md +++ b/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/summary.md @@ -19,6 +19,7 @@ Negotiated Relaycast delivery cursors during agent registration, keyed broker cu ## Key Decisions ### Use a negotiated server-authoritative cursor handshake + - **Chose:** Use a negotiated server-authoritative cursor handshake - **Reasoning:** Inferring a cursor from the first replay could skip a genuine gap, while broker-local persistence can become stale if the engine or identity changes. Relaycast returns delivery_ack_seq only after the broker advertises relay:delivery-cursor-v1; Relay keys it to agent_id and retains cursor+1 validation. @@ -27,6 +28,7 @@ Negotiated Relaycast delivery cursors during agent registration, keyed broker cu ## Chapters ### 1. Work -*Agent: default* + +_Agent: default_ - Use a negotiated server-authoritative cursor handshake: Use a negotiated server-authoritative cursor handshake diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/trajectory.json index 1e4a9903c..4bd298371 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/trajectory.json +++ b/.agentworkforce/trajectories/completed/2026-07/traj_sm3f7yqrhz8z/trajectory.json @@ -54,4 +54,4 @@ "startRef": "58ec87cc2699877fe5bf3b8c54a38be13504e0be", "endRef": "58ec87cc2699877fe5bf3b8c54a38be13504e0be" } -} \ No newline at end of file +}