feat(cost-tracker): P6 — ADR-097 Phase 3 federation_spend consumer (v0.14.0)

ADR-097 Phase 1 (federation_send budget envelope) is landed; Phase 3
(federation_spend event bus that publishes per-completion {peerId,
taskId, tokensUsed, usdSpent, ts}) is deferred upstream. This plugin
is the declared consumer of those events. P6 wires the consumer side
so it activates the moment Phase 3 emits.

What lands:

- scripts/federation.mjs — reads namespace `federation-spend` for keys
  `fed-spend-*`, aggregates per-peer 1h/24h/7d rolling windows, flags
  peers exceeding the suspension threshold (default $5/24h per ADR-097
  Phase 2). FED_FORMAT=json for machine-readable; FED_NAMESPACE and
  FED_SUSPEND_THRESHOLD_USD overrides.
- skills/cost-federation/SKILL.md — documents the consumer contract
  (event shape, namespace, key prefix) so upstream knows where to write
- `cost federation` subcommand
- 1 new smoke check (39b — pre-39a)

Live verified by injecting two mock events:
  - peer-alice $0.045 → no flag
  - peer-bob $7.50 → ⚠ flagged over $5/24h threshold

Phase 3 status: not yet emitting upstream. Skill prints "no events
found, will activate when upstream lands" and exits cleanly.

Version: 0.13.0 → 0.14.0. Keywords: + federation.

Co-Authored-By: RuFlo <ruv@ruv.net>
This commit is contained in:
ReuvenandRuFlo committed 2026-05-05 00:10:33 -04:00
1 parent 3ea80a9a9c
commit 1c08043159
5 files changed
+216 -7

No files matched your search

@@ -1,7 +1,7 @@
{
"name": "ruflo-cost-tracker",
"description": "Token usage tracking, model cost attribution per agent, budget alerts, and optimization recommendations — uses memory_* (namespace-routed) for cost-tracking and cost-patterns; pairs with federation budget circuit breaker (ADR-097)",
"version": "0.13.0",
"version": "0.14.0",
"author": {
"name": "ruvnet",
"url": "https://github.com/ruvnet"
@@ -29,6 +29,7 @@
"trend-analysis",
"ci",
"prometheus",
"observability"
"observability",
"federation"
]
}
@@ -55,6 +55,18 @@ Cost tracking commands:
3. The script wraps `npx @claude-flow/cli hooks model-outcome -t ... -m ... -o ...` with explicit-argv spawnSync so quoting is safe
4. Without this, the router doesn't learn from cost-optimize recommendations and the Tier 1 bypass rate doesn't tighten over time
**`cost summary [--format json|markdown]`** -- Single-shot programmatic dump of all cost data. Other plugins/scripts can shell out and parse the JSON.
1. Run `node plugins/ruflo-cost-tracker/scripts/summary.mjs --format json`
2. Output: total_cost_usd, sessionCount, byTier, byModel, topSession, budget, federation aggregate
3. Default `--format markdown`; JSON contract is stable for programmatic consumers
4. ADR-0002 considered an MCP-tool form but deferred (requires v3 source change); this is the plugin-local equivalent
**`cost federation`** -- Consumer-side wiring for ADR-097 Phase 3 federation_spend events. Aggregates per-peer 1h/24h/7d rolling windows and flags peers exceeding the suspension threshold (default $5/24h).
1. Run `node plugins/ruflo-cost-tracker/scripts/federation.mjs`
2. Optional: `FED_FORMAT=json`, `FED_NAMESPACE=federation-spend`, `FED_SUSPEND_THRESHOLD_USD=5.0`
3. Reports gracefully when no events present (Phase 3 not yet landed upstream)
4. Activates automatically when upstream publishes `{peerId, taskId, tokensUsed, usdSpent, ts}` to the `federation-spend` namespace
**`cost export [--prometheus <path>] [--webhook <url>]`** -- Export cost-tracking telemetry to external observability systems.
1. `--prometheus <path>` writes the node_exporter textfile-collector format (gauges + counters with session labels)
2. `--webhook <url>` POSTs JSON; auth via `EXPORT_WEBHOOK_HEADER='K: V'`
+128
View File
@@ -0,0 +1,128 @@
#!/usr/bin/env node
// cost-federation — aggregate per-peer federation_spend events into rolling
// 1h / 24h / 7d windows. Implements the consumer side of ADR-097 Phase 3.
//
// Phase 3 contract (from ADR-097):
// Event: federation_spend = { peerId, taskId, tokensUsed, usdSpent, ts }
// Storage: namespace `federation-spend`, key `fed-spend-<peerId>-<ts>`
// Aggregation: per-peer rolling sums (1h, 24h, 7d)
// Consumer: `cost-report` (this plugin), `breaker` (federation suspension)
//
// Phase 3 isn't landed upstream yet (federation_spend events aren't emitted).
// This script reads whatever IS in the namespace today (empty by default) and
// reports gracefully. The moment upstream emits, this works without changes.
//
// Usage:
// node scripts/federation.mjs # markdown summary
// FED_FORMAT=json node scripts/federation.mjs # JSON
// FED_NAMESPACE=federation-spend (default)
import { spawnSync } from 'node:child_process';
const NS = process.env.FED_NAMESPACE || 'federation-spend';
function memoryList() {
const r = spawnSync('npx', [
'@claude-flow/cli@latest', 'memory', 'list',
'--namespace', NS, '--format', 'json',
], { stdio: ['ignore', 'pipe', 'pipe'], encoding: 'utf-8' });
if (r.status !== 0) return [];
const m = /\[[\s\S]*\]/.exec(r.stdout || '');
if (!m) return [];
try { return JSON.parse(m[0]).map((e) => e.key).filter(Boolean); } catch { return []; }
}
function memoryRetrieve(key) {
const r = spawnSync('npx', [
'@claude-flow/cli@latest', 'memory', 'retrieve',
'--namespace', NS, '--key', key,
], { stdio: ['ignore', 'pipe', 'pipe'], encoding: 'utf-8' });
if (r.status !== 0) return null;
const m = /\{[\s\S]*\}/.exec(r.stdout || '');
if (!m) return null;
try { return JSON.parse(m[0]); } catch { return null; }
}
function gather() {
const keys = memoryList();
const events = keys
.filter((k) => k.startsWith('fed-spend-'))
.map(memoryRetrieve)
.filter(Boolean);
return events;
}
function bucket(events) {
const now = Date.now();
const h = 3600 * 1000;
const day = 24 * h;
const perPeer = {};
for (const e of events) {
const peer = e.peerId || 'unknown';
const ts = e.ts ? new Date(e.ts).getTime() : 0;
if (!perPeer[peer]) perPeer[peer] = {
peer, count1h: 0, count24h: 0, count7d: 0,
tokens1h: 0, tokens24h: 0, tokens7d: 0,
usd1h: 0, usd24h: 0, usd7d: 0,
total: { tokens: 0, usd: 0, count: 0 },
};
const p = perPeer[peer];
const tokens = e.tokensUsed || 0;
const usd = e.usdSpent || 0;
p.total.tokens += tokens; p.total.usd += usd; p.total.count++;
if (ts && now - ts < 7 * day) { p.count7d++; p.tokens7d += tokens; p.usd7d += usd; }
if (ts && now - ts < day) { p.count24h++; p.tokens24h += tokens; p.usd24h += usd; }
if (ts && now - ts < h) { p.count1h++; p.tokens1h += tokens; p.usd1h += usd; }
}
return Object.values(perPeer).sort((a, b) => b.usd24h - a.usd24h);
}
function fmtUsd(n) { return `$${(n || 0).toFixed(4)}`; }
function main() {
const events = gather();
const peers = bucket(events);
const totalUsd24h = peers.reduce((s, p) => s + p.usd24h, 0);
if (process.env.FED_FORMAT === 'json') {
console.log(JSON.stringify({
eventCount: events.length,
peers,
totalUsd24h,
phase3Active: events.length > 0,
}, null, 2));
return;
}
console.log(`# cost-federation — per-peer federation_spend rolling windows`);
console.log('');
console.log(`Namespace: \`${NS}\` · Events: ${events.length} · Peers: ${peers.length}`);
if (events.length === 0) {
console.log('');
console.log('> No federation_spend events found. Phase 3 of ADR-097 is not yet emitting.');
console.log('> This skill is the consumer-side wiring; it will activate the moment upstream');
console.log('> publishes events with shape `{peerId, taskId, tokensUsed, usdSpent, ts}` to');
console.log('> the `' + NS + '` namespace.');
return;
}
console.log('');
console.log(`Total 24h spend: **${fmtUsd(totalUsd24h)}**`);
console.log('');
console.log('| Peer | 1h count | 1h $ | 24h count | 24h $ | 7d count | 7d $ |');
console.log('|---|---:|---:|---:|---:|---:|---:|');
for (const p of peers) {
console.log(`| \`${p.peer}\` | ${p.count1h} | ${fmtUsd(p.usd1h)} | ${p.count24h} | ${fmtUsd(p.usd24h)} | ${p.count7d} | ${fmtUsd(p.usd7d)} |`);
}
console.log('');
console.log('## Suspension threshold check (ADR-097 default $5/24h)');
console.log('');
const SUSPEND_THRESHOLD = parseFloat(process.env.FED_SUSPEND_THRESHOLD_USD || '5.0');
const offenders = peers.filter((p) => p.usd24h > SUSPEND_THRESHOLD);
if (offenders.length === 0) {
console.log(`No peer exceeds the $${SUSPEND_THRESHOLD.toFixed(2)}/24h threshold.`);
} else {
console.log(`**⚠ ${offenders.length} peer(s) exceed $${SUSPEND_THRESHOLD.toFixed(2)}/24h** (Phase 2 breaker would suspend):`);
for (const p of offenders) console.log(`- \`${p.peer}\` — 24h $${p.usd24h.toFixed(2)}`);
}
}
main();
+18 -5
View File
@@ -8,10 +8,10 @@ step() { printf "→ %s ... " "$1"; }
ok() { printf "PASS\n"; PASS=$((PASS+1)); }
bad() { printf "FAIL: %s\n" "$1"; FAIL=$((FAIL+1)); }
step "1. plugin.json declares 0.13.0 with new keywords"
step "1. plugin.json declares 0.14.0 with new keywords"
v=$(grep -E '"version"' "$ROOT/.claude-plugin/plugin.json" | grep -oE '[0-9]+\.[0-9]+\.[0-9]+' | head -1)
if [[ "$v" != "0.13.0" ]]; then
bad "expected 0.13.0, got '$v'"
if [[ "$v" != "0.14.0" ]]; then
bad "expected 0.14.0, got '$v'"
else
miss=""
for k in namespace-routing mcp agentic-flow agent-booster tier1-routing model-routing benchmarking verified telemetry budget; do
@@ -20,9 +20,9 @@ else
[[ -z "$miss" ]] && ok || bad "missing keywords:$miss"
fi
step "2. all eleven skills present with valid frontmatter"
step "2. all twelve skills present with valid frontmatter"
miss=""
for s in cost-report cost-optimize cost-booster-route cost-booster-edit cost-compact-context cost-benchmark cost-track cost-budget-check cost-trend cost-conversation cost-export; do
for s in cost-report cost-optimize cost-booster-route cost-booster-edit cost-compact-context cost-benchmark cost-track cost-budget-check cost-trend cost-conversation cost-export cost-federation; do
f="$ROOT/skills/$s/SKILL.md"
[[ -f "$f" ]] || { miss="$miss missing-$s"; continue; }
for k in 'name:' 'description:' 'allowed-tools:'; do
@@ -321,6 +321,19 @@ step "39. ruflo-cost.md documents 'cost trend' subcommand"
grep -q "cost trend" "$ROOT/commands/ruflo-cost.md" \
&& ok || bad "missing"
step "39b. cost-federation skill + federation.mjs (ADR-097 Phase 3 consumer)"
F1="$ROOT/scripts/federation.mjs"
F2="$ROOT/skills/cost-federation/SKILL.md"
miss=""
[[ -x "$F1" ]] || miss="$miss fed-not-executable"
node --check "$F1" 2>/dev/null || miss="$miss syntax-error"
grep -q "federation_spend" "$F1" || miss="$miss no-event-type"
grep -q "1h\|24h\|7d" "$F1" || miss="$miss no-rolling-windows"
grep -qE "ADR-097|Phase 3" "$F2" || miss="$miss no-adr-ref"
grep -q "fed-spend-" "$F2" || miss="$miss no-key-prefix"
grep -q '^allowed-tools:[[:space:]]*\*' "$F2" && miss="$miss wildcard"
[[ -z "$miss" ]] && ok || bad "$miss"
step "39a. cost-export skill + export.mjs (Prometheus + webhook)"
F1="$ROOT/scripts/export.mjs"
F2="$ROOT/skills/cost-export/SKILL.md"
@@ -0,0 +1,55 @@
---
name: cost-federation
description: Consumer-side wiring for ADR-097 Phase 3 federation_spend events — per-peer rolling windows + suspension-threshold check
argument-hint: ""
allowed-tools: Bash
---
# Cost Federation
ADR-097 Phase 3 specifies a `federation_spend` event bus that publishes one event per `federation_send` completion: `{peerId, taskId, tokensUsed, usdSpent, ts}`. Cost-tracker is the declared consumer — it aggregates per-peer rolling windows (1h / 24h / 7d) and the federation breaker queries that aggregate to suspend peers exceeding the configured threshold.
**Phase 3 isn't landed upstream yet.** This skill is the consumer-side wiring; it activates the moment upstream publishes events to the `federation-spend` namespace. Until then it reports cleanly with a "no events found" notice.
## When to use
- Before opening federation traffic to a new peer — establish baseline.
- Periodically (e.g. `/loop 5m`) to monitor per-peer spend across windows.
- After ADR-097 Phase 2/3 lands — to verify breaker suspension reasoning.
## Steps
1. **Run the script**:
```bash
node plugins/ruflo-cost-tracker/scripts/federation.mjs
```
Optional env:
- `FED_FORMAT=json` — JSON instead of markdown
- `FED_NAMESPACE=federation-spend` — override target namespace
- `FED_SUSPEND_THRESHOLD_USD=5.0` — breaker threshold (ADR-097 default)
2. **Inspect** — per-peer 1h/24h/7d windows of count + USD spent, plus a "Suspension threshold check" block flagging peers over the breaker line.
## Storage contract (consumer side)
When Phase 3 lands, events should be persisted to namespace `federation-spend` with key prefix `fed-spend-` and JSON value matching:
```json
{
"peerId": "<peer-uuid>",
"taskId": "<task-uuid>",
"tokensUsed": 12345,
"usdSpent": 0.0287,
"ts": "2026-05-05T..."
}
```
The script reads any record matching that prefix, regardless of how upstream produces them. Multiple events per peer accumulate cleanly into the rolling-window sums.
## Cross-references
- [ADR-097: Federation budget circuit breaker](../../v3/docs/adr/ADR-097-federation-budget-circuit-breaker.md) — the complete spec
- `cost-report` — same data, different lens (cost-report focuses on local agent spend; this skill on per-peer federated spend)
- `ruflo-federation` plugin — the producer side (when Phase 3 lands)