From af7d911206f80ccc241eb5dd68c124ca057874ea Mon Sep 17 00:00:00 2001 From: kjgbot Date: Thu, 6 Aug 2026 12:00:17 +0200 Subject: [PATCH 1/3] Report unbounded node load as unmeasured --- CHANGELOG.md | 1 + crates/broker/src/fleet_wire.rs | 86 +++++++++++++++++--- crates/broker/src/node_control.rs | 31 ++++++- packages/sdk/src/__tests__/messaging.test.ts | 1 + 4 files changed, 104 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a9ef68e9e..5e2e4b629 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fleet node heartbeats no longer report unbounded capacity as `load: 0`; `load` is omitted when `max_agents` is `0` and remains a measured `[0,1]` utilization ratio when a finite agent limit is configured. - `agent-relay node up` resolves its installed broker through canonical package-manager links and Relay's user install directories, so mise-managed and minimal-`PATH` launches no longer fail when the broker binary is already installed. - `agent-relay node up` warns instead of silently ignoring stored Cloud fleet enrollments when the project workspace pin has no enrolled node id. That combination started the broker in the pinned workspace while the node never heartbeat, leaving the Cloud dashboard and `agent-relay fleet nodes` showing different rosters with no error from either. - `agent-relay cloud enroll` records the enrolled node on the project workspace pin, so `node up` in that repo serves the node it just enrolled. A pin that already names a different node is reported and left untouched rather than repointed. diff --git a/crates/broker/src/fleet_wire.rs b/crates/broker/src/fleet_wire.rs index 9ba34c870..2d0cc8a5b 100644 --- a/crates/broker/src/fleet_wire.rs +++ b/crates/broker/src/fleet_wire.rs @@ -168,11 +168,15 @@ pub struct NodeHeartbeat { pub capabilities: Vec, pub max_agents: u32, pub version: String, + // Capacity utilization is undefined for an unbounded provider + // (`max_agents == 0`). Omit it instead of reporting a false idle `0`. #[serde( - deserialize_with = "deserialize_finite_nonnegative_f64", - serialize_with = "serialize_finite_nonnegative_f64" + default, + deserialize_with = "deserialize_optional_finite_nonnegative_f64", + serialize_with = "serialize_optional_finite_nonnegative_f64", + skip_serializing_if = "Option::is_none" )] - pub load: f64, + pub load: Option, pub active_agents: u32, pub handlers_live: bool, } @@ -322,12 +326,16 @@ where T::deserialize(deserializer).map(Some) } -fn deserialize_finite_nonnegative_f64<'de, D>(deserializer: D) -> Result +fn deserialize_optional_finite_nonnegative_f64<'de, D>( + deserializer: D, +) -> Result, D::Error> where D: Deserializer<'de>, { - let value = f64::deserialize(deserializer)?; - validate_finite_nonnegative_f64(value).map_err(de::Error::custom) + Option::::deserialize(deserializer)? + .map(validate_finite_nonnegative_f64) + .transpose() + .map_err(de::Error::custom) } fn serialize_finite_nonnegative_f64(value: &f64, serializer: S) -> Result @@ -338,6 +346,19 @@ where serializer.serialize_f64(*value) } +fn serialize_optional_finite_nonnegative_f64( + value: &Option, + serializer: S, +) -> Result +where + S: Serializer, +{ + match value { + Some(value) => serialize_finite_nonnegative_f64(value, serializer), + None => serializer.serialize_none(), + } +} + fn validate_finite_nonnegative_f64(value: f64) -> Result { if !value.is_finite() { return Err("load must be finite"); @@ -345,6 +366,9 @@ fn validate_finite_nonnegative_f64(value: f64) -> Result { if value < 0.0 { return Err("load must be nonnegative"); } + if value > 1.0 { + return Err("load must be at most 1"); + } Ok(value) } @@ -633,11 +657,10 @@ pub type RelaycastToBroker = ServerToNode; #[cfg(test)] mod tests { - use serde::de::{value::Error as DeError, IntoDeserializer}; use serde_json::{json, Value}; use super::{ - deserialize_finite_nonnegative_f64, validate_agent_register_reply_data, ActionResult, + validate_agent_register_reply_data, validate_finite_nonnegative_f64, ActionResult, ActionResultError, ActionResultPayload, AgentRegister, BrokerToRelaycast, Deliver, DeliveryMode, Error, FleetCapability, NodeHeartbeat, RelaycastToBroker, Reply, FLEET_WIRE_VERSION, @@ -716,10 +739,25 @@ mod tests { }); assert!(serde_json::from_value::(negative).is_err()); + let over_capacity = json!({ + "type": "node.heartbeat", + "v": 1, + "name": "builder-1", + "node_id": "node_1", + "capabilities": [], + "max_agents": 1, + "version": "relay-broker/test", + "load": 1.1, + "active_agents": 2, + "handlers_live": true + }); + assert!(serde_json::from_value::(over_capacity).is_err()); + for load in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] { - let result: Result = - deserialize_finite_nonnegative_f64(load.into_deserializer()); - assert!(result.is_err(), "expected load {load:?} to be rejected"); + assert!( + validate_finite_nonnegative_f64(load).is_err(), + "expected load {load:?} to be rejected" + ); } let invalid = BrokerToRelaycast::NodeHeartbeat(NodeHeartbeat { @@ -731,7 +769,7 @@ mod tests { capabilities: vec![], max_agents: 1, version: "relay-broker/test".to_string(), - load: f64::INFINITY, + load: Some(f64::INFINITY), active_agents: 0, handlers_live: true, }); @@ -765,7 +803,7 @@ mod tests { }], max_agents: 4, version: "relay-broker/test".to_string(), - load: 0.25, + load: Some(0.25), active_agents: 1, handlers_live: true, }); @@ -799,6 +837,28 @@ mod tests { ); } + #[test] + fn node_heartbeat_omits_unreported_load() { + let msg = BrokerToRelaycast::NodeHeartbeat(NodeHeartbeat { + v: FLEET_WIRE_VERSION, + id: None, + provider: None, + name: "unbounded-builder".to_string(), + node_id: "node_unbounded".to_string(), + capabilities: vec![], + max_agents: 0, + version: "relay-broker/test".to_string(), + load: None, + active_agents: 25, + handlers_live: true, + }); + + let value = serde_json::to_value(msg).unwrap(); + assert_eq!(value.get("load"), None); + assert_eq!(value["active_agents"], 25); + assert_eq!(value["max_agents"], 0); + } + #[test] fn node_register_absent_resume_cursor_serializes_as_null() { let decoded: BrokerToRelaycast = serde_json::from_value(json!({ diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index 501e49abb..64faee49b 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -464,9 +464,9 @@ impl FleetLoadSnapshot { /// time server-side as the single source of truth for liveness. fn heartbeat(&self, node: &NodeRegister) -> NodeHeartbeat { let load = if self.max_agents == 0 { - 0.0 + None } else { - (self.active_agents as f64 / self.max_agents as f64).clamp(0.0, 1.0) + Some((self.active_agents as f64 / self.max_agents as f64).clamp(0.0, 1.0)) }; NodeHeartbeat { v: FLEET_WIRE_VERSION, @@ -3545,6 +3545,33 @@ mod tests { } } + #[test] + fn heartbeat_reports_only_measured_capacity_load() { + let register = build_node_register( + &test_manifest(), + "node-default", + "host-default", + "broker/test", + None, + ); + + let measured = FleetLoadSnapshot { + active_agents: 3, + max_agents: 4, + handlers_live: true, + } + .heartbeat(®ister); + assert_eq!(measured.load, Some(0.75)); + + let unbounded = FleetLoadSnapshot { + active_agents: 25, + max_agents: 0, + handlers_live: true, + } + .heartbeat(®ister); + assert_eq!(unbounded.load, None); + } + #[test] fn load_node_token_round_trips_when_node_and_workspace_match() { let dir = tempfile::tempdir().unwrap(); diff --git a/packages/sdk/src/__tests__/messaging.test.ts b/packages/sdk/src/__tests__/messaging.test.ts index 286146657..f7f01a884 100644 --- a/packages/sdk/src/__tests__/messaging.test.ts +++ b/packages/sdk/src/__tests__/messaging.test.ts @@ -485,6 +485,7 @@ describe('RelaycastMessagingClient', () => { }).repoKeys ).toEqual(['explicit']); expect(toRelayNode({ name: 'builder-6', tags: ['factory'] }).repoKeys).toBeUndefined(); + expect(toRelayNode({ name: 'builder-unbounded', max_agents: 0, load: null }).load).toBeUndefined(); await expect(client.nodes.get('builder-2')).resolves.toMatchObject({ name: 'builder-2', From 7cc7dffb3e43b376b2f6a192e753f74584fa3a41 Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Fri, 7 Aug 2026 17:47:42 +0200 Subject: [PATCH 2/3] fix(broker): add decode-side coverage for unreported load and note deploy-order dependency Extends node_heartbeat_omits_unreported_load with round-trip decode assertions for a heartbeat with load absent entirely and one with an explicit `"load": null`, both of which must decode to `load: None`. Previously only the serialize-side (Rust -> JSON) was covered. Also records the Relaycast #307 deploy-order dependency inline in the NodeHeartbeat field doc, not just the PR description, so it survives merge. Addresses CodeRabbit nitpicks on PR #1445. --- crates/broker/src/fleet_wire.rs | 44 +++++++++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/crates/broker/src/fleet_wire.rs b/crates/broker/src/fleet_wire.rs index 2d0cc8a5b..8926f9123 100644 --- a/crates/broker/src/fleet_wire.rs +++ b/crates/broker/src/fleet_wire.rs @@ -170,6 +170,8 @@ pub struct NodeHeartbeat { pub version: String, // Capacity utilization is undefined for an unbounded provider // (`max_agents == 0`). Omit it instead of reporting a false idle `0`. + // Requires the Relaycast engine to accept an omitted/null `load` before + // this broker version is deployed (relaycast#307). #[serde( default, deserialize_with = "deserialize_optional_finite_nonnegative_f64", @@ -857,6 +859,48 @@ mod tests { assert_eq!(value.get("load"), None); assert_eq!(value["active_agents"], 25); assert_eq!(value["max_agents"], 0); + + // Decode side: a heartbeat with the `load` field absent entirely, and + // one with an explicit `"load": null` (e.g. from a relay that + // round-trips the omitted value), must both decode to `load: None`. + let missing_field = json!({ + "type": "node.heartbeat", + "v": 1, + "name": "unbounded-builder", + "node_id": "node_unbounded", + "capabilities": [], + "max_agents": 0, + "version": "relay-broker/test", + "active_agents": 25, + "handlers_live": true + }); + let decoded: BrokerToRelaycast = serde_json::from_value(missing_field).unwrap(); + match decoded { + BrokerToRelaycast::NodeHeartbeat(hb) => { + assert_eq!(hb.load, None); + assert_eq!(hb.active_agents, 25); + assert_eq!(hb.max_agents, 0); + } + other => panic!("expected NodeHeartbeat, got {other:?}"), + } + + let explicit_null = json!({ + "type": "node.heartbeat", + "v": 1, + "name": "unbounded-builder", + "node_id": "node_unbounded", + "capabilities": [], + "max_agents": 0, + "version": "relay-broker/test", + "load": null, + "active_agents": 25, + "handlers_live": true + }); + let decoded: BrokerToRelaycast = serde_json::from_value(explicit_null).unwrap(); + match decoded { + BrokerToRelaycast::NodeHeartbeat(hb) => assert_eq!(hb.load, None), + other => panic!("expected NodeHeartbeat, got {other:?}"), + } } #[test] From b044376b125751d5c88e24db5a1de7067e5de203 Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Fri, 7 Aug 2026 17:56:25 +0200 Subject: [PATCH 3/3] fix(broker): cover accepted load boundary values 0.0 and 1.0 node_heartbeat_rejects_invalid_loads only proved 1.1 and -0.1 are rejected; it never proved the inclusive boundaries 0.0 and 1.0 validate and decode successfully. Adds both to the existing test. Addresses a CodeRabbit nitpick from the PR #1445 re-review. --- crates/broker/src/fleet_wire.rs | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/crates/broker/src/fleet_wire.rs b/crates/broker/src/fleet_wire.rs index 8926f9123..b44b42e15 100644 --- a/crates/broker/src/fleet_wire.rs +++ b/crates/broker/src/fleet_wire.rs @@ -762,6 +762,32 @@ mod tests { ); } + for load in [0.0, 1.0] { + assert_eq!( + validate_finite_nonnegative_f64(load), + Ok(load), + "expected load {load:?} to be accepted" + ); + let boundary = json!({ + "type": "node.heartbeat", + "v": 1, + "name": "builder-1", + "node_id": "node_1", + "capabilities": [], + "max_agents": 1, + "version": "relay-broker/test", + "load": load, + "active_agents": 1, + "handlers_live": true + }); + let decoded: BrokerToRelaycast = serde_json::from_value(boundary) + .unwrap_or_else(|err| panic!("expected load {load:?} to decode: {err}")); + match decoded { + BrokerToRelaycast::NodeHeartbeat(hb) => assert_eq!(hb.load, Some(load)), + other => panic!("expected NodeHeartbeat, got {other:?}"), + } + } + let invalid = BrokerToRelaycast::NodeHeartbeat(NodeHeartbeat { v: FLEET_WIRE_VERSION, id: None,