diff --git a/.agentworkforce/trajectories/completed/2026-08/traj_ae0vmeog5qmo/summary.md b/.agentworkforce/trajectories/completed/2026-08/traj_ae0vmeog5qmo/summary.md new file mode 100644 index 00000000..3399461d --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-08/traj_ae0vmeog5qmo/summary.md @@ -0,0 +1,41 @@ +# Trajectory: Make node load telemetry honest and explicit + +> **Status:** ✅ Completed +> **Confidence:** 95% +> **Started:** August 6, 2026 at 06:58 AM +> **Completed:** August 6, 2026 at 07:00 AM + +--- + +## Summary + +Made node load unavailable by default, require explicit measurement provenance, keep max_agents=0 consistently unlimited, guard future heartbeat timestamps, and update all SDK/API types with comprehensive tests. + +**Approach:** Standard approach + +--- + +## Key Decisions + +### Require an explicit load_reported signal before trusting numeric node load +- **Chose:** Require an explicit load_reported signal before trusting numeric node load +- **Reasoning:** Every released provider hard-coded numeric placeholders, including finite-capacity providers, so neither legacy zero nor max_agents alone proves a measurement; explicit additive provenance preserves compatibility and honesty. + +### Keep max_agents zero as unlimited and aggregate mixed providers as unlimited +- **Chose:** Keep max_agents zero as unlimited and aggregate mixed providers as unlimited +- **Reasoning:** Placement already treats zero as unlimited, and a single unlimited provider makes the node aggregate unbounded; additive zero previously produced a false finite cap. + +### Reject negative heartbeat ages as fresh +- **Chose:** Reject negative heartbeat ages as fresh +- **Reasoning:** Node timestamps are server-stamped, but persisted future timestamps can still occur; an explicit lower bound prevents future data from bypassing the freshness TTL. + +--- + +## Chapters + +### 1. Work +*Agent: default* + +- Require an explicit load_reported signal before trusting numeric node load: Require an explicit load_reported signal before trusting numeric node load +- Keep max_agents zero as unlimited and aggregate mixed providers as unlimited: Keep max_agents zero as unlimited and aggregate mixed providers as unlimited +- Reject negative heartbeat ages as fresh: Reject negative heartbeat ages as fresh diff --git a/.agentworkforce/trajectories/completed/2026-08/traj_ae0vmeog5qmo/trajectory.json b/.agentworkforce/trajectories/completed/2026-08/traj_ae0vmeog5qmo/trajectory.json new file mode 100644 index 00000000..ff1e6ea2 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-08/traj_ae0vmeog5qmo/trajectory.json @@ -0,0 +1,77 @@ +{ + "id": "traj_ae0vmeog5qmo", + "version": 1, + "task": { + "title": "Make node load telemetry honest and explicit" + }, + "status": "completed", + "startedAt": "2026-08-06T10:58:56.890Z", + "completedAt": "2026-08-06T11:00:17.298Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-08-06T10:59:06.998Z" + } + ], + "chapters": [ + { + "id": "chap_x6t8c9x0k1a9", + "title": "Work", + "agentName": "default", + "startedAt": "2026-08-06T10:59:06.998Z", + "endedAt": "2026-08-06T11:00:17.298Z", + "events": [ + { + "ts": 1786013947002, + "type": "decision", + "content": "Require an explicit load_reported signal before trusting numeric node load: Require an explicit load_reported signal before trusting numeric node load", + "raw": { + "question": "Require an explicit load_reported signal before trusting numeric node load", + "chosen": "Require an explicit load_reported signal before trusting numeric node load", + "alternatives": [], + "reasoning": "Every released provider hard-coded numeric placeholders, including finite-capacity providers, so neither legacy zero nor max_agents alone proves a measurement; explicit additive provenance preserves compatibility and honesty." + }, + "significance": "high" + }, + { + "ts": 1786013948396, + "type": "decision", + "content": "Keep max_agents zero as unlimited and aggregate mixed providers as unlimited: Keep max_agents zero as unlimited and aggregate mixed providers as unlimited", + "raw": { + "question": "Keep max_agents zero as unlimited and aggregate mixed providers as unlimited", + "chosen": "Keep max_agents zero as unlimited and aggregate mixed providers as unlimited", + "alternatives": [], + "reasoning": "Placement already treats zero as unlimited, and a single unlimited provider makes the node aggregate unbounded; additive zero previously produced a false finite cap." + }, + "significance": "high" + }, + { + "ts": 1786013949800, + "type": "decision", + "content": "Reject negative heartbeat ages as fresh: Reject negative heartbeat ages as fresh", + "raw": { + "question": "Reject negative heartbeat ages as fresh", + "chosen": "Reject negative heartbeat ages as fresh", + "alternatives": [], + "reasoning": "Node timestamps are server-stamped, but persisted future timestamps can still occur; an explicit lower bound prevents future data from bypassing the freshness TTL." + }, + "significance": "high" + } + ] + } + ], + "retrospective": { + "summary": "Made node load unavailable by default, require explicit measurement provenance, keep max_agents=0 consistently unlimited, guard future heartbeat timestamps, and update all SDK/API types with comprehensive tests.", + "approach": "Standard approach", + "confidence": 0.95 + }, + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relaycast", + "tags": [], + "_trace": { + "startRef": "f0b084bd7e8e849bfa122c097a23d60cec0a4399", + "endRef": "f0b084bd7e8e849bfa122c097a23d60cec0a4399" + } +} \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c81e106..37acb987 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,12 @@ This project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). Packages without a separate changelog are covered by the cross-package notes below. -## [Unreleased] +## [Unreleased - Major] + +### Fixed + +- Fleet node rosters now return `load: null` until a direct node, or every constituent provider of a broker node, explicitly reports a genuine normalized capacity-utilization measurement. +- Fleet capacity now treats `max_agents: 0` consistently as unlimited. ## [6.3.2] - 2026-08-02 diff --git a/README.md b/README.md index 7850cf30..da0aa1a3 100644 --- a/README.md +++ b/README.md @@ -466,6 +466,10 @@ ephemeral `node.online`, `node.heartbeat`, and `node.offline` events. Each carries a `node` payload matching the `GET /nodes` roster entry (capabilities, tags, `load`, `active_agents`/`max_agents`, `handlers_live`, `last_heartbeat_at`), so a single event fully refreshes a node's row. +`load` is normalized managed-agent capacity utilization and remains `null` +unless a direct node, or every constituent provider of a broker node, +explicitly reports a genuine measurement; +`max_agents: 0` means unlimited capacity, not zero capacity. Nodes are first-class delivery hosts and every agent has a node route. `kind` describes transport (`ws`, `http_push`, or `poll`), `role` describes ownership diff --git a/openapi.yaml b/openapi.yaml index 072f15d2..21ffb6c7 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -732,10 +732,16 @@ components: type: boolean load: type: number + nullable: true + minimum: 0 + maximum: 1 + description: Normalized managed-agent capacity utilization. Null unless a direct node, or every constituent provider of a broker node, explicitly reports a genuine measurement. active_agents: type: integer max_agents: type: integer + minimum: 0 + description: Maximum managed agents. Zero means unlimited capacity. last_heartbeat_at: type: string format: date-time diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md index c1d82fd8..d1e55334 100644 --- a/packages/engine/CHANGELOG.md +++ b/packages/engine/CHANGELOG.md @@ -7,7 +7,15 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Major] + +### Fixed + +- Node heartbeats now accept absent/null `load` and require `load_reported` before trusting a numeric measurement. +- Migration `0034` leaves historical placeholder load values unreported. +- `GET /v1/nodes` now returns null load until a direct node, or every constituent provider of a broker node, reports a genuine measurement. +- Future-dated heartbeats no longer count as fresh. +- Broker capacity remains unlimited when any constituent provider is unbounded. ## [6.3.2] - 2026-08-02 diff --git a/packages/engine/src/__tests__/conformance/node.test.ts b/packages/engine/src/__tests__/conformance/node.test.ts index a90bf3bf..d363c040 100644 --- a/packages/engine/src/__tests__/conformance/node.test.ts +++ b/packages/engine/src/__tests__/conformance/node.test.ts @@ -438,7 +438,7 @@ describe('node adapter conformance', () => { id: string; name: string; capabilities: Array>; - load?: number; + load?: number | null; maxAgents?: number; }, ) { @@ -497,13 +497,107 @@ describe('node adapter conformance', () => { await handle.handleMessage(JSON.stringify({ v: 1, type: 'node.heartbeat', - load: opts.load ?? 0, + ...(typeof opts.load === 'number' ? { load: opts.load, load_reported: true } : {}), active_agents: 0, handlers_live: true, })); return { sock, handle }; } + it('reports placeholder load as unavailable until a finite node explicitly marks it measured', async () => { + const ws = await createWorkspace(stack.app, 'fleet-unreported-load-ws'); + const unbounded = await enrollAndAttachNode(ws, { + id: 'node_unbounded', + name: 'unbounded', + capabilities: [capability('spawn:codex', 'spawn')], + maxAgents: 0, + load: null, + }); + + const roster = await stack.app.request('/v1/nodes?name=unbounded', { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + expect(roster.status).toBe(200); + const body = await roster.json() as { data: Array> }; + expect(body.data[0]).toMatchObject({ + name: 'unbounded', + load: null, + active_agents: 0, + max_agents: 0, + }); + + const [stored] = await stack.runtime.handle.db + .select({ load: nodes.load, loadReported: nodes.loadReported }) + .from(nodes) + .where(and(eq(nodes.workspaceId, ws.workspaceId), eq(nodes.id, 'node_unbounded'))); + expect(stored).toEqual({ load: 0, loadReported: false }); + + // An unbounded node cannot report normalized utilization because it has + // no finite denominator, even when a client incorrectly claims it did. + await unbounded.handle.handleMessage(JSON.stringify({ + v: 1, + type: 'node.heartbeat', + load: 0.75, + load_reported: true, + active_agents: 25, + handlers_live: true, + })); + const explicitlyReportedRoster = await stack.app.request('/v1/nodes?name=unbounded', { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + const explicitlyReportedBody = await explicitlyReportedRoster.json() as { data: Array> }; + expect(explicitlyReportedBody.data[0]).toMatchObject({ load: null, active_agents: 25, max_agents: 0 }); + + // Older brokers sent a literal zero for the same unbounded state. The + // engine knows the denominator is absent and must keep treating it as + // unreported during the rolling upgrade. + await unbounded.handle.handleMessage(JSON.stringify({ + v: 1, + type: 'node.heartbeat', + load: 0, + active_agents: 25, + handlers_live: true, + })); + const legacyRoster = await stack.app.request('/v1/nodes?name=unbounded', { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + const legacyBody = await legacyRoster.json() as { data: Array> }; + expect(legacyBody.data[0]).toMatchObject({ load: null, active_agents: 25, max_agents: 0 }); + + const finite = await enrollAndAttachNode(ws, { + id: 'node_finite', + name: 'finite', + capabilities: [capability('spawn:claude', 'spawn')], + maxAgents: 4, + }); + await finite.handle.handleMessage(JSON.stringify({ + v: 1, + type: 'node.heartbeat', + load: 0, + active_agents: 0, + handlers_live: true, + })); + let finiteRoster = await stack.app.request('/v1/nodes?name=finite', { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + let finiteBody = await finiteRoster.json() as { data: Array> }; + expect(finiteBody.data[0]).toMatchObject({ load: null, max_agents: 4 }); + + await finite.handle.handleMessage(JSON.stringify({ + v: 1, + type: 'node.heartbeat', + load: 0, + load_reported: true, + active_agents: 0, + handlers_live: true, + })); + finiteRoster = await stack.app.request('/v1/nodes?name=finite', { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + finiteBody = await finiteRoster.json() as { data: Array> }; + expect(finiteBody.data[0]).toMatchObject({ load: 0, max_agents: 4 }); + }); + it('drives node control directly without the websocket route wrapper', async () => { const ws = await createWorkspace(stack.app, 'node-control-direct-dispatch'); const db = stack.runtime.handle.db; @@ -586,6 +680,7 @@ describe('node adapter conformance', () => { id: 'control-broker-heartbeat', type: 'node.heartbeat', load: 0.25, + load_reported: true, active_agents: 1, handlers_live: true, node_id: 'node_control_broker', @@ -671,6 +766,7 @@ describe('node adapter conformance', () => { id: 'control-direct-heartbeat', type: 'node.heartbeat', load: 0.5, + load_reported: true, active_agents: 7, handlers_live: true, node_id: 'node_control_direct', @@ -1692,6 +1788,7 @@ describe('node adapter conformance', () => { v: 1, type: 'node.heartbeat', load: 0.5, + load_reported: true, active_agents: 1, handlers_live: true, node_id: 'node_alpha', @@ -1749,6 +1846,7 @@ describe('node adapter conformance', () => { v: 1, type: 'node.heartbeat', load: 0.1, + load_reported: true, active_agents: 0, handlers_live: true, })); diff --git a/packages/engine/src/__tests__/conformance/nodeLiveness.test.ts b/packages/engine/src/__tests__/conformance/nodeLiveness.test.ts new file mode 100644 index 00000000..ac886bd4 --- /dev/null +++ b/packages/engine/src/__tests__/conformance/nodeLiveness.test.ts @@ -0,0 +1,19 @@ +import { describe, expect, it } from 'vitest'; +import { isProviderLive } from '../../engine/nodeProvider.js'; +import { isNodeLive, NODE_LIVENESS_TTL_MS } from '../../engine/placement.js'; + +const NOW = 1_700_000_000_000; + +describe('node heartbeat freshness', () => { + it('accepts a recent server timestamp and rejects stale or future timestamps', () => { + expect(isNodeLive({ status: 'online', lastHeartbeatAt: new Date(NOW - 1_000) }, NOW)).toBe(true); + expect(isNodeLive({ status: 'online', lastHeartbeatAt: new Date(NOW - NODE_LIVENESS_TTL_MS - 1) }, NOW)).toBe(false); + expect(isNodeLive({ status: 'online', lastHeartbeatAt: new Date(NOW + 1) }, NOW)).toBe(false); + }); + + it('applies the same explicit negative-age guard to provider freshness', () => { + const provider = { status: 'online', handlersLive: true, lastHeartbeatAt: new Date(NOW + 1) }; + expect(isProviderLive(provider, NOW)).toBe(false); + expect(isProviderLive({ ...provider, lastHeartbeatAt: new Date(NOW) }, NOW)).toBe(true); + }); +}); diff --git a/packages/engine/src/__tests__/conformance/nodeProviders.test.ts b/packages/engine/src/__tests__/conformance/nodeProviders.test.ts index f34b1135..c3a38210 100644 --- a/packages/engine/src/__tests__/conformance/nodeProviders.test.ts +++ b/packages/engine/src/__tests__/conformance/nodeProviders.test.ts @@ -62,17 +62,55 @@ describe('node providers', () => { nodeName: string, providerName: string | undefined, capabilities: Cap[], - opts: { instanceId?: string; maxAgents?: number } = {}, + opts: { instanceId?: string; maxAgents?: number; load?: number | null } = {}, ) { const provider = providerName ? { name: providerName, instance_id: opts.instanceId ?? `${providerName}-i1` } : undefined; const { sock, handle } = attachSocket(workspaceId, nodeId); await handle.handleMessage(registerFrame(nodeId, nodeName, provider, capabilities, opts.maxAgents)); await handle.handleMessage(JSON.stringify({ - v: 1, type: 'node.heartbeat', ...(provider ? { provider } : {}), load: 0, active_agents: 0, handlers_live: true, + v: 1, + type: 'node.heartbeat', + ...(provider ? { provider } : {}), + ...(typeof opts.load === 'number' ? { load: opts.load, load_reported: true } : {}), + active_agents: 0, + handlers_live: true, })); return { sock, handle }; } + it('keeps a mixed finite and unbounded provider aggregate unlimited with unreported load', async () => { + const ws = await createWorkspace(stack.app, 'np-unbounded-load'); + await enrollNode(ws, 'node_a', 'alpha'); + await attachProvider( + ws.workspaceId, + 'node_a', + 'alpha', + 'finite', + [{ name: 'run-etl', kind: 'action' }], + { maxAgents: 4, load: 0.5 }, + ); + await attachProvider( + ws.workspaceId, + 'node_a', + 'alpha', + 'unbounded', + [{ name: 'spawn:codex', kind: 'capacity' }], + { maxAgents: 0, load: 0 }, + ); + + const [node] = await stack.runtime.handle.db + .select({ maxAgents: nodes.maxAgents, load: nodes.load, loadReported: nodes.loadReported }) + .from(nodes) + .where(and(eq(nodes.workspaceId, ws.workspaceId), eq(nodes.id, 'node_a'))); + expect(node).toEqual({ maxAgents: 0, load: 0, loadReported: false }); + + const roster = await stack.app.request('/v1/nodes?name=alpha', { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + const body = await roster.json() as { data: Array> }; + expect(body.data[0]).toMatchObject({ max_agents: 0, load: null }); + }); + it('keys a registration with no provider field to the synthetic default provider', async () => { const ws = await createWorkspace(stack.app, 'np-default'); await enrollNode(ws, 'node_a', 'alpha'); diff --git a/packages/engine/src/db/migrations/0034_node_load_reporting.sql b/packages/engine/src/db/migrations/0034_node_load_reporting.sql new file mode 100644 index 00000000..4f3d6e24 --- /dev/null +++ b/packages/engine/src/db/migrations/0034_node_load_reporting.sql @@ -0,0 +1,23 @@ +-- A numeric zero must mean measured idle, never "no measurement available". +-- Keep the existing numeric columns for SQLite/D1 compatibility and carry +-- measurement presence explicitly alongside them. +ALTER TABLE nodes ADD COLUMN load_reported INTEGER NOT NULL DEFAULT 0; +ALTER TABLE node_providers ADD COLUMN load_reported INTEGER NOT NULL DEFAULT 0; + +-- Do not backfill load_reported. Every released provider implementation sent a +-- placeholder load value, including finite-capacity providers, so no historic +-- numeric value is known to be a measurement. + +-- Provider capacity uses the same sentinel: any unbounded provider makes its +-- aggregate broker node unbounded. Correct aggregates written by the prior +-- additive-zero behavior before deciding whether their load was measured. +UPDATE nodes +SET max_agents = 0 +WHERE role = 'broker' + AND EXISTS ( + SELECT 1 + FROM node_providers + WHERE node_providers.workspace_id = nodes.workspace_id + AND node_providers.node_id = nodes.id + AND node_providers.max_agents = 0 +); diff --git a/packages/engine/src/db/schema.ts b/packages/engine/src/db/schema.ts index 27cf3a9c..6aaea9e1 100644 --- a/packages/engine/src/db/schema.ts +++ b/packages/engine/src/db/schema.ts @@ -111,6 +111,7 @@ export const nodes = sqliteTable( deliveryAdapter: text('delivery_adapter').notNull().default('ws.node.v1'), deliveryConfig: text('delivery_config', { mode: 'json' }).$type>(), capabilities: text('capabilities', { mode: 'json' }).$type().notNull().default([]), + // Zero is the fleet-wide sentinel for unlimited capacity. maxAgents: integer('max_agents').notNull().default(0), activeAgents: integer('active_agents').notNull().default(0), reservedAgents: integer('reserved_agents').notNull().default(0), @@ -119,6 +120,7 @@ export const nodes = sqliteTable( status: text('status').notNull().default('offline'), handlersLive: integer('handlers_live', { mode: 'boolean' }).notNull().default(false), load: real('load').notNull().default(0), + loadReported: integer('load_reported', { mode: 'boolean' }).notNull().default(false), lastHeartbeatAt: integer('last_heartbeat_at', { mode: 'timestamp' }), createdAt: integer('created_at', { mode: 'timestamp' }).notNull().default(sql`(unixepoch())`), }, @@ -155,9 +157,11 @@ export const nodeProviders = sqliteTable( name: text('name').notNull(), instanceId: text('instance_id').notNull(), capabilities: text('capabilities', { mode: 'json' }).$type().notNull().default([]), + // Zero is the fleet-wide sentinel for unlimited provider capacity. maxAgents: integer('max_agents').notNull().default(0), activeAgents: integer('active_agents').notNull().default(0), load: real('load').notNull().default(0), + loadReported: integer('load_reported', { mode: 'boolean' }).notNull().default(false), handlersLive: integer('handlers_live', { mode: 'boolean' }).notNull().default(false), status: text('status').notNull().default('offline'), version: text('version').notNull().default('unknown'), diff --git a/packages/engine/src/engine/node.ts b/packages/engine/src/engine/node.ts index 3ce93381..51cc85f0 100644 --- a/packages/engine/src/engine/node.ts +++ b/packages/engine/src/engine/node.ts @@ -124,7 +124,7 @@ function publicNode(row: NodeRow) { status: live ? 'online' : 'offline', live, handlers_live: live && row.handlersLive, - load: row.load, + load: row.loadReported ? row.load : null, active_agents: row.activeAgents, max_agents: row.maxAgents, last_heartbeat_at: row.lastHeartbeatAt?.toISOString() ?? null, @@ -263,6 +263,8 @@ export async function createNodeToken( version: data.version ?? existing.version, status: 'offline', handlersLive: false, + load: 0, + loadReported: false, }) .where(eq(nodes.id, existing.id)) .returning(); @@ -287,6 +289,7 @@ export async function createNodeToken( status: 'offline', handlersLive: false, load: 0, + loadReported: false, activeAgents: 0, createdAt: now, }) @@ -383,6 +386,8 @@ export async function registerNode( version: message.version, status: 'online', handlersLive: false, + load: 0, + loadReported: false, lastHeartbeatAt: now, }) .where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, authenticatedNodeId))) @@ -454,7 +459,8 @@ export async function heartbeatNode( .set({ ...rosterUpdate, status: 'online', - load: message.load, + load: message.load_reported === true && typeof message.load === 'number' ? message.load : 0, + loadReported: message.load_reported === true && typeof message.load === 'number', activeAgents: 1, handlersLive: false, lastHeartbeatAt: new Date(), @@ -485,6 +491,7 @@ export async function heartbeatNode( }); await heartbeatProvider(tx, workspaceId, nodeId, providerName, { load: message.load, + loadReported: message.load_reported, activeAgents: message.active_agents, handlersLive: message.handlers_live, }); @@ -492,6 +499,7 @@ export async function heartbeatNode( } else { await heartbeatProvider(tx, workspaceId, nodeId, providerName, { load: message.load, + loadReported: message.load_reported, activeAgents: message.active_agents, handlersLive: message.handlers_live, }); @@ -530,6 +538,7 @@ export async function markNodeOffline( status: 'offline', handlersLive: false, load: 0, + loadReported: false, activeAgents: 0, lastHeartbeatAt: new Date(), }) @@ -540,7 +549,7 @@ export async function markNodeOffline( // recomputeNodeAggregate never resurrects a dropped provider's agent count. await db .update(nodeProviders) - .set({ status: 'offline', handlersLive: false, load: 0, activeAgents: 0, lastHeartbeatAt: new Date() }) + .set({ status: 'offline', handlersLive: false, load: 0, loadReported: false, activeAgents: 0, lastHeartbeatAt: new Date() }) .where(and(eq(nodeProviders.workspaceId, workspaceId), eq(nodeProviders.nodeId, nodeId))); await db @@ -843,6 +852,7 @@ async function ensureDirectNodeForAgentInTx( version: 'implicit', handlersLive: false, load: 0, + loadReported: false, }; if (opts.online) { update.status = 'online'; @@ -888,6 +898,7 @@ async function ensureDirectNodeForAgentInTx( status: opts.online ? 'online' : 'offline', handlersLive: false, load: 0, + loadReported: false, lastHeartbeatAt: opts.online ? now : null, createdAt: now, }) @@ -909,6 +920,7 @@ async function ensureDirectNodeForAgentInTx( version: 'implicit', handlersLive: false, load: 0, + loadReported: false, }; if (opts.online) { update.status = 'online'; @@ -955,6 +967,7 @@ export async function markDirectNodeOfflineForAgent( status: 'offline', handlersLive: false, load: 0, + loadReported: false, lastHeartbeatAt: new Date(), }) .where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, nodeId))); diff --git a/packages/engine/src/engine/nodeProvider.ts b/packages/engine/src/engine/nodeProvider.ts index 3064dd26..6b7c80b6 100644 --- a/packages/engine/src/engine/nodeProvider.ts +++ b/packages/engine/src/engine/nodeProvider.ts @@ -34,11 +34,13 @@ export function isProviderLive( provider: Pick, now = Date.now(), ): boolean { + const age = provider.lastHeartbeatAt ? now - provider.lastHeartbeatAt.getTime() : null; return ( provider.status === 'online' && provider.handlersLive && - !!provider.lastHeartbeatAt && - now - provider.lastHeartbeatAt.getTime() <= NODE_LIVENESS_TTL_MS + age !== null && + age >= 0 && + age <= NODE_LIVENESS_TTL_MS ); } @@ -105,6 +107,8 @@ export async function upsertProvider( capabilities: data.capabilities, maxAgents: data.maxAgents, version: data.version, + load: 0, + loadReported: false, handlersLive: data.handlersLive, status: 'online', lastHeartbeatAt: now, @@ -122,6 +126,7 @@ export async function upsertProvider( maxAgents: data.maxAgents, activeAgents: 0, load: 0, + loadReported: false, handlersLive: data.handlersLive, status: 'online', version: data.version, @@ -135,12 +140,22 @@ export async function heartbeatProvider( workspaceId: string, nodeId: string, name: string, - data: { load: number; activeAgents: number; handlersLive: boolean }, + data: { load?: number | null; loadReported?: boolean; activeAgents: number; handlersLive: boolean }, ): Promise { + const provider = await getProvider(db, workspaceId, nodeId, name); + // Every released provider sent placeholder zeroes. Only the additive + // load_reported signal can turn a numeric value into a measurement. + const reportedLoad = provider !== null + && provider.maxAgents > 0 + && data.loadReported === true + && typeof data.load === 'number' + ? data.load + : undefined; await db .update(nodeProviders) .set({ - load: data.load, + load: reportedLoad ?? 0, + loadReported: reportedLoad !== undefined, activeAgents: data.activeAgents, handlersLive: data.handlersLive, status: 'online', @@ -159,7 +174,7 @@ export async function markProviderOffline(db: Db, workspaceId: string, nodeId: s // on the node. A reconnect/heartbeat repopulates it from the provider's report. await db .update(nodeProviders) - .set({ status: 'offline', handlersLive: false, load: 0, activeAgents: 0, lastHeartbeatAt: new Date() }) + .set({ status: 'offline', handlersLive: false, load: 0, loadReported: false, activeAgents: 0, lastHeartbeatAt: new Date() }) .where(and( eq(nodeProviders.workspaceId, workspaceId), eq(nodeProviders.nodeId, nodeId), @@ -306,9 +321,17 @@ export async function recomputeNodeAggregate( } } - const maxAgents = providers.reduce((sum, p) => sum + p.maxAgents, 0); + // A single unbounded provider makes the aggregate node unbounded. Treating + // its sentinel 0 as additive zero would falsely cap a mixed node at the sum + // of only its finite providers. + const maxAgents = providers.some((p) => p.maxAgents === 0) + ? 0 + : providers.reduce((sum, p) => sum + p.maxAgents, 0); const activeAgents = providers.reduce((sum, p) => sum + p.activeAgents, 0); - const load = providers.reduce((max, p) => Math.max(max, p.load), 0); + // A max is only a measurement when every constituent is measured; one + // unknown provider could be busier than every reported provider. + const loadReported = providers.length > 0 && providers.every((p) => p.loadReported); + const load = loadReported ? Math.max(...providers.map((p) => p.load)) : 0; const handlersLive = online.some((p) => p.handlersLive); // Node liveness must reflect an online provider's heartbeat. An offline // provider's disconnect timestamp is fresh, so including it here could keep a @@ -323,6 +346,7 @@ export async function recomputeNodeAggregate( maxAgents, activeAgents, load, + loadReported, handlersLive, status: online.length > 0 ? 'online' : 'offline', lastHeartbeatAt: lastHeartbeatAt ?? new Date(), diff --git a/packages/engine/src/engine/placement.ts b/packages/engine/src/engine/placement.ts index b62315d0..dd720a31 100644 --- a/packages/engine/src/engine/placement.ts +++ b/packages/engine/src/engine/placement.ts @@ -69,10 +69,12 @@ export function providerAttachDecision(input: ProviderAttachDecisionInput): Prov } export function isNodeLive(node: Pick, now = Date.now()): boolean { + const age = node.lastHeartbeatAt ? now - node.lastHeartbeatAt.getTime() : null; return ( node.status === 'online' && - !!node.lastHeartbeatAt && - now - node.lastHeartbeatAt.getTime() <= NODE_LIVENESS_TTL_MS + age !== null && + age >= 0 && + age <= NODE_LIVENESS_TTL_MS ); } @@ -88,6 +90,16 @@ export function nodeHasCapacity(node: Pick 0 ? target.trim() : undefined; } @@ -248,7 +260,7 @@ export async function claimSpawnNode( node.handlersLive && nodeHasCapacity(node), ) - .sort((a, b) => (a.load - b.load) || (a.activeAgents - b.activeAgents) || a.name.localeCompare(b.name)); + .sort(compareNodeCapacityLoad); for (const node of eligible) { const reserved = await reserveNodeCapacity(tx, workspaceId, node.id); @@ -350,7 +362,7 @@ export async function chooseNodeForAction( node.handlersLive && nodeHasCapacity(node), ) - .sort((a, b) => (a.load - b.load) || (a.activeAgents - b.activeAgents) || a.name.localeCompare(b.name)); + .sort(compareNodeCapacityLoad); const node = eligible[0]; if (!node) { diff --git a/packages/sdk-python/src/relay_sdk/models.py b/packages/sdk-python/src/relay_sdk/models.py index 3fb50316..fe3f8360 100644 --- a/packages/sdk-python/src/relay_sdk/models.py +++ b/packages/sdk-python/src/relay_sdk/models.py @@ -520,7 +520,7 @@ class NodeRosterEntry(BaseModel): status: str live: bool handlers_live: bool - load: float + load: float | None active_agents: int max_agents: int last_heartbeat_at: str | None = None diff --git a/packages/sdk-python/src/relay_sdk/node.py b/packages/sdk-python/src/relay_sdk/node.py index 3219e4fe..40133c93 100644 --- a/packages/sdk-python/src/relay_sdk/node.py +++ b/packages/sdk-python/src/relay_sdk/node.py @@ -613,15 +613,12 @@ async def _heartbeat_loop(self) -> None: try: while True: await asyncio.sleep(self._heartbeat_interval) - await self._send_frame( - { - "type": "node.heartbeat", - "provider": self._provider_identity(), - "load": 0, - "active_agents": 0, - "handlers_live": True, - } - ) + await self._send_frame({ + "type": "node.heartbeat", + "provider": self._provider_identity(), + "active_agents": 0, + "handlers_live": True, + }) except asyncio.CancelledError: raise diff --git a/packages/sdk-python/src/relay_sdk/ws.py b/packages/sdk-python/src/relay_sdk/ws.py index 8f2a7c6c..0277c237 100644 --- a/packages/sdk-python/src/relay_sdk/ws.py +++ b/packages/sdk-python/src/relay_sdk/ws.py @@ -282,7 +282,7 @@ async def _ping_loop(self, node_registration: dict[str, Any] | None = None) -> N "type": "node.heartbeat", "node_id": node_registration.get("node_id"), "name": node_registration.get("name"), - "load": 0, + "load": None, "active_agents": 1, "handlers_live": False, }) diff --git a/packages/sdk-python/tests/test_node.py b/packages/sdk-python/tests/test_node.py index 04e6baed..6633fa85 100644 --- a/packages/sdk-python/tests/test_node.py +++ b/packages/sdk-python/tests/test_node.py @@ -321,9 +321,9 @@ async def test_unknown_action_errors_rather_than_dropping(): @pytest.mark.asyncio -async def test_heartbeat_is_provider_scoped_with_no_last_heartbeat_at(): +async def test_finite_heartbeat_is_provider_scoped_without_placeholder_load_or_last_heartbeat_at(): server = FakeNodeServer() - node = NodeProvider(**base_kwargs(server, heartbeat_interval=0.02)) + node = NodeProvider(**base_kwargs(server, max_agents=4, heartbeat_interval=0.02)) task = asyncio.create_task(node.serve()) conn = await server.next_connection() @@ -333,7 +333,7 @@ async def test_heartbeat_is_provider_scoped_with_no_last_heartbeat_at(): await wait_until(lambda: len(conn.sent_of_type("node.heartbeat")) >= 1) hb = conn.sent_of_type("node.heartbeat")[-1] assert hb["provider"] == {"name": "py", "instance_id": node._instance_id} - assert hb["load"] == 0 + assert "load" not in hb assert hb["active_agents"] == 0 assert hb["handlers_live"] is True assert "last_heartbeat_at" not in hb diff --git a/packages/sdk-python/tests/test_ws.py b/packages/sdk-python/tests/test_ws.py index 6b00e473..3af05437 100644 --- a/packages/sdk-python/tests/test_ws.py +++ b/packages/sdk-python/tests/test_ws.py @@ -1,5 +1,6 @@ """Tests for WsClient.""" +import asyncio import json from unittest.mock import AsyncMock, MagicMock, patch @@ -146,6 +147,21 @@ async def test_send_noop_when_not_connected(self): # _ws is None; should not raise await ws.send({"type": "custom"}) + @pytest.mark.asyncio + async def test_direct_node_ping_keeps_load_unreported(self): + ws = WsClient(token="at_xxx") + ws._ws = MagicMock() + ws._ws.send = AsyncMock() + registration = {"node_id": "node_1", "name": "direct-alice"} + + sleep = AsyncMock(side_effect=[None, asyncio.CancelledError]) + with patch("relay_sdk.ws.asyncio.sleep", sleep): + await ws._ping_loop(registration) + + heartbeat = json.loads(ws._ws.send.await_args.args[0]) + assert heartbeat["load"] is None + assert "load_reported" not in heartbeat + class TestWsClientAutoPong: @pytest.mark.asyncio diff --git a/packages/sdk-rust/CHANGELOG.md b/packages/sdk-rust/CHANGELOG.md index 40f96413..9fab5cbc 100644 --- a/packages/sdk-rust/CHANGELOG.md +++ b/packages/sdk-rust/CHANGELOG.md @@ -6,7 +6,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on Keep a Changelog, and this project follows Semantic Versioning. -## [Unreleased] +## [Unreleased - Major] + +### Changed + +- `NodeRosterEntry.load` is now `Option`, matching the API's explicit unreported state; direct-agent heartbeats no longer label a constant utilization as measured. ## [4.2.0] - 2026-06-24 diff --git a/packages/sdk-rust/src/types.rs b/packages/sdk-rust/src/types.rs index 8a1ba15e..2b10234f 100644 --- a/packages/sdk-rust/src/types.rs +++ b/packages/sdk-rust/src/types.rs @@ -1879,7 +1879,7 @@ pub struct NodeRosterEntry { pub status: String, pub live: bool, pub handlers_live: bool, - pub load: f64, + pub load: Option, pub active_agents: i64, pub max_agents: i64, #[serde(default)] diff --git a/packages/sdk-rust/src/ws.rs b/packages/sdk-rust/src/ws.rs index fadfd299..1c01b2be 100644 --- a/packages/sdk-rust/src/ws.rs +++ b/packages/sdk-rust/src/ws.rs @@ -479,18 +479,7 @@ impl WsClient { } _ = ping_interval.tick() => { let ping = if let Some(registration) = &node_registration { - serde_json::json!({ - "v": 1, - "type": "node.heartbeat", - "load": 0, - "active_agents": 1, - "handlers_live": false, - "node_id": registration.node_id, - "name": registration.name, - "capabilities": [], - "max_agents": 1, - "version": SDK_VERSION, - }) + node_heartbeat(registration) } else { serde_json::json!({"type": "ping"}) }; @@ -597,6 +586,21 @@ fn reconnect_delay_ms(attempt: u32, max_delay_ms: u64) -> u64 { delay.min(max_delay_ms.max(1_000)) } +fn node_heartbeat(registration: &NodeRegistration) -> serde_json::Value { + serde_json::json!({ + "v": 1, + "type": "node.heartbeat", + "load": null, + "active_agents": 1, + "handlers_live": false, + "node_id": registration.node_id, + "name": registration.name, + "capabilities": [], + "max_agents": 1, + "version": SDK_VERSION, + }) +} + async fn send_node_register( write: &mut S, registration: &NodeRegistration, @@ -726,7 +730,7 @@ fn truncate_str(s: &str, max_chars: usize) -> &str { #[cfg(test)] mod tests { - use super::{normalize_node_message, truncate_str}; + use super::{node_heartbeat, normalize_node_message, truncate_str, NodeRegistration}; use crate::types::WsEvent; use serde_json::json; @@ -744,6 +748,18 @@ mod tests { assert_eq!(truncate_str("hello", 200), "hello"); } + #[test] + fn direct_node_heartbeat_keeps_load_unreported() { + let heartbeat = node_heartbeat(&NodeRegistration { + node_id: "node_1".to_string(), + name: "direct-alice".to_string(), + agent_name: "alice".to_string(), + }); + + assert_eq!(heartbeat.get("load"), Some(&serde_json::Value::Null)); + assert!(heartbeat.get("load_reported").is_none()); + } + #[test] fn normalizes_node_context_update_frames() { let normalized = normalize_node_message(json!({ diff --git a/packages/sdk-swift/CHANGELOG.md b/packages/sdk-swift/CHANGELOG.md index f201e49f..ebfdedcc 100644 --- a/packages/sdk-swift/CHANGELOG.md +++ b/packages/sdk-swift/CHANGELOG.md @@ -4,7 +4,9 @@ All notable changes to `relaycast-swift` will be documented in this file. See the [root changelog](../../CHANGELOG.md) for cross-package release highlights. -## [Unreleased] +## [Unreleased - Major] + +- `NodeRosterEntry.load` is optional; provider and direct-agent heartbeats no longer label placeholder utilization as measured. ## [6.1.0] - 2026-07-16 diff --git a/packages/sdk-swift/Sources/Relaycast/Models.swift b/packages/sdk-swift/Sources/Relaycast/Models.swift index 7f11419b..fb6b48e7 100644 --- a/packages/sdk-swift/Sources/Relaycast/Models.swift +++ b/packages/sdk-swift/Sources/Relaycast/Models.swift @@ -1674,7 +1674,7 @@ public struct NodeRosterEntry: Codable, Equatable, Sendable { public let status: String public let live: Bool public let handlersLive: Bool - public let load: Double + public let load: Double? public let activeAgents: Int public let maxAgents: Int public let lastHeartbeatAt: String? @@ -1793,7 +1793,7 @@ public struct CreateNodeResponse: Codable, Equatable, Sendable { public let status: String public let live: Bool public let handlersLive: Bool - public let load: Double + public let load: Double? public let activeAgents: Int public let maxAgents: Int public let lastHeartbeatAt: String? diff --git a/packages/sdk-swift/Sources/Relaycast/NodeProvider.swift b/packages/sdk-swift/Sources/Relaycast/NodeProvider.swift index f3334887..4a9c0a94 100644 --- a/packages/sdk-swift/Sources/Relaycast/NodeProvider.swift +++ b/packages/sdk-swift/Sources/Relaycast/NodeProvider.swift @@ -633,7 +633,6 @@ public actor NodeProvider { "v": .int(1), "type": .string("node.heartbeat"), "provider": .object(["name": .string(providerName), "instance_id": .string(instanceID)]), - "load": .int(0), "active_agents": .int(0), "handlers_live": .bool(true) ] diff --git a/packages/sdk-swift/Sources/Relaycast/WsClient.swift b/packages/sdk-swift/Sources/Relaycast/WsClient.swift index 06cc6329..17cf5e1f 100644 --- a/packages/sdk-swift/Sources/Relaycast/WsClient.swift +++ b/packages/sdk-swift/Sources/Relaycast/WsClient.swift @@ -394,7 +394,7 @@ public final class WsClient: @unchecked Sendable { let heartbeat: [String: JSONValue] = [ "v": 1, "type": "node.heartbeat", - "load": 0, + "load": .null, "active_agents": 1, "handlers_live": false, "node_id": .string(registration.nodeId), diff --git a/packages/sdk-swift/Tests/RelaycastTests/NodeProviderTests.swift b/packages/sdk-swift/Tests/RelaycastTests/NodeProviderTests.swift index 62a8d6dc..beff1eae 100644 --- a/packages/sdk-swift/Tests/RelaycastTests/NodeProviderTests.swift +++ b/packages/sdk-swift/Tests/RelaycastTests/NodeProviderTests.swift @@ -396,13 +396,14 @@ final class NodeProviderTests: XCTestCase { try await serveTask.value } - func testSendsProviderScopedHeartbeatsWhileServing() async throws { + func testSendsFiniteProviderScopedHeartbeatsWithoutPlaceholderLoad() async throws { let factory = FakeTransportFactory() let node = NodeProvider( nodeToken: baseOptions.nodeToken, nodeID: baseOptions.nodeID, nodeName: baseOptions.nodeName, provider: (name: "py", instanceID: nil), + maxAgents: 4, heartbeatIntervalMilliseconds: 30, transport: { factory.make() } ) @@ -414,7 +415,7 @@ final class NodeProviderTests: XCTestCase { let heartbeat = await transport.sentOfType("node.heartbeat").last! guard case .object(let provider)? = heartbeat["provider"] else { return XCTFail("missing provider") } XCTAssertEqual(provider["name"], .string("py")) - XCTAssertEqual(heartbeat["load"], .int(0)) + XCTAssertNil(heartbeat["load"]) XCTAssertEqual(heartbeat["active_agents"], .int(0)) XCTAssertEqual(heartbeat["handlers_live"], .bool(true)) XCTAssertNil(heartbeat["last_heartbeat_at"]) diff --git a/packages/sdk-typescript/CHANGELOG.md b/packages/sdk-typescript/CHANGELOG.md index 5abf7a3f..c98d61a2 100644 --- a/packages/sdk-typescript/CHANGELOG.md +++ b/packages/sdk-typescript/CHANGELOG.md @@ -7,7 +7,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Major] + +### Changed + +- `NodeRosterEntry.load` is now `number | null`; provider and direct-agent heartbeats no longer label placeholder utilization as measured. ## [6.3.0] - 2026-07-28 diff --git a/packages/sdk-typescript/src/__tests__/agent-ws.test.ts b/packages/sdk-typescript/src/__tests__/agent-ws.test.ts index 9ee5486f..b3a1a0cc 100644 --- a/packages/sdk-typescript/src/__tests__/agent-ws.test.ts +++ b/packages/sdk-typescript/src/__tests__/agent-ws.test.ts @@ -113,6 +113,20 @@ describe('AgentClient WebSocket integration', () => { expect(url.searchParams.get('origin_version')).toBeDefined(); }); + it('keeps direct-node heartbeat load explicitly unreported', async () => { + const agent = createAgent(); + agent.connect(); + const ws = await nextSocket(); + ws.simulateOpen(); + + await vi.advanceTimersByTimeAsync(30_000); + const frames = ws.send.mock.calls.map(([raw]) => JSON.parse(String(raw)) as Record); + const heartbeat = frames.find((frame) => frame.type === 'node.heartbeat'); + + expect(heartbeat).toMatchObject({ load: null, active_agents: 1, handlers_live: false }); + expect(heartbeat).not.toHaveProperty('load_reported'); + }); + it('connect() normalizes trailing slash base URL', async () => { const client = new HttpClient({ apiKey: 'at_live_test', diff --git a/packages/sdk-typescript/src/__tests__/node-provider.test.ts b/packages/sdk-typescript/src/__tests__/node-provider.test.ts index 65bd6c3f..8c4a5460 100644 --- a/packages/sdk-typescript/src/__tests__/node-provider.test.ts +++ b/packages/sdk-typescript/src/__tests__/node-provider.test.ts @@ -205,8 +205,8 @@ describe('NodeProviderClient', () => { expect(sock.sentOfType('action.result').at(-1)).toMatchObject({ invocation_id: 'inv-x', error: expect.stringContaining('nope') }); }); - it('sends provider-scoped heartbeats while serving', async () => { - const node = new NodeProviderClient({ ...baseOptions, heartbeatIntervalMs: 1_000 }); + it('sends provider-scoped heartbeats without inventing finite-capacity load', async () => { + const node = new NodeProviderClient({ ...baseOptions, maxAgents: 4, heartbeatIntervalMs: 1_000 }); node.serve(); const sock = newSocket(); sock.open(); @@ -215,7 +215,8 @@ describe('NodeProviderClient', () => { await vi.advanceTimersByTimeAsync(1_000); const hb = sock.sentOfType('node.heartbeat').at(-1)!; - expect(hb).toMatchObject({ provider: { name: 'py' }, load: 0, active_agents: 0, handlers_live: true }); + expect(hb).toMatchObject({ provider: { name: 'py' }, active_agents: 0, handlers_live: true }); + expect(hb).not.toHaveProperty('load'); expect(hb).not.toHaveProperty('last_heartbeat_at'); }); diff --git a/packages/sdk-typescript/src/node-provider.ts b/packages/sdk-typescript/src/node-provider.ts index 2e6286c1..9093af81 100644 --- a/packages/sdk-typescript/src/node-provider.ts +++ b/packages/sdk-typescript/src/node-provider.ts @@ -563,7 +563,6 @@ export class NodeProviderClient { this.sendFrame({ type: 'node.heartbeat', provider: this.providerIdentity(), - load: 0, active_agents: 0, handlers_live: true, }); diff --git a/packages/sdk-typescript/src/types.ts b/packages/sdk-typescript/src/types.ts index bfb94551..1e52f475 100644 --- a/packages/sdk-typescript/src/types.ts +++ b/packages/sdk-typescript/src/types.ts @@ -342,7 +342,8 @@ export interface NodeRosterEntry { status: 'online' | 'offline' | string; live: boolean; handlersLive: boolean; - load: number; + /** Managed-agent capacity utilization in [0,1], or null when unreported. */ + load: number | null; activeAgents: number; maxAgents: number; lastHeartbeatAt: string | null; diff --git a/packages/sdk-typescript/src/ws.ts b/packages/sdk-typescript/src/ws.ts index e39d0561..14f5d723 100644 --- a/packages/sdk-typescript/src/ws.ts +++ b/packages/sdk-typescript/src/ws.ts @@ -612,7 +612,7 @@ export class WsClient { v: 1, type: 'node.heartbeat', ...(nodeId ? { node_id: nodeId } : {}), - load: 0, + load: null, active_agents: 1, handlers_live: false, }); diff --git a/packages/types/CHANGELOG.md b/packages/types/CHANGELOG.md index e4cf256e..0e1fce59 100644 --- a/packages/types/CHANGELOG.md +++ b/packages/types/CHANGELOG.md @@ -7,7 +7,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Major] + +### Changed + +- `node.heartbeat.load` may be absent or null when capacity utilization is unreported; `load_reported: true` explicitly identifies a numeric `[0,1]` value as a measurement while legacy placeholder numbers remain accepted but untrusted. ## [6.3.0] - 2026-07-28 diff --git a/packages/types/src/__tests__/fleet-wire-fixtures.test.ts b/packages/types/src/__tests__/fleet-wire-fixtures.test.ts index bba1b137..c5bd1f34 100644 --- a/packages/types/src/__tests__/fleet-wire-fixtures.test.ts +++ b/packages/types/src/__tests__/fleet-wire-fixtures.test.ts @@ -120,6 +120,29 @@ describe('fleet wire fixtures', () => { }); } + it('accepts unreported heartbeat load and rejects values outside [0,1]', () => { + const heartbeat = { + v: 1 as const, + type: 'node.heartbeat' as const, + active_agents: 25, + handlers_live: true, + }; + expect(parseFleetBrokerToRelaycastMessage(heartbeat)).toEqual(heartbeat); + expect(parseFleetBrokerToRelaycastMessage({ ...heartbeat, load: null })).toEqual({ + ...heartbeat, + load: null, + }); + expect(parseFleetBrokerToRelaycastMessage({ ...heartbeat, load: 0.25, load_reported: true })).toEqual({ + ...heartbeat, + load: 0.25, + load_reported: true, + }); + expect(() => parseFleetBrokerToRelaycastMessage({ ...heartbeat, load_reported: true })).toThrow(); + expect(() => parseFleetBrokerToRelaycastMessage({ ...heartbeat, load: null, load_reported: true })).toThrow(); + expect(() => parseFleetBrokerToRelaycastMessage({ ...heartbeat, load: -0.01 })).toThrow(); + expect(() => parseFleetBrokerToRelaycastMessage({ ...heartbeat, load: 1.01 })).toThrow(); + }); + it('accepts action.result error variants as the same message type', () => { expect( parseFleetBrokerToRelaycastMessage({ diff --git a/packages/types/src/fleet-wire.ts b/packages/types/src/fleet-wire.ts index dcce76cf..4439a99b 100644 --- a/packages/types/src/fleet-wire.ts +++ b/packages/types/src/fleet-wire.ts @@ -138,7 +138,11 @@ export const FleetNodeHeartbeatMessageSchema = z // synthetic `default` provider. Load/active_agents/handlers_live describe // the sending provider, and the node figures aggregate across providers. provider: FleetProviderIdentitySchema.optional(), - load: z.number().finite().nonnegative(), + // Capacity utilization is absent/null until the provider has a genuine + // measurement. Numeric values from legacy clients remain accepted for wire + // compatibility but are only trusted when load_reported is explicitly true. + load: z.number().finite().min(0).max(1).nullable().optional(), + load_reported: z.boolean().optional(), active_agents: z.number().int().nonnegative(), handlers_live: z.boolean(), // Roster snapshot carried for liveness: lets the engine refresh a node's @@ -156,6 +160,15 @@ export const FleetNodeHeartbeatMessageSchema = z max_agents: z.number().int().nonnegative().optional(), version: z.string().optional(), }) + .superRefine((message, ctx) => { + if (message.load_reported === true && typeof message.load !== 'number') { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['load'], + message: 'load must be numeric when load_reported is true', + }); + } + }) .strict(); export type FleetNodeHeartbeatMessage = z.infer;