From 78f28708ba96c1af5defb170fa2d225e14e57d70 Mon Sep 17 00:00:00 2001 From: dubemoyibe-star Date: Sun, 19 Jul 2026 12:13:43 +0100 Subject: [PATCH] feat: implement agent heartbeat and automatic dead-agent cleanup --- backend/src/agents/base/BaseAgent.ts | 17 +++++++++ backend/src/agents/heartbeat.ts | 51 ++++++++++++++++++++++++++ backend/src/agents/registry.ts | 14 +++++++ backend/src/api/routes/agents.ts | 16 +++++++- backend/src/coordinator/coordinator.ts | 4 +- backend/src/db/agents.ts | 24 +++++++++--- backend/src/index.ts | 14 ++++++- backend/src/registry/sync.ts | 17 +++++---- backend/src/services/agentCleanup.ts | 50 +++++++++++++++++++++++++ backend/src/types/agent.ts | 1 + backend/tests/agents.test.ts | 25 ++++++++++++- 11 files changed, 213 insertions(+), 20 deletions(-) create mode 100644 backend/src/agents/heartbeat.ts create mode 100644 backend/src/services/agentCleanup.ts diff --git a/backend/src/agents/base/BaseAgent.ts b/backend/src/agents/base/BaseAgent.ts index 566231d..7a3c8c9 100644 --- a/backend/src/agents/base/BaseAgent.ts +++ b/backend/src/agents/base/BaseAgent.ts @@ -1,5 +1,6 @@ import { z } from 'zod'; import { VeniceClient, type AgentType } from '../../venice/index.js'; +import { HeartbeatClient } from '../heartbeat.js'; export interface BaseAgentConfig { veniceClient?: VeniceClient; @@ -23,6 +24,7 @@ export abstract class BaseAgent { protected readonly venice: VeniceClient; protected readonly apiBaseUrl: string; protected readonly agentId: string; + private readonly heartbeatClient: HeartbeatClient | null = null; constructor(config: BaseAgentConfig = {}) { if (config.veniceClient) { @@ -39,6 +41,13 @@ export abstract class BaseAgent { } this.apiBaseUrl = config.apiBaseUrl ?? 'http://127.0.0.1:3001'; this.agentId = config.agentId ?? `${this.getCapability()}-agent-1`; + + if (config.apiBaseUrl) { + this.heartbeatClient = new HeartbeatClient({ + apiBaseUrl: this.apiBaseUrl, + agentId: this.agentId, + }); + } } abstract execute(task: AgentTask): Promise; @@ -85,6 +94,14 @@ export abstract class BaseAgent { } } + startHeartbeat(): void { + this.heartbeatClient?.start(); + } + + stopHeartbeat(): void { + this.heartbeatClient?.stop(); + } + protected validateOutput(raw: unknown): unknown | null { const result = this.getOutputSchema().safeParse(raw); return result.success ? result.data : null; diff --git a/backend/src/agents/heartbeat.ts b/backend/src/agents/heartbeat.ts new file mode 100644 index 0000000..365edea --- /dev/null +++ b/backend/src/agents/heartbeat.ts @@ -0,0 +1,51 @@ +export interface HeartbeatClientOptions { + apiBaseUrl: string; + agentId: string; + intervalMs?: number; +} + +export class HeartbeatClient { + private readonly apiBaseUrl: string; + private readonly agentId: string; + private readonly intervalMs: number; + private interval: NodeJS.Timeout | null = null; + private stopped = false; + + constructor(options: HeartbeatClientOptions) { + this.apiBaseUrl = options.apiBaseUrl.replace(/\/$/, ''); + this.agentId = options.agentId; + this.intervalMs = options.intervalMs ?? 30_000; + } + + start(): void { + if (this.interval) return; + this.stopped = false; + this.send(); + this.interval = setInterval(() => { + this.send(); + }, this.intervalMs); + } + + stop(): void { + if (this.interval) { + clearInterval(this.interval); + this.interval = null; + } + this.stopped = true; + } + + private async send(): Promise { + if (this.stopped) return; + + try { + const response = await fetch(`${this.apiBaseUrl}/api/agents/${encodeURIComponent(this.agentId)}/heartbeat`, { + method: 'POST', + }); + if (!response.ok) { + console.warn(`[Heartbeat] Heartbeat failed for ${this.agentId}: ${response.status}`); + } + } catch (err) { + console.warn(`[Heartbeat] Heartbeat error for ${this.agentId}:`, err instanceof Error ? err.message : 'unknown'); + } + } +} diff --git a/backend/src/agents/registry.ts b/backend/src/agents/registry.ts index 19658ff..7d59609 100644 --- a/backend/src/agents/registry.ts +++ b/backend/src/agents/registry.ts @@ -51,6 +51,7 @@ export class AgentRegistry { const registrations = this.agents.map(async ({ instance, capability }) => { try { await instance.register(); + instance.startHeartbeat(); } catch (error) { console.error(`[AgentRegistry] Failed to register ${capability} agent:`, error instanceof Error ? error.message : 'unknown'); } @@ -61,6 +62,19 @@ export class AgentRegistry { } } + /** + * Stop heartbeats for all agents. + */ + async shutdown(): Promise { + for (const { instance } of this.agents) { + try { + instance.stopHeartbeat(); + } catch { + // ignore shutdown errors + } + } + } + /** * Get all registered agents. */ diff --git a/backend/src/api/routes/agents.ts b/backend/src/api/routes/agents.ts index c681a32..7809e70 100644 --- a/backend/src/api/routes/agents.ts +++ b/backend/src/api/routes/agents.ts @@ -114,7 +114,8 @@ export function createAgentsRouter(options: AgentsRouterOptions = {}): Router { endpoint: data.endpoint, stellarPublicKey: data.stellarPublicKey, reputationScore: 0, - lastSeenAt: new Date().toISOString() + lastSeenAt: new Date().toISOString(), + status: 'online' as const }; db.upsert(agent); @@ -122,6 +123,19 @@ export function createAgentsRouter(options: AgentsRouterOptions = {}): Router { res.status(201).json(agent); }); + // POST /api/agents/:id/heartbeat + router.post("/:id/heartbeat", (req: Request, res: Response): void => { + const db = getDb(); + const agent = db.findById(req.params.id); + if (!agent) { + res.status(404).json({ error: "Agent not found" }); + return; + } + + db.upsert({ ...agent, lastSeenAt: new Date().toISOString(), status: 'online' }); + res.status(204).send(); + }); + // DELETE /api/agents/:id router.delete("/:id", (req: Request, res: Response): void => { const db = getDb(); diff --git a/backend/src/coordinator/coordinator.ts b/backend/src/coordinator/coordinator.ts index 4ae9ce8..5327714 100644 --- a/backend/src/coordinator/coordinator.ts +++ b/backend/src/coordinator/coordinator.ts @@ -378,7 +378,9 @@ export class Coordinator { throw new Error(`No agent registry configured for type: ${agentType}`); } - const agents = sortByCost(await this.agentRegistry.getAgents(agentType)); + const agents = sortByCost(await this.agentRegistry.getAgents(agentType)).filter( + (agent) => agent.status === 'online' + ); if (agents.length === 0) { throw new Error(`No agent registered for type: ${agentType}`); } diff --git a/backend/src/db/agents.ts b/backend/src/db/agents.ts index 8dcb89d..842322a 100644 --- a/backend/src/db/agents.ts +++ b/backend/src/db/agents.ts @@ -9,6 +9,7 @@ export interface AgentRecord { stellarPublicKey: string; reputationScore: number; lastSeenAt: string; + status: 'online' | 'offline'; } let _agentDb: Database.Database | null = null; @@ -25,7 +26,8 @@ export function getAgentDb(dbPath?: string): Database.Database { endpoint TEXT NOT NULL, stellarPublicKey TEXT NOT NULL, reputationScore REAL NOT NULL DEFAULT 0, - lastSeenAt TEXT NOT NULL + lastSeenAt TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'online' ) `); } @@ -40,23 +42,25 @@ export function closeAgentDb(): void { export interface AgentDb { upsert(agent: AgentRecord): void; findById(id: string): AgentRecord | undefined; - list(filters?: { capability?: string; minReputation?: number; maxPriceXLM?: number }): AgentRecord[]; + list(filters?: { capability?: string; minReputation?: number; maxPriceXLM?: number; status?: string }): AgentRecord[]; delete(id: string): void; updateReputation(id: string, delta: number): void; + markOffline(olderThan: string): void; } export function createAgentDb(db: Database.Database): AgentDb { return { upsert(agent: AgentRecord): void { db.prepare(` - INSERT INTO agents (id, capabilities, pricingXLM, endpoint, stellarPublicKey, reputationScore, lastSeenAt) - VALUES (@id, @capabilities, @pricingXLM, @endpoint, @stellarPublicKey, @reputationScore, @lastSeenAt) + INSERT INTO agents (id, capabilities, pricingXLM, endpoint, stellarPublicKey, reputationScore, lastSeenAt, status) + VALUES (@id, @capabilities, @pricingXLM, @endpoint, @stellarPublicKey, @reputationScore, @lastSeenAt, @status) ON CONFLICT(id) DO UPDATE SET capabilities = excluded.capabilities, pricingXLM = excluded.pricingXLM, endpoint = excluded.endpoint, stellarPublicKey = excluded.stellarPublicKey, - lastSeenAt = excluded.lastSeenAt + lastSeenAt = excluded.lastSeenAt, + status = excluded.status `).run({ ...agent, capabilities: JSON.stringify(agent.capabilities) @@ -72,7 +76,7 @@ export function createAgentDb(db: Database.Database): AgentDb { }; }, - list(filters?: { capability?: string; minReputation?: number; maxPriceXLM?: number }): AgentRecord[] { + list(filters?: { capability?: string; minReputation?: number; maxPriceXLM?: number; status?: string }): AgentRecord[] { let query = "SELECT * FROM agents WHERE 1=1"; const params: any[] = []; @@ -88,6 +92,10 @@ export function createAgentDb(db: Database.Database): AgentDb { query += " AND EXISTS (SELECT 1 FROM json_each(capabilities) WHERE value = ?)"; params.push(filters.capability); } + if (filters?.status !== undefined) { + query += " AND status = ?"; + params.push(filters.status); + } const rows = db.prepare(query).all(...params) as any[]; return rows.map(row => ({ @@ -102,6 +110,10 @@ export function createAgentDb(db: Database.Database): AgentDb { updateReputation(id: string, delta: number): void { db.prepare("UPDATE agents SET reputationScore = reputationScore + ? WHERE id = ?").run(delta, id); + }, + + markOffline(olderThan: string): void { + db.prepare("UPDATE agents SET status = 'offline' WHERE lastSeenAt < ? AND status = 'online'").run(olderThan); } }; } diff --git a/backend/src/index.ts b/backend/src/index.ts index 10fbe46..e53b7f7 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -5,9 +5,10 @@ */ import { createApp } from "./api/app"; -import { initializeAgents } from "./agents"; -import { startAgentSync } from "./registry/sync"; +import { initializeAgents, globalAgentRegistry } from "./agents"; +import { startAgentSync, stopAgentSync } from "./registry/sync"; import { loadConfig, getConfig } from "./config"; +import { AgentCleanupService } from "./services/agentCleanup"; async function main() { // ── Validate env config at startup ────────────────────────────────────────── @@ -24,6 +25,10 @@ async function main() { console.log("[ai-net-backend] Initializing agents..."); await initializeAgents(); + // Start agent cleanup service + const cleanupService = new AgentCleanupService(); + cleanupService.start(); + // Create and start the server const { httpServer } = createApp(); @@ -40,6 +45,7 @@ async function main() { console.log(" - POST /api/agents/register - Register new agents"); console.log(" - GET /api/agents - List all agents"); console.log(" - GET /api/agents/capability/:type - Find agents by capability"); + console.log(" - POST /api/agents/:id/heartbeat - Agent heartbeat"); }); // ── Graceful shutdown ────────────────────────────────────────────────────── @@ -50,6 +56,10 @@ async function main() { process.exit(1); }, 10_000); + cleanupService.stop(); + globalAgentRegistry.shutdown(); + stopAgentSync(); + httpServer.close(() => { clearTimeout(timeout); console.log("[ai-net-backend] Server closed."); diff --git a/backend/src/registry/sync.ts b/backend/src/registry/sync.ts index 72b82db..c24540c 100644 --- a/backend/src/registry/sync.ts +++ b/backend/src/registry/sync.ts @@ -58,14 +58,15 @@ export function startAgentSync(): void { if (val && typeof val === "object" && val.id) { db.upsert({ - id: val.id, - capabilities: Array.isArray(val.capabilities) ? val.capabilities : (val.capabilities ? [val.capabilities] : []), - pricingXLM: Number(val.pricingXLM) || 0, - endpoint: val.endpoint || "", - stellarPublicKey: val.stellarPublicKey || "", - reputationScore: Number(val.reputationScore) || 0, - lastSeenAt: new Date().toISOString() - }); + id: val.id, + capabilities: Array.isArray(val.capabilities) ? val.capabilities : (val.capabilities ? [val.capabilities] : []), + pricingXLM: Number(val.pricingXLM) || 0, + endpoint: val.endpoint || "", + stellarPublicKey: val.stellarPublicKey || "", + reputationScore: Number(val.reputationScore) || 0, + lastSeenAt: new Date().toISOString(), + status: 'online' + }); } } } catch (e) { diff --git a/backend/src/services/agentCleanup.ts b/backend/src/services/agentCleanup.ts new file mode 100644 index 0000000..d5cf697 --- /dev/null +++ b/backend/src/services/agentCleanup.ts @@ -0,0 +1,50 @@ +import { getAgentDb, createAgentDb } from '../db/agents'; +import { createLogger } from '../utils/logger'; + +export interface AgentCleanupOptions { + intervalMs?: number; + ttlMs?: number; +} + +export class AgentCleanupService { + private readonly intervalMs: number; + private readonly ttlMs: number; + private interval: NodeJS.Timeout | null = null; + private stopped = false; + private readonly log = createLogger({ component: 'AgentCleanup' }); + + constructor(options: AgentCleanupOptions = {}) { + this.intervalMs = options.intervalMs ?? 60_000; + this.ttlMs = options.ttlMs ?? 90_000; + } + + start(): void { + if (this.interval) return; + this.stopped = false; + this.tick(); + this.interval = setInterval(() => { + this.tick(); + }, this.intervalMs); + } + + stop(): void { + if (this.interval) { + clearInterval(this.interval); + this.interval = null; + } + this.stopped = true; + } + + private tick(): void { + if (this.stopped) return; + + try { + const cutoff = new Date(Date.now() - this.ttlMs).toISOString(); + const db = createAgentDb(getAgentDb()); + db.markOffline(cutoff); + this.log.info({ cutoff }, 'marked stale agents offline'); + } catch (err) { + this.log.error({ err }, 'cleanup tick failed'); + } + } +} diff --git a/backend/src/types/agent.ts b/backend/src/types/agent.ts index da4227f..0149a37 100644 --- a/backend/src/types/agent.ts +++ b/backend/src/types/agent.ts @@ -3,6 +3,7 @@ export interface AgentRegistration { type: string; endpoint: string; cost: number; + status: 'online' | 'offline'; } export interface AgentRegistry { diff --git a/backend/tests/agents.test.ts b/backend/tests/agents.test.ts index 06bf52b..1ced3ae 100644 --- a/backend/tests/agents.test.ts +++ b/backend/tests/agents.test.ts @@ -12,7 +12,8 @@ const codingAgent: AgentRecord = { endpoint: "http://127.0.0.1:3001/health", stellarPublicKey: "GBXX...", reputationScore: 0, - lastSeenAt: new Date().toISOString() + lastSeenAt: new Date().toISOString(), + status: "online" }; function createTestApp(initialAgents: AgentRecord[] = [], healthTimeoutMs = 500) { @@ -25,7 +26,8 @@ function createTestApp(initialAgents: AgentRecord[] = [], healthTimeoutMs = 500) endpoint TEXT NOT NULL, stellarPublicKey TEXT NOT NULL, reputationScore REAL NOT NULL DEFAULT 0, - lastSeenAt TEXT NOT NULL + lastSeenAt TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'online' ) `); const db = createAgentDb(rawDb); @@ -123,4 +125,23 @@ describe("Agents API route", () => { expect(response.status).toBe(404); expect(response.body).toEqual({ error: "Agent not found" }); }); + + it("returns 204 and updates lastSeenAt on heartbeat", async () => { + const app = createTestApp([codingAgent]); + const before = new Date(); + + const response = await request(app).post("/api/agents/coding-1/heartbeat"); + + expect(response.status).toBe(204); + const updated = await request(app).get("/api/agents/coding-1"); + expect(new Date(updated.body.lastSeenAt).getTime()).toBeGreaterThanOrEqual(before.getTime()); + expect(updated.body.status).toBe("online"); + }); + + it("returns 404 when sending heartbeat to an unknown agent", async () => { + const response = await request(createTestApp()).post("/api/agents/missing/heartbeat"); + + expect(response.status).toBe(404); + expect(response.body).toEqual({ error: "Agent not found" }); + }); });