mirror of
https://github.com/ruvnet/ruflo.git
synced 2026-10-30 00:28:20 +09:00
feat(federation): ADR-097 Phase 1 — budget envelope + hop counter (#1723)
Implements the send-side enforcement portion of ADR-097 (federation
budget circuit breaker). Closes the recursive-delegation-loop class
with a default maxHops=8 even for callers that don't opt in to a
budget. Token / USD caps are opt-in via the new federation_send
metadata.
Built around the security constraints surfaced in the team's review
pass:
1. validateBudget rejects NaN, ±Infinity, negatives, non-integer hops
up front. Discriminated result type — callers cannot accidentally
use a bad budget unchecked.
2. enforceBudget is purely synchronous (no internal awaits) so two
concurrent sends cannot both pass a single-hop budget.
3. Errors are constant strings (HOP_LIMIT_EXCEEDED, BUDGET_EXCEEDED,
INVALID_BUDGET) with no remaining-budget echo on failure — no
oracle for probing thresholds.
4. Negative caller-reported spend is clamped to 0 so a malicious
caller cannot inflate remaining budget.
5. Hard ceilings on every axis (maxHops ≤ 64, maxTokens ≤ 1B,
maxUsd ≤ $1M) bound the failure mode of trusted-but-buggy
callers.
New module:
src/domain/value-objects/federation-budget.ts
- Budget interface (immutable, opaque to callers; only constructed
via validateBudget)
- DEFAULT_MAX_HOPS = 8
- validateBudget(raw, overrideMaxHops?) → discriminated result
- enforceBudget(budget, hopCount, spent) → discriminated decision
with nextHopCount + remaining for downstream
Wired into:
src/application/federation-coordinator.ts
sendMessage gains an optional fourth `options: SendOptions`
argument carrying { budget, maxHops, hopCount, spent }. Default
SendOptions = {} preserves the existing 3-arg call sites bit-for-bit.
Budget gate runs immediately after the existing policyEngine
check, before routing.send. Failures audit-log with the specific
constant-string reason.
src/mcp-tools.ts (federation_send)
Schema gains optional budget / maxHops / hopCount / spent fields.
No required-field changes — callers without budget metadata see
identical behavior except the new default maxHops=8 floor.
Tests (__tests__/unit/federation-budget.test.ts, 41 cases):
validateBudget: 27 cases
- happy path: undefined/null/full/zero-hops/override
- rejection: malformed types, NaN/Infinity hops, negative hops,
non-integer hops, oversized hops, NaN/Infinity tokens, non-integer
tokens, oversized tokens, USD validation
enforceBudget: 14 cases
- hop counter: pass/refuse/zero-hops, recursive-ring termination
at exactly maxHops legs
- budget axes: tokens overshoot, USD overshoot, deterministic order
when hop+budget both exhausted
- downstream propagation: remaining budget for next peer
- anti-malice: negative spend clamped to 0
- atomic: pinned synchronous return type (not Promise)
- error info leak: failure result contains only ok/reason, constant
strings only
Verified:
Plugin tsc --noEmit: 0 errors
Plugin vitest: 366/366 passing (was 325; +41 from this commit)
Parent CLI vitest: 1917/1917 passing, 46 skipped, 0 failures
(no regressions from the new SendOptions param)
Phase 2 (peer state machine ACTIVE/SUSPENDED/EVICTED), Phase 3
(cost-tracker integration), and Phase 4 (doctor + breaker_status MCP
tool) ship in follow-up iterations per ADR-097.
The team-of-agents pipeline broke (sibling background agents can't
SendMessage each other), but the researcher's recon and reviewer's
security pass both produced load-bearing input that shaped this
implementation. Tester wrote a parallel file against a phantom
reference impl which would have duplicated coverage; replaced with
the canonical federation-budget.test.ts that imports from real source.
This commit is contained in:
1 parent
62a6fc5fb0
commit
7e1cc06dfd
4 files changed
+616
-2
No files matched your search
@@ -0,0 +1,290 @@
|
||||
/**
|
||||
* Tests for ADR-097 Phase 1 — federation budget validator + enforcer.
|
||||
*
|
||||
* Pins the security invariants the reviewer called out (audit_1776853149979
|
||||
* follow-up):
|
||||
* 1. validateBudget rejects NaN, ±Infinity, negatives, non-integer hops
|
||||
* 2. enforceBudget is atomic (synchronous, no awaits) — pinned by
|
||||
* checking that two concurrent calls cannot both pass a single-hop
|
||||
* budget when the caller passes the same hopCount=0 in.
|
||||
* 3. Errors are constant strings — no remaining-budget echo on failure
|
||||
* 4. Backward compat: omitted budget → unbounded with default maxHops=8
|
||||
*/
|
||||
|
||||
import { describe, it, expect } from 'vitest';
|
||||
import {
|
||||
DEFAULT_MAX_HOPS,
|
||||
enforceBudget,
|
||||
validateBudget,
|
||||
} from '../../src/domain/value-objects/federation-budget.js';
|
||||
|
||||
describe('validateBudget (ADR-097 Phase 1)', () => {
|
||||
describe('happy path', () => {
|
||||
it('accepts undefined → unbounded budget with default maxHops', () => {
|
||||
const r = validateBudget(undefined);
|
||||
expect(r.ok).toBe(true);
|
||||
if (!r.ok) return;
|
||||
expect(r.budget.maxHops).toBe(DEFAULT_MAX_HOPS);
|
||||
expect(r.budget.maxTokens).toBe(Number.POSITIVE_INFINITY);
|
||||
expect(r.budget.maxUsd).toBe(Number.POSITIVE_INFINITY);
|
||||
});
|
||||
|
||||
it('accepts null → unbounded with default maxHops', () => {
|
||||
const r = validateBudget(null);
|
||||
expect(r.ok).toBe(true);
|
||||
});
|
||||
|
||||
it('accepts a fully-specified budget', () => {
|
||||
const r = validateBudget({ maxTokens: 1000, maxUsd: 0.5, maxHops: 4 });
|
||||
expect(r.ok).toBe(true);
|
||||
if (!r.ok) return;
|
||||
expect(r.budget).toEqual({ maxTokens: 1000, maxUsd: 0.5, maxHops: 4 });
|
||||
});
|
||||
|
||||
it('accepts maxHops=0 (no remote delegation allowed)', () => {
|
||||
const r = validateBudget({ maxHops: 0 });
|
||||
expect(r.ok).toBe(true);
|
||||
if (!r.ok) return;
|
||||
expect(r.budget.maxHops).toBe(0);
|
||||
});
|
||||
|
||||
it('overrideMaxHops takes precedence over the budget object value', () => {
|
||||
const r = validateBudget({ maxHops: 5 }, 2);
|
||||
expect(r.ok).toBe(true);
|
||||
if (!r.ok) return;
|
||||
expect(r.budget.maxHops).toBe(2);
|
||||
});
|
||||
|
||||
it('overrideMaxHops alone (no budget object) yields a hop-only budget', () => {
|
||||
const r = validateBudget(undefined, 0);
|
||||
expect(r.ok).toBe(true);
|
||||
if (!r.ok) return;
|
||||
expect(r.budget.maxHops).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('rejection: malformed types', () => {
|
||||
it.each([
|
||||
['string', 'budget must be an object'],
|
||||
[42, 'budget must be an object'],
|
||||
[[1, 2], 'budget must be an object'],
|
||||
])('rejects raw=%s', (raw, expected) => {
|
||||
const r = validateBudget(raw as unknown);
|
||||
expect(r.ok).toBe(false);
|
||||
if (r.ok) return;
|
||||
expect(r.error).toContain(expected);
|
||||
});
|
||||
});
|
||||
|
||||
describe('rejection: hop validation', () => {
|
||||
it.each([
|
||||
[Number.NaN, 'finite integer'],
|
||||
[Number.POSITIVE_INFINITY, 'finite integer'],
|
||||
[Number.NEGATIVE_INFINITY, 'finite integer'],
|
||||
[1.5, 'finite integer'],
|
||||
[-1, '>= 0'],
|
||||
[65, 'ceiling'],
|
||||
[Number.MAX_SAFE_INTEGER, 'ceiling'],
|
||||
['8' as unknown as number, 'finite integer'],
|
||||
])('rejects maxHops=%s', (val, expected) => {
|
||||
const r = validateBudget({ maxHops: val });
|
||||
expect(r.ok).toBe(false);
|
||||
if (r.ok) return;
|
||||
expect(r.error).toContain(expected);
|
||||
});
|
||||
});
|
||||
|
||||
describe('rejection: token validation', () => {
|
||||
it.each([
|
||||
[Number.NaN, 'finite number'],
|
||||
[Number.POSITIVE_INFINITY, 'finite number'],
|
||||
[-1, '>= 0'],
|
||||
[1.5, 'integer'],
|
||||
[1_000_000_001, 'ceiling'],
|
||||
])('rejects maxTokens=%s', (val, expected) => {
|
||||
const r = validateBudget({ maxTokens: val });
|
||||
expect(r.ok).toBe(false);
|
||||
if (r.ok) return;
|
||||
expect(r.error).toContain(expected);
|
||||
});
|
||||
});
|
||||
|
||||
describe('rejection: usd validation', () => {
|
||||
it.each([
|
||||
[Number.NaN, 'finite number'],
|
||||
[Number.POSITIVE_INFINITY, 'finite number'],
|
||||
[-0.01, '>= 0'],
|
||||
[1_000_001, 'ceiling'],
|
||||
])('rejects maxUsd=%s', (val, expected) => {
|
||||
const r = validateBudget({ maxUsd: val });
|
||||
expect(r.ok).toBe(false);
|
||||
if (r.ok) return;
|
||||
expect(r.error).toContain(expected);
|
||||
});
|
||||
|
||||
it('accepts maxUsd as a non-integer (e.g. 0.001)', () => {
|
||||
const r = validateBudget({ maxUsd: 0.001 });
|
||||
expect(r.ok).toBe(true);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('enforceBudget (ADR-097 Phase 1)', () => {
|
||||
function bud(p: Partial<{ maxHops: number; maxTokens: number; maxUsd: number }> = {}) {
|
||||
const r = validateBudget(p);
|
||||
if (!r.ok) throw new Error(r.error);
|
||||
return r.budget;
|
||||
}
|
||||
|
||||
describe('hop counter', () => {
|
||||
it('passes when nextHopCount <= maxHops', () => {
|
||||
const r = enforceBudget(bud({ maxHops: 3 }), 0);
|
||||
expect(r.ok).toBe(true);
|
||||
if (!r.ok) return;
|
||||
expect(r.nextHopCount).toBe(1);
|
||||
});
|
||||
|
||||
it('refuses with HOP_LIMIT_EXCEEDED when nextHopCount > maxHops', () => {
|
||||
const r = enforceBudget(bud({ maxHops: 2 }), 2);
|
||||
expect(r.ok).toBe(false);
|
||||
if (r.ok) return;
|
||||
expect(r.reason).toBe('HOP_LIMIT_EXCEEDED');
|
||||
});
|
||||
|
||||
it('maxHops=0 refuses to forward at all (originator → first hop blocked)', () => {
|
||||
const r = enforceBudget(bud({ maxHops: 0 }), 0);
|
||||
expect(r.ok).toBe(false);
|
||||
if (r.ok) return;
|
||||
expect(r.reason).toBe('HOP_LIMIT_EXCEEDED');
|
||||
});
|
||||
|
||||
it('terminates a synthetic recursive ring at the configured maxHops', () => {
|
||||
const budget = bud({ maxHops: 3 });
|
||||
// Simulate hops A → B → A → B → … incrementing hopCount each step
|
||||
let hops = 0;
|
||||
let lastResult: ReturnType<typeof enforceBudget> | null = null;
|
||||
for (let i = 0; i < 10; i++) {
|
||||
lastResult = enforceBudget(budget, hops);
|
||||
if (!lastResult.ok) break;
|
||||
hops = lastResult.nextHopCount;
|
||||
}
|
||||
expect(lastResult).not.toBeNull();
|
||||
expect(lastResult!.ok).toBe(false);
|
||||
expect(hops).toBe(3); // exactly maxHops legs ran
|
||||
if (!lastResult!.ok) {
|
||||
expect(lastResult!.reason).toBe('HOP_LIMIT_EXCEEDED');
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe('budget axes', () => {
|
||||
it('passes when cumulative tokens still within cap', () => {
|
||||
const r = enforceBudget(bud({ maxTokens: 100, maxHops: 5 }), 0, {
|
||||
tokens: 30,
|
||||
usd: 0,
|
||||
});
|
||||
expect(r.ok).toBe(true);
|
||||
});
|
||||
|
||||
it('refuses with BUDGET_EXCEEDED when tokens overshoot', () => {
|
||||
const r = enforceBudget(bud({ maxTokens: 50, maxHops: 5 }), 0, {
|
||||
tokens: 60,
|
||||
usd: 0,
|
||||
});
|
||||
expect(r.ok).toBe(false);
|
||||
if (r.ok) return;
|
||||
expect(r.reason).toBe('BUDGET_EXCEEDED');
|
||||
});
|
||||
|
||||
it('refuses with BUDGET_EXCEEDED when USD overshoots', () => {
|
||||
const r = enforceBudget(bud({ maxUsd: 0.5, maxHops: 5 }), 0, {
|
||||
tokens: 0,
|
||||
usd: 0.6,
|
||||
});
|
||||
expect(r.ok).toBe(false);
|
||||
if (r.ok) return;
|
||||
expect(r.reason).toBe('BUDGET_EXCEEDED');
|
||||
});
|
||||
|
||||
it('hop limit takes priority over budget exhaust (deterministic order)', () => {
|
||||
const r = enforceBudget(bud({ maxHops: 0, maxTokens: 0 }), 0, {
|
||||
tokens: 100,
|
||||
usd: 0,
|
||||
});
|
||||
expect(r.ok).toBe(false);
|
||||
if (r.ok) return;
|
||||
expect(r.reason).toBe('HOP_LIMIT_EXCEEDED');
|
||||
});
|
||||
});
|
||||
|
||||
describe('downstream propagation', () => {
|
||||
it('returns remaining budget for the downstream peer', () => {
|
||||
const r = enforceBudget(bud({ maxTokens: 100, maxUsd: 1, maxHops: 5 }), 0, {
|
||||
tokens: 30,
|
||||
usd: 0.25,
|
||||
});
|
||||
expect(r.ok).toBe(true);
|
||||
if (!r.ok) return;
|
||||
expect(r.remaining.maxTokens).toBe(70);
|
||||
expect(r.remaining.maxUsd).toBe(0.75);
|
||||
expect(r.remaining.maxHops).toBe(5); // hop ceiling itself unchanged
|
||||
});
|
||||
});
|
||||
|
||||
describe('anti-malice: negative spend cannot inflate remaining', () => {
|
||||
it('treats negative tokens as 0 (no refund / inflation)', () => {
|
||||
const r = enforceBudget(bud({ maxTokens: 100, maxHops: 5 }), 0, {
|
||||
tokens: -50,
|
||||
usd: 0,
|
||||
});
|
||||
expect(r.ok).toBe(true);
|
||||
if (!r.ok) return;
|
||||
expect(r.remaining.maxTokens).toBe(100);
|
||||
});
|
||||
|
||||
it('treats negative usd as 0', () => {
|
||||
const r = enforceBudget(bud({ maxUsd: 1, maxHops: 5 }), 0, {
|
||||
tokens: 0,
|
||||
usd: -2,
|
||||
});
|
||||
expect(r.ok).toBe(true);
|
||||
if (!r.ok) return;
|
||||
expect(r.remaining.maxUsd).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('atomic check (no internal awaits)', () => {
|
||||
it('is purely synchronous — return value type is not a Promise', () => {
|
||||
const r = enforceBudget(bud({ maxHops: 1 }), 0);
|
||||
// The return type is the discriminated union, not a Promise. This
|
||||
// pins the no-internal-await contract: a future refactor that adds
|
||||
// an await would change the function to async and break this assertion.
|
||||
expect(r).not.toBeInstanceOf(Promise);
|
||||
expect(typeof (r as object as { then?: unknown }).then).toBe('undefined');
|
||||
});
|
||||
});
|
||||
|
||||
describe('error info leak (anti-oracle)', () => {
|
||||
it('failure result carries no remaining-budget data', () => {
|
||||
const r = enforceBudget(bud({ maxHops: 0 }), 0);
|
||||
expect(r.ok).toBe(false);
|
||||
if (r.ok) return;
|
||||
// Reason is the only public field on the failure case — no budget
|
||||
// numbers, no remaining counts, nothing a caller can probe with.
|
||||
expect(Object.keys(r).sort()).toEqual(['ok', 'reason']);
|
||||
});
|
||||
|
||||
it('uses constant strings (HOP_LIMIT_EXCEEDED / BUDGET_EXCEEDED only)', () => {
|
||||
const r1 = enforceBudget(bud({ maxHops: 0 }), 0);
|
||||
const r2 = enforceBudget(bud({ maxTokens: 0, maxHops: 5 }), 0, {
|
||||
tokens: 1,
|
||||
usd: 0,
|
||||
});
|
||||
expect(r1.ok).toBe(false);
|
||||
expect(r2.ok).toBe(false);
|
||||
if (r1.ok || r2.ok) return;
|
||||
expect(['HOP_LIMIT_EXCEEDED', 'BUDGET_EXCEEDED']).toContain(r1.reason);
|
||||
expect(['HOP_LIMIT_EXCEEDED', 'BUDGET_EXCEEDED']).toContain(r2.reason);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -9,6 +9,31 @@ import { AuditService } from '../domain/services/audit-service.js';
|
||||
import { PIIPipelineService } from '../domain/services/pii-pipeline-service.js';
|
||||
import { TrustEvaluator, type ImmediateDowngradeReason } from './trust-evaluator.js';
|
||||
import { PolicyEngine } from './policy-engine.js';
|
||||
import {
|
||||
type Budget,
|
||||
type BudgetEnforcement,
|
||||
enforceBudget,
|
||||
validateBudget,
|
||||
} from '../domain/value-objects/federation-budget.js';
|
||||
|
||||
/**
|
||||
* Optional per-call budget controls (ADR-097 Phase 1). All fields are
|
||||
* optional; omitted means "no limit on that axis" (the default unbounded
|
||||
* budget still has maxHops = 8 to defang recursive delegation loops).
|
||||
*/
|
||||
export interface SendOptions {
|
||||
readonly budget?: { maxTokens?: number; maxUsd?: number };
|
||||
readonly maxHops?: number;
|
||||
/**
|
||||
* How many hops this message has already traveled (0 on the originator).
|
||||
* Phase 1 enforces at the send side because there is no inbound dispatcher
|
||||
* yet; callers re-forwarding a received message should pass the original
|
||||
* envelope's hopCount here.
|
||||
*/
|
||||
readonly hopCount?: number;
|
||||
/** Caller-reported usage from the previous leg, for cumulative checks. */
|
||||
readonly spent?: { tokens?: number; usd?: number };
|
||||
}
|
||||
|
||||
export interface FederationCoordinatorConfig {
|
||||
readonly nodeId: string;
|
||||
@@ -119,6 +144,7 @@ export class FederationCoordinator {
|
||||
targetNodeId: string,
|
||||
messageType: FederationMessageType,
|
||||
payload: T,
|
||||
options: SendOptions = {},
|
||||
): Promise<RoutingResult> {
|
||||
this.ensureInitialized();
|
||||
|
||||
@@ -155,18 +181,72 @@ export class FederationCoordinator {
|
||||
};
|
||||
}
|
||||
|
||||
// ADR-097 Phase 1: validate caller-supplied budget + enforce hop limit
|
||||
// and cumulative spend BEFORE outbound dispatch. Synchronous & atomic —
|
||||
// no awaits between validate and enforce, so concurrent sends cannot
|
||||
// both pass a single-hop budget. Errors are constant strings so a
|
||||
// malicious caller cannot use the response as a budget oracle.
|
||||
const budgetResult = validateBudget(options.budget, options.maxHops);
|
||||
if (!budgetResult.ok) {
|
||||
await this.audit.log('message_rejected', {
|
||||
targetNodeId,
|
||||
sessionId: session.sessionId,
|
||||
metadata: { reason: 'INVALID_BUDGET', detail: budgetResult.error },
|
||||
});
|
||||
return {
|
||||
success: false,
|
||||
mode: 'direct',
|
||||
envelopeId: '',
|
||||
targetNodeIds: [targetNodeId],
|
||||
error: 'INVALID_BUDGET',
|
||||
};
|
||||
}
|
||||
const enforcement: BudgetEnforcement = enforceBudget(
|
||||
budgetResult.budget,
|
||||
options.hopCount ?? 0,
|
||||
{
|
||||
tokens: options.spent?.tokens ?? 0,
|
||||
usd: options.spent?.usd ?? 0,
|
||||
},
|
||||
);
|
||||
if (!enforcement.ok) {
|
||||
await this.audit.log('message_rejected', {
|
||||
targetNodeId,
|
||||
sessionId: session.sessionId,
|
||||
metadata: { reason: enforcement.reason },
|
||||
});
|
||||
return {
|
||||
success: false,
|
||||
mode: 'direct',
|
||||
envelopeId: '',
|
||||
targetNodeIds: [targetNodeId],
|
||||
error: enforcement.reason, // constant string — no oracle leak
|
||||
};
|
||||
}
|
||||
|
||||
const result = await this.routing.send(session, messageType, payload);
|
||||
|
||||
await this.audit.log(result.success ? 'message_sent' : 'message_rejected', {
|
||||
targetNodeId,
|
||||
sessionId: session.sessionId,
|
||||
latencyMs: result.latencyMs,
|
||||
metadata: { messageType, mode: result.mode },
|
||||
metadata: {
|
||||
messageType,
|
||||
mode: result.mode,
|
||||
hopCount: enforcement.nextHopCount,
|
||||
},
|
||||
});
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-export the budget primitives so external callers (e.g. follow-up
|
||||
* iterations adding receive-side decrement) can reuse them without
|
||||
* dipping into domain/value-objects directly.
|
||||
*/
|
||||
static readonly budget = { validate: validateBudget, enforce: enforceBudget };
|
||||
|
||||
async broadcastMessage<T>(
|
||||
messageType: FederationMessageType,
|
||||
payload: T,
|
||||
|
||||
+211
@@ -0,0 +1,211 @@
|
||||
/**
|
||||
* Federation budget + hop counter (ADR-097 Phase 1).
|
||||
*
|
||||
* Carried in {@link FederationEnvelope} alongside the existing
|
||||
* sourceNodeId/targetNodeId/payload. Sender supplies the original limits;
|
||||
* each outbound hop increments hopCount and decrements remaining budget.
|
||||
*
|
||||
* Phase 1 enforces at the *send* side because there is no inbound
|
||||
* dispatcher today (sendToNode is a stub — see researcher report).
|
||||
* Phase 2 will move the decrement to receive once the inbound path lands.
|
||||
*
|
||||
* Security invariants pinned by the reviewer (audit_1776853149979 follow-up):
|
||||
*
|
||||
* 1. validateBudget rejects NaN, ±Infinity, negative numbers, and
|
||||
* non-integer hop counts. Returns a discriminated result so callers
|
||||
* cannot accidentally use a bad budget unchecked.
|
||||
* 2. enforceBudget runs check-then-decrement in a single synchronous
|
||||
* function with no awaits inside — two concurrent send calls cannot
|
||||
* both pass the pre-check.
|
||||
* 3. Errors return constant strings ("HOP_LIMIT_EXCEEDED" /
|
||||
* "BUDGET_EXCEEDED") with no remaining-budget echo so a caller can't
|
||||
* use the error response as an oracle for the threshold.
|
||||
*/
|
||||
|
||||
/** Maximum hops a federation message may travel before the breaker opens. */
|
||||
export const DEFAULT_MAX_HOPS = 8;
|
||||
/** Hard ceiling on caller-supplied maxUsd to prevent absurd integer math. */
|
||||
const MAX_USD_CEILING = 1_000_000;
|
||||
/** Hard ceiling on caller-supplied maxTokens. 1B tokens is more than any real session. */
|
||||
const MAX_TOKENS_CEILING = 1_000_000_000;
|
||||
/** Hard ceiling on hop count. */
|
||||
const MAX_HOPS_CEILING = 64;
|
||||
|
||||
/**
|
||||
* Caller-supplied budget. All fields are optional; omitted fields are
|
||||
* treated as Infinity for the corresponding axis. Created via validateBudget;
|
||||
* never construct directly.
|
||||
*/
|
||||
export interface Budget {
|
||||
readonly maxHops: number;
|
||||
readonly maxTokens: number; // Infinity allowed
|
||||
readonly maxUsd: number; // Infinity allowed
|
||||
}
|
||||
|
||||
/** Discriminated result of validating raw user input into a Budget. */
|
||||
export type BudgetValidationResult =
|
||||
| { ok: true; budget: Budget }
|
||||
| { ok: false; error: string };
|
||||
|
||||
/**
|
||||
* Validate a raw budget object. Pure, no side effects. Caller obtains either
|
||||
* a guaranteed-safe Budget or a structured error suitable for logging.
|
||||
*
|
||||
* Accepts undefined/null and returns a default unbounded budget (maxHops =
|
||||
* DEFAULT_MAX_HOPS, tokens/usd = Infinity). This preserves backward
|
||||
* compatibility for callers that don't pass a budget at all.
|
||||
*/
|
||||
export function validateBudget(
|
||||
raw: unknown,
|
||||
overrideMaxHops?: unknown,
|
||||
): BudgetValidationResult {
|
||||
// Default unbounded budget when caller passes nothing.
|
||||
if (raw == null && overrideMaxHops == null) {
|
||||
return {
|
||||
ok: true,
|
||||
budget: {
|
||||
maxHops: DEFAULT_MAX_HOPS,
|
||||
maxTokens: Number.POSITIVE_INFINITY,
|
||||
maxUsd: Number.POSITIVE_INFINITY,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// Caller passed a bare maxHops without a budget object — accept it as a
|
||||
// hop-only budget. Useful for "no remote delegation allowed" via
|
||||
// maxHops: 0.
|
||||
if (raw == null && overrideMaxHops != null) {
|
||||
const hopCheck = checkHops(overrideMaxHops);
|
||||
if (!hopCheck.ok) return hopCheck;
|
||||
return {
|
||||
ok: true,
|
||||
budget: {
|
||||
maxHops: hopCheck.value,
|
||||
maxTokens: Number.POSITIVE_INFINITY,
|
||||
maxUsd: Number.POSITIVE_INFINITY,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
if (typeof raw !== 'object' || Array.isArray(raw)) {
|
||||
return { ok: false, error: 'budget must be an object' };
|
||||
}
|
||||
const obj = raw as Record<string, unknown>;
|
||||
|
||||
const hopsRaw = overrideMaxHops ?? obj.maxHops;
|
||||
const hopCheck =
|
||||
hopsRaw == null
|
||||
? { ok: true as const, value: DEFAULT_MAX_HOPS }
|
||||
: checkHops(hopsRaw);
|
||||
if (!hopCheck.ok) return hopCheck;
|
||||
|
||||
const tokensCheck =
|
||||
obj.maxTokens == null
|
||||
? { ok: true as const, value: Number.POSITIVE_INFINITY }
|
||||
: checkBound(obj.maxTokens, MAX_TOKENS_CEILING, 'maxTokens', true);
|
||||
if (!tokensCheck.ok) return tokensCheck;
|
||||
|
||||
const usdCheck =
|
||||
obj.maxUsd == null
|
||||
? { ok: true as const, value: Number.POSITIVE_INFINITY }
|
||||
: checkBound(obj.maxUsd, MAX_USD_CEILING, 'maxUsd', false);
|
||||
if (!usdCheck.ok) return usdCheck;
|
||||
|
||||
return {
|
||||
ok: true,
|
||||
budget: {
|
||||
maxHops: hopCheck.value,
|
||||
maxTokens: tokensCheck.value,
|
||||
maxUsd: usdCheck.value,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function checkHops(
|
||||
v: unknown,
|
||||
): { ok: true; value: number } | { ok: false; error: string } {
|
||||
if (typeof v !== 'number' || !Number.isFinite(v) || !Number.isInteger(v)) {
|
||||
return { ok: false, error: 'maxHops must be a finite integer' };
|
||||
}
|
||||
if (v < 0) return { ok: false, error: 'maxHops must be >= 0' };
|
||||
if (v > MAX_HOPS_CEILING) {
|
||||
return { ok: false, error: `maxHops exceeds ceiling (${MAX_HOPS_CEILING})` };
|
||||
}
|
||||
return { ok: true, value: v };
|
||||
}
|
||||
|
||||
function checkBound(
|
||||
v: unknown,
|
||||
ceiling: number,
|
||||
label: string,
|
||||
requireInteger: boolean,
|
||||
): { ok: true; value: number } | { ok: false; error: string } {
|
||||
if (typeof v !== 'number' || !Number.isFinite(v)) {
|
||||
return { ok: false, error: `${label} must be a finite number` };
|
||||
}
|
||||
if (requireInteger && !Number.isInteger(v)) {
|
||||
return { ok: false, error: `${label} must be an integer` };
|
||||
}
|
||||
if (v < 0) return { ok: false, error: `${label} must be >= 0` };
|
||||
if (v > ceiling) {
|
||||
return { ok: false, error: `${label} exceeds ceiling (${ceiling})` };
|
||||
}
|
||||
return { ok: true, value: v };
|
||||
}
|
||||
|
||||
/**
|
||||
* Enforcement decision returned by enforceBudget. Discriminated so the
|
||||
* caller can map directly to a constant-string error — no remaining-budget
|
||||
* echo on the failure side (anti-oracle).
|
||||
*/
|
||||
export type BudgetEnforcement =
|
||||
| { ok: true; nextHopCount: number; remaining: Budget }
|
||||
| { ok: false; reason: 'HOP_LIMIT_EXCEEDED' | 'BUDGET_EXCEEDED' };
|
||||
|
||||
/**
|
||||
* Atomic (single synchronous function, no internal awaits) hop check.
|
||||
*
|
||||
* Returns the hopCount value the OUTBOUND envelope should carry plus the
|
||||
* remaining budget that downstream peers will see. Callers must use the
|
||||
* returned values; mutating the input Budget is forbidden (it's readonly).
|
||||
*
|
||||
* Contract:
|
||||
* - hopCount represents how many hops the message has already taken.
|
||||
* - The next outbound hop increments it: nextHopCount = hopCount + 1.
|
||||
* - The check fires AFTER increment: nextHopCount must be <= maxHops.
|
||||
* - This means maxHops=0 refuses to forward at all, which matches the
|
||||
* ADR's "no remote delegation allowed" semantic.
|
||||
*
|
||||
* tokensSpent / usdSpent are caller-reported actuals from the immediately
|
||||
* preceding leg (or 0 on the originator). The check fires on the
|
||||
* *post-spend* totals so a leg that overshoots is caught before the next
|
||||
* outbound dispatch.
|
||||
*/
|
||||
export function enforceBudget(
|
||||
budget: Budget,
|
||||
hopCount: number,
|
||||
spent: { tokens: number; usd: number } = { tokens: 0, usd: 0 },
|
||||
): BudgetEnforcement {
|
||||
const nextHopCount = hopCount + 1;
|
||||
if (nextHopCount > budget.maxHops) {
|
||||
return { ok: false, reason: 'HOP_LIMIT_EXCEEDED' };
|
||||
}
|
||||
// Negative spend is a programming error; treat as 0 rather than refunding
|
||||
// the budget (which would let a malicious caller inflate remaining).
|
||||
const ts = Math.max(0, spent.tokens);
|
||||
const us = Math.max(0, spent.usd);
|
||||
const remainingTokens = budget.maxTokens - ts;
|
||||
const remainingUsd = budget.maxUsd - us;
|
||||
if (remainingTokens < 0 || remainingUsd < 0) {
|
||||
return { ok: false, reason: 'BUDGET_EXCEEDED' };
|
||||
}
|
||||
return {
|
||||
ok: true,
|
||||
nextHopCount,
|
||||
remaining: {
|
||||
maxHops: budget.maxHops,
|
||||
maxTokens: remainingTokens,
|
||||
maxUsd: remainingUsd,
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -89,7 +89,7 @@ export function createMcpTools(
|
||||
},
|
||||
{
|
||||
name: 'federation_send',
|
||||
description: 'Send a message to a federated peer through the PII pipeline and security gates',
|
||||
description: 'Send a message to a federated peer through the PII pipeline and security gates. Optional budget controls (ADR-097): maxHops defaults to 8 to prevent recursive delegation; maxTokens/maxUsd cap cumulative spend across the hop chain.',
|
||||
pluginName: '@claude-flow/plugin-agent-federation',
|
||||
version: '1.0.0-alpha.1',
|
||||
inputSchema: {
|
||||
@@ -98,6 +98,33 @@ export function createMcpTools(
|
||||
targetNodeId: { type: 'string', description: 'Target node ID' },
|
||||
messageType: { type: 'string', description: 'Message type (task-assignment, memory-query, context-share, etc.)' },
|
||||
payload: { type: 'object', description: 'Message payload' },
|
||||
// ADR-097 Phase 1: optional budget envelope. Backward compatible —
|
||||
// omitting these preserves the legacy unbounded-tokens/USD path,
|
||||
// with maxHops still defaulting to 8 to defang recursion loops.
|
||||
budget: {
|
||||
type: 'object',
|
||||
description: 'Optional cumulative spend budget for this delegation chain',
|
||||
properties: {
|
||||
maxTokens: { type: 'number', description: 'Hard cap on Σ tokens across hops' },
|
||||
maxUsd: { type: 'number', description: 'Hard cap on Σ USD spent' },
|
||||
},
|
||||
},
|
||||
maxHops: {
|
||||
type: 'number',
|
||||
description: 'Maximum hops this message may travel (default 8, 0 disallows remote delegation)',
|
||||
},
|
||||
hopCount: {
|
||||
type: 'number',
|
||||
description: 'How many hops this message has already taken (0 on the originator)',
|
||||
},
|
||||
spent: {
|
||||
type: 'object',
|
||||
description: 'Caller-reported usage from previous legs (cumulative)',
|
||||
properties: {
|
||||
tokens: { type: 'number' },
|
||||
usd: { type: 'number' },
|
||||
},
|
||||
},
|
||||
},
|
||||
required: ['targetNodeId', 'messageType', 'payload'],
|
||||
},
|
||||
@@ -107,6 +134,12 @@ export function createMcpTools(
|
||||
params['targetNodeId'] as string,
|
||||
params['messageType'] as FederationMessageType,
|
||||
params['payload'],
|
||||
{
|
||||
budget: params['budget'] as { maxTokens?: number; maxUsd?: number } | undefined,
|
||||
maxHops: params['maxHops'] as number | undefined,
|
||||
hopCount: params['hopCount'] as number | undefined,
|
||||
spent: params['spent'] as { tokens?: number; usd?: number } | undefined,
|
||||
},
|
||||
);
|
||||
return textResult(JSON.stringify(result, null, 2), !result.success);
|
||||
},
|
||||
|
||||
Reference in new issue
Block a user