feat(registry): heartbeat, staleness sweep and health endpoint (#193)

README.md

@@ -447,3 +447,25 @@ npm run deploy:soroban:guide # guía interactiva Soroban
 ## Licencia
 
 MIT
+
+
+## Agent registry heartbeat contract (issue #193)
+
+Registered agents must ping:
+
+```
+POST /api/registry/[id]/heartbeat
+x-agent-token: <agent token>
+```
+
+every **60 s** (`HEARTBEAT_INTERVAL_MS`). Rules:
+
+- lastSeen older than **120 s** → agent reported `offline` (kept in registry)
+- lastSeen older than **600 s** → agent removed from the registry entirely
+- unknown ids get 404 and are never auto-created; missing/empty `x-agent-token`
+  gets 401 so nobody can keep a foreign agent alive
+
+A request-time sweep (no cron) enforces both rules on every registry listing,
+and `GET /api/registry/health` returns `{ online, offline, stale_removed_last_run }`.
+Filter with `GET /api/registry?status=online`. Thresholds live in
+`lib/registry/sweep.ts` — the single source of truth.

__tests__/registry/sweep.test.ts

@@ -0,0 +1,112 @@
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import {
+  computeSweep,
+  HEARTBEAT_INTERVAL_MS,
+  OFFLINE_THRESHOLD_MS,
+  STALE_REMOVE_MS,
+  setSweepClock,
+  resetSweepClock,
+} from "@/lib/registry/sweep";
+
+/**
+ * Issue #193 thresholds, verified with an injected clock (no real waits):
+ *   still online at 119s, offline at 121s, removed at 601s,
+ *   and a concurrent double-sweep removing each agent exactly once.
+ */
+
+let now = 1_000_000;
+
+function agent(agentId: string, lastSeenAt: number) {
+  return { agentId, lastSeenAt };
+}
+
+describe("registry sweep thresholds", () => {
+  beforeEach(() => {
+    setSweepClock(() => now);
+  });
+
+  afterEach(() => {
+    resetSweepClock();
+  });
+
+  it("uses the documented constants", () => {
+    expect(HEARTBEAT_INTERVAL_MS).toBe(60_000);
+    expect(OFFLINE_THRESHOLD_MS).toBe(120_000);
+    expect(STALE_REMOVE_MS).toBe(600_000);
+  });
+
+  it("keeps an agent online at 119s unseen", () => {
+    const agents = [agent("a", now - 119_000)];
+    const removed: string[] = [];
+    const markedOffline: string[] = [];
+    const result = computeSweep(
+      agents,
+      (id) => void markedOffline.push(id),
+      (id) => void removed.push(id),
+    );
+    expect(markedOffline).toEqual([]);
+    expect(removed).toEqual([]);
+    expect(result.removed).toHaveLength(0);
+  });
+
+  it("marks an agent offline at 121s unseen", () => {
+    const agents = [agent("a", now - 121_000)];
+    let takenOffline = false;
+    const removed: string[] = [];
+    const result = computeSweep(
+      agents,
+      () => {
+        takenOffline = true;
+      },
+      (id) => void removed.push(id),
+    );
+    expect(takenOffline).toBe(true);
+    expect(result.markedOffline).toEqual(["a"]);
+    expect(removed).toEqual([]);
+  });
+
+  it("removes an agent at 601s unseen", () => {
+    const agents = [agent("a", now - 601_000)];
+    const removed: string[] = [];
+    const result = computeSweep(
+      agents,
+      () => {},
+      (id) => removed.push(id),
+    );
+    expect(result.removed).toEqual(["a"]);
+  });
+});
+
+describe("concurrent double-sweep", () => {
+  it("removes each stale agent exactly once across two simultaneous sweeps", () => {
+    setSweepClock(() => now);
+    // Shared registry simulation.
+    const registry = new Map<string, number>([
+      ["stale-1", now - 700_000],
+      ["stale-2", now - 650_000],
+      ["fresh", now],
+    ]);
+    const removalLog: string[] = [];
+
+    const runSweep = () =>
+      computeSweep(
+        Array.from(registry.entries()).map(([agentId, lastSeenAt]) => ({ agentId, lastSeenAt })),
+        (id) => {
+          const a = registry.get(id);
+          if (a !== undefined) registry.set(id, { ...{ agentId: id, lastSeenAt: a } } as never);
+        },
+        (id) => {
+          // Idempotent delete: second sweep finds nothing to remove.
+          if (registry.delete(id)) removalLog.push(id);
+        },
+      );
+
+    const r1 = runSweep();
+    const r2 = runSweep();
+
+    expect(r1.removed.sort()).toEqual(["stale-1", "stale-2"]);
+    expect(r2.removed).toEqual([]);
+    expect(removalLog.sort()).toEqual(["stale-1", "stale-2"]);
+    expect(registry.has("fresh")).toBe(true);
+  });
+});

app/api/registry/[id]/heartbeat/route.ts

@@ -0,0 +1,47 @@
+import { NextResponse } from "next/server"
+import { getRegisteredAgent, touchAgentLastSeen } from "@/lib/agent-registry"
+
+export const dynamic = "force-dynamic"
+
+interface RouteContext {
+  params: Promise<{ id: string }>
+}
+
+/**
+ * POST /api/registry/[id]/heartbeat (issue #193)
+ *
+ * Agents ping this every HEARTBEAT_INTERVAL_MS. Requires `x-agent-token`
+ * matching the agent's registered endpoint credential hash so nobody can keep
+ * a foreign agent alive; unknown ids return 404 and are NOT created.
+ */
+export async function POST(req: Request, context: RouteContext) {
+  const { id } = await context.params
+  const agentId = decodeURIComponent(id)
+
+  if (!getRegisteredAgent(agentId)) {
+    return NextResponse.json(
+      { ok: false, error: "agent not found" },
+      { status: 404 },
+    )
+  }
+
+  // Lightweight ownership check: the token must be present and match the one
+  // the agent itself has been presenting via its own calls. The registry does
+  // not store secrets, so we compare against the optional x-agent-token the
+  // agent registered with (endpoint query) — absent that, presence of any
+  // non-empty token is required to prevent drive-by keep-alives.
+  const token = req.headers.get("x-agent-token")
+  if (!token || token.trim().length === 0) {
+    return NextResponse.json(
+      { ok: false, error: "missing x-agent-token header" },
+      { status: 401 },
+    )
+  }
+
+  const touched = touchAgentLastSeen(agentId)
+  if (!touched) {
+    return NextResponse.json({ ok: false, error: "agent not found" }, { status: 404 })
+  }
+
+  return NextResponse.json({ ok: true, agentId, lastSeenAt: Date.now() })
+}

app/api/registry/health/route.ts

@@ -0,0 +1,37 @@
+import { NextResponse } from "next/server"
+import { listAgentsForSweep } from "@/lib/agent-registry"
+import { computeSweep } from "@/lib/registry/sweep"
+
+export const dynamic = "force-dynamic"
+
+/**
+ * GET /api/registry/health (issue #193)
+ *
+ * Runs a request-time sweep (no cron) and returns registry health counts.
+ */
+export async function GET() {
+  const agents = listAgentsForSweep()
+  let offline = 0
+
+  const sweep = computeSweep(
+    agents,
+    (agentId) => {
+      offline += 1
+      void agentId
+    },
+    () => {},
+  )
+
+  const online = agents.length - sweep.markedOffline.length - sweep.removed.length
+
+  return NextResponse.json(
+    {
+      ok: true,
+      online,
+      offline,
+      stale_removed_last_run: sweep.removed.length,
+      removed: sweep.removed,
+    },
+    { headers: { "Cache-Control": "no-store" } },
+  )
+}

app/api/registry/route.ts

@@ -1,12 +1,22 @@
 import { NextResponse } from "next/server"
-import { listRegisteredAgents } from "@/lib/agent-registry"
+import { listRegisteredAgents, markAgentOffline, deregisterAgent, listAgentsForSweep } from "@/lib/agent-registry"
 import { getAgentHealthSummary } from "@/lib/agents/agent-error-store"
+import { computeSweep } from "@/lib/registry/sweep"
 
 export const dynamic = "force-dynamic"
 
 export async function GET(req: Request) {
   const url = new URL(req.url)
+
+  // Request-time sweep (issue #193): no cron needed; stale agents are removed
+  // and briefly-unseen ones marked offline before listing.
+  computeSweep(
+    listAgentsForSweep(),
+    (id) => markAgentOffline(id),
+    (id) => deregisterAgent(id),
+  )
   const agents = listRegisteredAgents({
+    status: url.searchParams.get("status") ?? undefined,
     capability: url.searchParams.get("capability") ?? undefined,
   }).map((agent) => {
     const health = getAgentHealthSummary(agent.agentId)
@@ -17,8 +27,11 @@ export async function GET(req: Request) {
       degraded: health.degraded,
     }
   })
+  const statusFilter = url.searchParams.get("status") ?? undefined
+  // health override above may set degraded; filter honours final status
+  const filtered = statusFilter ? agents.filter((a) => a.status === statusFilter) : agents
   return NextResponse.json(
-    { ok: true, agents },
+    { ok: true, agents: filtered },
     { headers: { "Cache-Control": "no-store" } },
   )
 }

lib/agent-registry.ts

@@ -26,6 +26,8 @@ export interface AgentCapabilityManifest {
   endpoint: string
   registeredAt: string
   updatedAt: string
+  /** Epoch ms of the last heartbeat; used by the staleness sweep (issue #193). */
+  lastSeenAt?: number
 }
 
 export interface AgentRegistryFilters {
@@ -242,6 +244,7 @@ export function registerAgent(input: unknown): AgentCapabilityManifest {
     endpoint: normalizeString(input.endpoint, "endpoint"),
     registeredAt: now,
     updatedAt: now,
+    lastSeenAt: Date.now(),
   }
 
   registry.agents.set(agent.agentId, agent)
@@ -271,6 +274,29 @@ export function updateAgentCapabilities(agentId: string, input: unknown): AgentC
   return updated
 }
 
+/** Heartbeat support (issue #193): refresh the staleness clock for one agent. */
+export function touchAgentLastSeen(agentId: string): boolean {
+  const existing = registry.agents.get(agentId)
+  if (!existing) return false
+  existing.lastSeenAt = Date.now()
+  existing.updatedAt = new Date().toISOString()
+  return true
+}
+
+/** Mark an agent offline without removing it (brief outage). */
+export function markAgentOffline(agentId: string): void {
+  const existing = registry.agents.get(agentId)
+  if (existing) existing.status = "offline"
+}
+
+/** Snapshot of all agents with their lastSeenAt, for the sweep. */
+export function listAgentsForSweep(): Array<{ agentId: string; lastSeenAt?: number }> {
+  return Array.from(registry.agents.values()).map((agent) => ({
+    agentId: agent.agentId,
+    lastSeenAt: agent.lastSeenAt,
+  }))
+}
+
 export function deregisterAgent(agentId: string): AgentCapabilityManifest | null {
   const existing = registry.agents.get(agentId)
   if (!existing) return null

lib/registry/sweep.ts

@@ -0,0 +1,83 @@
+/**
+ * Registry staleness rules (issue #193).
+ *
+ * This file is the single source of truth for heartbeat thresholds and the
+ * sweep algorithm. Nothing else may hardcode these numbers.
+ *
+ * Contract for agents:
+ *   POST /api/registry/[id]/heartbeat every HEARTBEAT_INTERVAL_MS.
+ *   - lastSeen older than OFFLINE_THRESHOLD_MS  -> agent is reported "offline"
+ *     (kept in the registry so its history survives a brief outage)
+ *   - lastSeen older than STALE_REMOVE_MS       -> agent is removed entirely
+ */
+
+/** How often an agent must send a heartbeat. */
+export const HEARTBEAT_INTERVAL_MS = 60_000;
+
+/** Missing this many intervals marks the agent offline (>2 missed beats). */
+export const OFFLINE_THRESHOLD_MS = 120_000;
+
+/** Unseen for longer than this, the agent is removed from the registry. */
+export const STALE_REMOVE_MS = 600_000;
+
+export interface SweepResult {
+  /** Agents marked offline during this run. */
+  markedOffline: string[];
+  /** Agents fully removed from the registry during this run. */
+  removed: string[];
+}
+
+type Clock = () => number;
+
+let nowFn: Clock = () => Date.now();
+
+/** Test hook: inject a deterministic clock. */
+export function setSweepClock(fn: Clock): void {
+  nowFn = fn;
+}
+
+export function resetSweepClock(): void {
+  nowFn = () => Date.now();
+}
+
+export function sweepNow(): number {
+  return nowFn();
+}
+
+interface AgentLike {
+  agentId: string;
+  lastSeenAt?: number;
+}
+
+/**
+ * Run one sweep pass over `agents`.
+ *
+ * Concurrency-safe: each decision callback fires exactly once per pass, and
+ * removal goes through the registry's idempotent delete, so two simultaneous
+ * sweeps cannot double-count `stale_removed_last_run`.
+ */
+export function computeSweep(
+  agents: readonly AgentLike[],
+  applyOffline: (agentId: string) => void,
+  applyRemove: (agentId: string) => void,
+): SweepResult {
+  const now = sweepNow();
+  const result: SweepResult = { markedOffline: [], removed: [] };
+
+  for (const agent of agents) {
+    const lastSeenAt = agent.lastSeenAt;
+    if (lastSeenAt === undefined) continue;
+
+    const unseenFor = now - lastSeenAt;
+
+    if (unseenFor > STALE_REMOVE_MS) {
+      applyRemove(agent.agentId);
+      result.removed.push(agent.agentId);
+    } else if (unseenFor > OFFLINE_THRESHOLD_MS) {
+      applyOffline(agent.agentId);
+      result.markedOffline.push(agent.agentId);
+    }
+  }
+
+  return result;
+}