diff --git a/backend/src/agents/base/BaseAgent.ts b/backend/src/agents/base/BaseAgent.ts index 0c96612..66d963a 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, type VeniceClientLike } from '../../services/venice/index.js'; +import { VeniceClient, type AgentType } from '../../venice/index.js'; +import { HeartbeatClient } from '../heartbeat.js'; export interface BaseAgentConfig { veniceClient?: VeniceClientLike; @@ -23,6 +24,7 @@ export abstract class BaseAgent { protected readonly venice: VeniceClientLike; 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 0badbb1..847566e 100644 --- a/backend/src/api/routes/agents.ts +++ b/backend/src/api/routes/agents.ts @@ -263,7 +263,7 @@ export function createAgentsRouter(options: AgentsRouterOptions = {}): Router { stellarPublicKey: data.stellarPublicKey, reputationScore: 0, lastSeenAt: new Date().toISOString(), - status: "online" + status: 'online' as const }; db.upsert(agent); @@ -271,59 +271,19 @@ export function createAgentsRouter(options: AgentsRouterOptions = {}): Router { res.status(201).json(agent); }); - /** - * @openapi - * /api/agents/{id}: - * delete: - * summary: Delete an agent - * operationId: deleteAgent - * description: > - * Requires a valid Stellar signature proving ownership of the - * agent's registered keypair. The caller must sign the value of - * the `x-challenge` header and pass the base64-encoded signature - * in `x-signature`. - * tags: [Agents] - * security: - * - AgentSignatureAuth: [] - * parameters: - * - in: path - * name: id - * required: true - * schema: { type: string } - * - in: header - * name: x-challenge - * required: true - * schema: { type: string } - * description: The challenge string the caller must sign - * - in: header - * name: x-signature - * required: true - * schema: { type: string } - * description: Base64-encoded signature of the challenge, signed with the agent's Stellar keypair - * responses: - * 200: - * description: Agent deleted - * content: - * application/json: - * schema: - * type: object - * properties: - * message: - * type: string - * example: Agent deleted successfully - * 401: - * description: Missing, invalid, or malformed signature - * content: - * application/json: - * schema: - * $ref: '#/components/schemas/Error' - * 404: - * description: Agent not found - * content: - * application/json: - * schema: - * $ref: '#/components/schemas/Error' - */ + // 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 c4689ad..5e2b9f1 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 9f4dec4..701773b 100644 --- a/backend/src/db/agents.ts +++ b/backend/src/db/agents.ts @@ -9,7 +9,7 @@ export interface AgentRecord { stellarPublicKey: string; reputationScore: number; lastSeenAt: string; - status?: string; + status: 'online' | 'offline'; } let _agentDb: Database.Database | null = null; @@ -27,7 +27,7 @@ export function getAgentDb(dbPath?: string): Database.Database { stellarPublicKey TEXT NOT NULL, reputationScore REAL NOT NULL DEFAULT 0, lastSeenAt TEXT NOT NULL, - status TEXT NOT NULL DEFAULT 'offline' + status TEXT NOT NULL DEFAULT 'online' ) `); try { @@ -47,10 +47,10 @@ 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; - markAllOffline(): void; + markOffline(olderThan: string): void; } export function createAgentDb(db: Database.Database): AgentDb { @@ -83,7 +83,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[] = []; @@ -99,6 +99,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 => ({ @@ -116,8 +120,8 @@ export function createAgentDb(db: Database.Database): AgentDb { db.prepare("UPDATE agents SET reputationScore = reputationScore + ? WHERE id = ?").run(delta, id); }, - markAllOffline(): void { - db.prepare("UPDATE agents SET status = 'offline' WHERE status = 'online'").run(); + 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 4bef6de..216fa76 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -5,12 +5,10 @@ */ import { createApp } from "./api/app"; -import { initializeAgents } from "./agents"; +import { initializeAgents, globalAgentRegistry } from "./agents"; import { startAgentSync, stopAgentSync } from "./registry/sync"; import { loadConfig, getConfig } from "./config"; -import { closeDb } from "./db"; -import { closeAgentDb, createAgentDb, getAgentDb } from "./db/agents"; -import { closeTaskDb, createTaskDb, getTaskDb } from "./db/tasks"; +import { AgentCleanupService } from "./services/agentCleanup"; async function main() { // ── Validate env config at startup ────────────────────────────────────────── @@ -27,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, close } = createApp(); @@ -43,10 +45,30 @@ 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 ────────────────────────────────────────────────────── - setupGracefulShutdown(httpServer, close, config); + const shutdown = (signal: string) => { + console.log(`[ai-net-backend] Received ${signal}, shutting down gracefully...`); + const timeout = setTimeout(() => { + console.error("[ai-net-backend] Forced shutdown after 10s timeout"); + process.exit(1); + }, 10_000); + + cleanupService.stop(); + globalAgentRegistry.shutdown(); + stopAgentSync(); + + httpServer.close(() => { + clearTimeout(timeout); + console.log("[ai-net-backend] Server closed."); + process.exit(0); + }); + }; + + process.on("SIGTERM", () => shutdown("SIGTERM")); + process.on("SIGINT", () => shutdown("SIGINT")); } catch (error) { console.error("[ai-net-backend] Failed to start server:", error); 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 5e1e717..1ced3ae 100644 --- a/backend/tests/agents.test.ts +++ b/backend/tests/agents.test.ts @@ -13,7 +13,7 @@ const codingAgent: AgentRecord = { stellarPublicKey: "GBXX...", reputationScore: 0, lastSeenAt: new Date().toISOString(), - status: "offline" + status: "online" }; function createTestApp(initialAgents: AgentRecord[] = [], healthTimeoutMs = 500) { @@ -27,7 +27,7 @@ function createTestApp(initialAgents: AgentRecord[] = [], healthTimeoutMs = 500) stellarPublicKey TEXT NOT NULL, reputationScore REAL NOT NULL DEFAULT 0, lastSeenAt TEXT NOT NULL, - status TEXT NOT NULL DEFAULT 'offline' + status TEXT NOT NULL DEFAULT 'online' ) `); const db = createAgentDb(rawDb); @@ -125,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" }); + }); });