diff --git a/.gitignore b/.gitignore index a5b2f653..d0f71961 100644 --- a/.gitignore +++ b/.gitignore @@ -65,3 +65,4 @@ uploads/ # keep infra .env.example !.env.example +.gitconfig diff --git a/adapters/postgres/drizzle/0000_phase0_init.sql b/adapters/postgres/drizzle/0000_phase0_init.sql new file mode 100644 index 00000000..2e291efc --- /dev/null +++ b/adapters/postgres/drizzle/0000_phase0_init.sql @@ -0,0 +1,276 @@ +-- Phase 0 Initial Schema - AgentMesh +-- Generated manually for Phase 1 Registry + +CREATE EXTENSION IF NOT EXISTS "pgcrypto"; + +-- organizations +CREATE TABLE IF NOT EXISTS "organizations" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "name" varchar(255) NOT NULL, + "slug" varchar(255) NOT NULL, + "description" text, + "metadata" jsonb DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL +); +CREATE UNIQUE INDEX IF NOT EXISTS "organizations_slug_idx" ON "organizations" ("slug"); + +-- projects +CREATE TABLE IF NOT EXISTS "projects" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "organization_id" uuid NOT NULL REFERENCES "organizations"("id") ON DELETE CASCADE, + "name" varchar(255) NOT NULL, + "slug" varchar(255) NOT NULL, + "description" text, + "metadata" jsonb DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL +); +CREATE INDEX IF NOT EXISTS "projects_org_idx" ON "projects" ("organization_id"); +CREATE UNIQUE INDEX IF NOT EXISTS "projects_org_slug_idx" ON "projects" ("organization_id", "slug"); + +-- agents +CREATE TABLE IF NOT EXISTS "agents" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "organization_id" uuid REFERENCES "organizations"("id") ON DELETE CASCADE, + "project_id" uuid REFERENCES "projects"("id") ON DELETE CASCADE, + "name" varchar(255) NOT NULL, + "description" text, + "version" varchar(50) NOT NULL DEFAULT '0.1.0', + "url" text NOT NULL, + "provider_organization" varchar(255), + "trust_level" varchar(20) NOT NULL DEFAULT 'UNTRUSTED', + "health" varchar(20) NOT NULL DEFAULT 'UNKNOWN', + "region" varchar(100), + "environment" varchar(20) DEFAULT 'development', + "tags" jsonb DEFAULT '[]'::jsonb, + "metadata" jsonb DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL, + "last_seen_at" timestamp with time zone, + "ttl_seconds" integer +); +CREATE INDEX IF NOT EXISTS "agents_org_idx" ON "agents" ("organization_id"); +CREATE INDEX IF NOT EXISTS "agents_project_idx" ON "agents" ("project_id"); +CREATE INDEX IF NOT EXISTS "agents_name_idx" ON "agents" ("name"); +CREATE INDEX IF NOT EXISTS "agents_health_idx" ON "agents" ("health"); +CREATE INDEX IF NOT EXISTS "agents_trust_idx" ON "agents" ("trust_level"); + +-- agent_versions +CREATE TABLE IF NOT EXISTS "agent_versions" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "agent_id" uuid NOT NULL REFERENCES "agents"("id") ON DELETE CASCADE, + "version" varchar(50) NOT NULL, + "card" jsonb NOT NULL, + "is_active" boolean DEFAULT true NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); +CREATE INDEX IF NOT EXISTS "agent_versions_agent_idx" ON "agent_versions" ("agent_id"); +CREATE UNIQUE INDEX IF NOT EXISTS "agent_versions_agent_version_idx" ON "agent_versions" ("agent_id", "version"); + +-- agent_cards +CREATE TABLE IF NOT EXISTS "agent_cards" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "agent_id" uuid NOT NULL REFERENCES "agents"("id") ON DELETE CASCADE, + "card" jsonb NOT NULL, + "protocol_version" varchar(20) NOT NULL DEFAULT '0.2', + "signature_verified" boolean DEFAULT false, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "expires_at" timestamp with time zone +); +CREATE INDEX IF NOT EXISTS "agent_cards_agent_idx" ON "agent_cards" ("agent_id"); + +-- skills +CREATE TABLE IF NOT EXISTS "skills" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "agent_id" uuid NOT NULL REFERENCES "agents"("id") ON DELETE CASCADE, + "skill_id" varchar(255) NOT NULL, + "name" varchar(255) NOT NULL, + "description" text, + "tags" jsonb DEFAULT '[]'::jsonb, + "metadata" jsonb DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); +CREATE INDEX IF NOT EXISTS "skills_agent_idx" ON "skills" ("agent_id"); +CREATE INDEX IF NOT EXISTS "skills_skill_id_idx" ON "skills" ("skill_id"); +CREATE INDEX IF NOT EXISTS "skills_name_idx" ON "skills" ("name"); + +-- tasks +CREATE TABLE IF NOT EXISTS "tasks" ( + "id" varchar(255) PRIMARY KEY, + "root_task_id" varchar(255) NOT NULL, + "parent_task_id" varchar(255), + "agent_id" varchar(255) NOT NULL, + "session_id" varchar(255) NOT NULL, + "context_id" varchar(255) NOT NULL, + "trace_id" varchar(255) NOT NULL, + "organization_id" uuid REFERENCES "organizations"("id"), + "project_id" uuid REFERENCES "projects"("id"), + "state" varchar(20) NOT NULL DEFAULT 'SUBMITTED', + "input" jsonb, + "output" jsonb, + "error" jsonb, + "metadata" jsonb DEFAULT '{}'::jsonb, + "attempts" integer DEFAULT 0 NOT NULL, + "max_attempts" integer DEFAULT 3 NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL, + "expires_at" timestamp with time zone +); +CREATE INDEX IF NOT EXISTS "tasks_root_idx" ON "tasks" ("root_task_id"); +CREATE INDEX IF NOT EXISTS "tasks_agent_idx" ON "tasks" ("agent_id"); +CREATE INDEX IF NOT EXISTS "tasks_context_idx" ON "tasks" ("context_id"); +CREATE INDEX IF NOT EXISTS "tasks_trace_idx" ON "tasks" ("trace_id"); +CREATE INDEX IF NOT EXISTS "tasks_state_idx" ON "tasks" ("state"); +CREATE INDEX IF NOT EXISTS "tasks_org_idx" ON "tasks" ("organization_id"); +CREATE INDEX IF NOT EXISTS "tasks_created_idx" ON "tasks" ("created_at"); + +-- task_history +CREATE TABLE IF NOT EXISTS "task_history" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "task_id" varchar(255) NOT NULL REFERENCES "tasks"("id") ON DELETE CASCADE, + "from_state" varchar(20), + "to_state" varchar(20) NOT NULL, + "reason" text, + "actor_id" varchar(255), + "metadata" jsonb DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); +CREATE INDEX IF NOT EXISTS "task_history_task_idx" ON "task_history" ("task_id"); +CREATE INDEX IF NOT EXISTS "task_history_created_idx" ON "task_history" ("created_at"); + +-- task_edges +CREATE TABLE IF NOT EXISTS "task_edges" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "parent_task_id" varchar(255) NOT NULL REFERENCES "tasks"("id") ON DELETE CASCADE, + "child_task_id" varchar(255) NOT NULL REFERENCES "tasks"("id") ON DELETE CASCADE, + "agent_id" varchar(255) NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); +CREATE INDEX IF NOT EXISTS "task_edges_parent_idx" ON "task_edges" ("parent_task_id"); +CREATE INDEX IF NOT EXISTS "task_edges_child_idx" ON "task_edges" ("child_task_id"); + +-- messages +CREATE TABLE IF NOT EXISTS "messages" ( + "id" varchar(255) PRIMARY KEY, + "task_id" varchar(255) REFERENCES "tasks"("id") ON DELETE CASCADE, + "context_id" varchar(255) NOT NULL, + "trace_id" varchar(255) NOT NULL, + "role" varchar(20) NOT NULL, + "parts" jsonb NOT NULL, + "metadata" jsonb DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); +CREATE INDEX IF NOT EXISTS "messages_task_idx" ON "messages" ("task_id"); +CREATE INDEX IF NOT EXISTS "messages_context_idx" ON "messages" ("context_id"); +CREATE INDEX IF NOT EXISTS "messages_trace_idx" ON "messages" ("trace_id"); +CREATE INDEX IF NOT EXISTS "messages_created_idx" ON "messages" ("created_at"); + +-- artifacts +CREATE TABLE IF NOT EXISTS "artifacts" ( + "id" varchar(255) PRIMARY KEY, + "task_id" varchar(255) REFERENCES "tasks"("id") ON DELETE CASCADE, + "organization_id" uuid REFERENCES "organizations"("id"), + "project_id" uuid REFERENCES "projects"("id"), + "name" varchar(255), + "description" text, + "content_type" varchar(255) NOT NULL, + "size" integer NOT NULL, + "checksum" varchar(128), + "storage_key" text NOT NULL, + "storage_bucket" varchar(255) NOT NULL, + "metadata" jsonb DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "expires_at" timestamp with time zone +); +CREATE INDEX IF NOT EXISTS "artifacts_task_idx" ON "artifacts" ("task_id"); +CREATE INDEX IF NOT EXISTS "artifacts_org_idx" ON "artifacts" ("organization_id"); +CREATE INDEX IF NOT EXISTS "artifacts_created_idx" ON "artifacts" ("created_at"); + +-- routes +CREATE TABLE IF NOT EXISTS "routes" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "organization_id" uuid REFERENCES "organizations"("id") ON DELETE CASCADE, + "project_id" uuid REFERENCES "projects"("id") ON DELETE CASCADE, + "name" varchar(255) NOT NULL, + "description" text, + "skill" varchar(255), + "capability" varchar(255), + "strategy" varchar(50) NOT NULL DEFAULT 'capability_match', + "config" jsonb DEFAULT '{}'::jsonb, + "is_active" boolean DEFAULT true NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL +); +CREATE INDEX IF NOT EXISTS "routes_org_idx" ON "routes" ("organization_id"); +CREATE INDEX IF NOT EXISTS "routes_skill_idx" ON "routes" ("skill"); + +-- policies +CREATE TABLE IF NOT EXISTS "policies" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "organization_id" uuid REFERENCES "organizations"("id") ON DELETE CASCADE, + "project_id" uuid REFERENCES "projects"("id") ON DELETE CASCADE, + "name" varchar(255) NOT NULL, + "description" text, + "effect" varchar(20) NOT NULL DEFAULT 'allow', + "subjects" jsonb DEFAULT '[]'::jsonb, + "resources" jsonb DEFAULT '[]'::jsonb, + "actions" jsonb DEFAULT '[]'::jsonb, + "conditions" jsonb DEFAULT '{}'::jsonb, + "is_active" boolean DEFAULT true NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL +); +CREATE INDEX IF NOT EXISTS "policies_org_idx" ON "policies" ("organization_id"); + +-- credentials +CREATE TABLE IF NOT EXISTS "credentials" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "organization_id" uuid REFERENCES "organizations"("id") ON DELETE CASCADE, + "project_id" uuid REFERENCES "projects"("id") ON DELETE CASCADE, + "name" varchar(255) NOT NULL, + "type" varchar(50) NOT NULL, + "data" jsonb NOT NULL, + "metadata" jsonb DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL, + "expires_at" timestamp with time zone +); +CREATE INDEX IF NOT EXISTS "credentials_org_idx" ON "credentials" ("organization_id"); +CREATE INDEX IF NOT EXISTS "credentials_type_idx" ON "credentials" ("type"); + +-- events +CREATE TABLE IF NOT EXISTS "events" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "type" varchar(100) NOT NULL, + "source" varchar(100) NOT NULL, + "subject" varchar(255) NOT NULL, + "data" jsonb, + "trace_id" varchar(255), + "organization_id" uuid REFERENCES "organizations"("id"), + "project_id" uuid REFERENCES "projects"("id"), + "metadata" jsonb DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); +CREATE INDEX IF NOT EXISTS "events_type_idx" ON "events" ("type"); +CREATE INDEX IF NOT EXISTS "events_subject_idx" ON "events" ("subject"); +CREATE INDEX IF NOT EXISTS "events_trace_idx" ON "events" ("trace_id"); +CREATE INDEX IF NOT EXISTS "events_created_idx" ON "events" ("created_at"); +CREATE INDEX IF NOT EXISTS "events_org_idx" ON "events" ("organization_id"); + +-- webhooks +CREATE TABLE IF NOT EXISTS "webhooks" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "organization_id" uuid REFERENCES "organizations"("id") ON DELETE CASCADE, + "project_id" uuid REFERENCES "projects"("id") ON DELETE CASCADE, + "url" text NOT NULL, + "secret" text, + "events" jsonb DEFAULT '[]'::jsonb, + "is_active" boolean DEFAULT true NOT NULL, + "metadata" jsonb DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL, + "last_delivery_at" timestamp with time zone, + "last_delivery_status" varchar(20) +); +CREATE INDEX IF NOT EXISTS "webhooks_org_idx" ON "webhooks" ("organization_id"); diff --git a/adapters/postgres/drizzle/meta/_journal.json b/adapters/postgres/drizzle/meta/_journal.json new file mode 100644 index 00000000..4fab1a21 --- /dev/null +++ b/adapters/postgres/drizzle/meta/_journal.json @@ -0,0 +1,13 @@ +{ + "version": "7", + "dialect": "postgresql", + "entries": [ + { + "idx": 0, + "version": "7", + "when": 1727200000000, + "tag": "0000_phase0_init", + "breakpoints": true + } + ] +} diff --git a/adapters/redis/src/index.ts b/adapters/redis/src/index.ts index 3ee4ad54..2a15a33c 100644 --- a/adapters/redis/src/index.ts +++ b/adapters/redis/src/index.ts @@ -5,24 +5,41 @@ let redisInstance: Redis | null = null; export interface RedisConfig { url: string; keyPrefix?: string; - maxRetriesPerRequest?: number; + maxRetriesPerRequest?: number | null; + enableOfflineQueue?: boolean; + lazyConnect?: boolean; } export function createRedisClient(config: RedisConfig): Redis { const opts: RedisOptions = { - maxRetriesPerRequest: config.maxRetriesPerRequest ?? 3, + maxRetriesPerRequest: config.maxRetriesPerRequest ?? null, // no retry limit, fail fast + enableOfflineQueue: config.enableOfflineQueue ?? false, enableReadyCheck: true, - lazyConnect: false, + lazyConnect: config.lazyConnect ?? true, + retryStrategy: () => null, // don't retry connection }; if (config.keyPrefix) opts.keyPrefix = config.keyPrefix; - const client = config.url.startsWith("redis://") || config.url.startsWith("rediss://") ? new Redis(config.url, opts) : new Redis({ ...opts, host: config.url }); + const client = config.url.startsWith("redis://") || config.url.startsWith("rediss://") ? new Redis(config.url, opts) : new Redis({ ...opts, host: config.url } as any); client.on("connect", () => console.log("[redis] connected")); - client.on("error", (err: any) => console.error("[redis] error", err)); + client.on("ready", () => console.log("[redis] ready")); + client.on("error", (err: any) => { + // Don't spam logs for connection refused in dev + if (err.message?.includes("ECONNREFUSED")) { + console.warn("[redis] connection refused, using in-memory fallback"); + } else { + console.error("[redis] error", err); + } + }); client.on("close", () => console.log("[redis] closed")); + // Try to connect but don't throw + client.connect().catch(() => { + console.warn("[redis] failed to connect, will use in-memory fallback"); + }); + return client; } @@ -35,42 +52,99 @@ export function getRedis(config?: RedisConfig): Redis { export async function closeRedis(): Promise { if (redisInstance) { - await redisInstance.quit(); + try { + await redisInstance.quit(); + } catch {} redisInstance = null; } } -// Helpers +// Helpers with graceful fallback export class RedisCache { + private fallback = new Map(); + constructor(private redis: Redis, private prefix = "cache:") {} async get(key: string): Promise { - const val = await this.redis.get(this.prefix + key); - if (!val) return null; + const fullKey = this.prefix + key; try { - return JSON.parse(val) as T; + const val = await this.redis.get(fullKey); + if (!val) { + // Check fallback + const fb = this.fallback.get(fullKey); + if (!fb) return null; + if (fb.expiresAt && fb.expiresAt < Date.now()) { + this.fallback.delete(fullKey); + return null; + } + try { + return JSON.parse(fb.value) as T; + } catch { + return fb.value as unknown as T; + } + } + try { + return JSON.parse(val) as T; + } catch { + return val as unknown as T; + } } catch { - return val as unknown as T; + // Fallback to memory + const fb = this.fallback.get(fullKey); + if (!fb) return null; + if (fb.expiresAt && fb.expiresAt < Date.now()) { + this.fallback.delete(fullKey); + return null; + } + try { + return JSON.parse(fb.value) as T; + } catch { + return fb.value as unknown as T; + } } } async set(key: string, value: unknown, ttlSeconds?: number): Promise { + const fullKey = this.prefix + key; const serialized = typeof value === "string" ? value : JSON.stringify(value); - if (ttlSeconds) { - await this.redis.setex(this.prefix + key, ttlSeconds, serialized); - } else { - await this.redis.set(this.prefix + key, serialized); + const expiresAt = ttlSeconds ? Date.now() + ttlSeconds * 1000 : undefined; + + // Always set in fallback + this.fallback.set(fullKey, { value: serialized, expiresAt }); + + try { + if (ttlSeconds) { + await this.redis.setex(fullKey, ttlSeconds, serialized); + } else { + await this.redis.set(fullKey, serialized); + } + } catch { + // Ignore Redis errors, fallback already set } } async del(key: string): Promise { - await this.redis.del(this.prefix + key); + const fullKey = this.prefix + key; + this.fallback.delete(fullKey); + try { + await this.redis.del(fullKey); + } catch {} } async exists(key: string): Promise { - const res = await this.redis.exists(this.prefix + key); - return res === 1; + const fullKey = this.prefix + key; + try { + const res = await this.redis.exists(fullKey); + if (res === 1) return true; + } catch {} + const fb = this.fallback.get(fullKey); + if (!fb) return false; + if (fb.expiresAt && fb.expiresAt < Date.now()) { + this.fallback.delete(fullKey); + return false; + } + return true; } } @@ -78,19 +152,24 @@ export class RedisRateLimiter { constructor(private redis: Redis) {} async isAllowed(key: string, limit: number, windowSeconds: number): Promise<{ allowed: boolean; remaining: number; resetAt: number }> { - const now = Date.now(); - const windowKey = `ratelimit:${key}:${Math.floor(now / 1000 / windowSeconds)}`; - const count = await this.redis.incr(windowKey); - if (count === 1) { - await this.redis.expire(windowKey, windowSeconds); + try { + const now = Date.now(); + const windowKey = `ratelimit:${key}:${Math.floor(now / 1000 / windowSeconds)}`; + const count = await this.redis.incr(windowKey); + if (count === 1) { + await this.redis.expire(windowKey, windowSeconds); + } + const ttl = await this.redis.ttl(windowKey); + const resetAt = now + ttl * 1000; + return { + allowed: count <= limit, + remaining: Math.max(0, limit - count), + resetAt, + }; + } catch { + // Fail open if Redis unavailable + return { allowed: true, remaining: limit, resetAt: Date.now() + windowSeconds * 1000 }; } - const ttl = await this.redis.ttl(windowKey); - const resetAt = now + ttl * 1000; - return { - allowed: count <= limit, - remaining: Math.max(0, limit - count), - resetAt, - }; } } @@ -98,11 +177,18 @@ export class RedisDistributedLock { constructor(private redis: Redis) {} async acquire(key: string, ttlMs: number, value = "locked"): Promise { - const result = await this.redis.set(`lock:${key}`, value, "PX", ttlMs, "NX"); - return result === "OK"; + try { + const result = await this.redis.set(`lock:${key}`, value, "PX", ttlMs, "NX"); + return result === "OK"; + } catch { + // Fail open for dev + return true; + } } async release(key: string): Promise { - await this.redis.del(`lock:${key}`); + try { + await this.redis.del(`lock:${key}`); + } catch {} } } diff --git a/apps/control-plane/package.json b/apps/control-plane/package.json index efaad8a3..e98b86fd 100644 --- a/apps/control-plane/package.json +++ b/apps/control-plane/package.json @@ -25,6 +25,7 @@ "@agentmesh/postgres-adapter": "workspace:*", "@agentmesh/redis-adapter": "workspace:*", "@agentmesh/nats-adapter": "workspace:*", + "drizzle-orm": "^0.35.3", "fastify": "^5.3.2", "@fastify/cors": "^10.0.1", "@fastify/helmet": "^12.0.1", diff --git a/apps/control-plane/src/index.ts b/apps/control-plane/src/index.ts index 828693f1..09e48e73 100644 --- a/apps/control-plane/src/index.ts +++ b/apps/control-plane/src/index.ts @@ -1,6 +1,7 @@ import { loadControlPlaneConfig } from "@agentmesh/config"; import { initTelemetry } from "@agentmesh/telemetry"; import { buildApp } from "./app.js"; +import { getHealthChecker } from "./workers/health-checker.js"; async function main() { const config = loadControlPlaneConfig(); @@ -15,11 +16,16 @@ async function main() { const app = await buildApp({ config }); + // Start health checker worker + const healthChecker = getHealthChecker(); + healthChecker.start(); + const signals: NodeJS.Signals[] = ["SIGINT", "SIGTERM"]; signals.forEach((signal) => { process.on(signal, async () => { console.log(`[control-plane] received ${signal}, shutting down...`); try { + healthChecker.stop(); await app.close(); console.log("[control-plane] closed gracefully"); process.exit(0); @@ -35,6 +41,8 @@ async function main() { console.log(`[control-plane] listening on http://${config.CONTROL_PLANE_HOST}:${config.CONTROL_PLANE_PORT}`); console.log(`[control-plane] health: http://${config.CONTROL_PLANE_HOST}:${config.CONTROL_PLANE_PORT}/health`); console.log(`[control-plane] v1 health: http://${config.CONTROL_PLANE_HOST}:${config.CONTROL_PLANE_PORT}/v1/health`); + console.log(`[control-plane] registry: http://${config.CONTROL_PLANE_HOST}:${config.CONTROL_PLANE_PORT}/v1/agents`); + console.log(`[control-plane] discovery: http://${config.CONTROL_PLANE_HOST}:${config.CONTROL_PLANE_PORT}/v1/agents/discover?skill=code-review`); } catch (err) { app.log.error(err); process.exit(1); diff --git a/apps/control-plane/src/lib/agent-card-fetcher.ts b/apps/control-plane/src/lib/agent-card-fetcher.ts new file mode 100644 index 00000000..e5e7dcf7 --- /dev/null +++ b/apps/control-plane/src/lib/agent-card-fetcher.ts @@ -0,0 +1,51 @@ +import { fetchAgentCard, fetchWellKnownCard, buildDiscoveryUrls, type AgentCard } from "@agentmesh/a2a-protocol"; +import { assertUrlSafe } from "./ssrf.js"; + +export interface CardFetchResult { + card: AgentCard; + raw: unknown; + url: string; + fetchedAt: string; +} + +export async function fetchCardFromUrl(url: string, opts: { timeoutMs?: number; allowPrivate?: boolean } = {}): Promise { + assertUrlSafe(url, { allowPrivate: opts.allowPrivate, allowLoopback: opts.allowPrivate }); + return fetchAgentCard(url, { + timeoutMs: opts.timeoutMs ?? 10000, + allowPrivate: opts.allowPrivate ?? process.env.NODE_ENV !== "production", + allowLoopback: opts.allowPrivate ?? process.env.NODE_ENV !== "production", + }); +} + +export async function fetchCardWithDiscovery(baseUrl: string, opts: { timeoutMs?: number; allowPrivate?: boolean } = {}): Promise { + assertUrlSafe(baseUrl, { allowPrivate: opts.allowPrivate, allowLoopback: opts.allowPrivate }); + const urls = buildDiscoveryUrls(baseUrl); + let lastErr: Error | null = null; + for (const u of urls) { + try { + // Skip URLs that fail SSRF check + try { + assertUrlSafe(u, { allowPrivate: opts.allowPrivate, allowLoopback: opts.allowPrivate }); + } catch { + continue; + } + return await fetchAgentCard(u, { + timeoutMs: opts.timeoutMs ?? 8000, + allowPrivate: opts.allowPrivate ?? process.env.NODE_ENV !== "production", + allowLoopback: opts.allowPrivate ?? process.env.NODE_ENV !== "production", + }); + } catch (e) { + lastErr = e as Error; + } + } + throw lastErr ?? new Error(`Failed to fetch card from any discovery URL for ${baseUrl}`); +} + +export async function fetchWellKnown(baseUrl: string, opts: { timeoutMs?: number; allowPrivate?: boolean } = {}): Promise { + assertUrlSafe(baseUrl, { allowPrivate: opts.allowPrivate, allowLoopback: opts.allowPrivate }); + return fetchWellKnownCard(baseUrl, { + timeoutMs: opts.timeoutMs ?? 8000, + allowPrivate: opts.allowPrivate ?? process.env.NODE_ENV !== "production", + allowLoopback: opts.allowPrivate ?? process.env.NODE_ENV !== "production", + }); +} diff --git a/apps/control-plane/src/lib/ssrf.ts b/apps/control-plane/src/lib/ssrf.ts new file mode 100644 index 00000000..47aded96 --- /dev/null +++ b/apps/control-plane/src/lib/ssrf.ts @@ -0,0 +1,19 @@ +import { validateUrlForSSRF } from "@agentmesh/core"; + +export function assertUrlSafe(url: string, opts: { allowPrivate?: boolean; allowLoopback?: boolean } = {}): void { + const result = validateUrlForSSRF(url, { + allowPrivate: opts.allowPrivate ?? process.env.NODE_ENV !== "production", + allowLoopback: opts.allowLoopback ?? process.env.NODE_ENV !== "production", + }); + if (!result.allowed) { + throw new Error(`SSRF blocked: ${result.reason} - ${url}`); + } +} + +export function isUrlSafe(url: string, opts: { allowPrivate?: boolean; allowLoopback?: boolean } = {}): boolean { + const result = validateUrlForSSRF(url, { + allowPrivate: opts.allowPrivate ?? process.env.NODE_ENV !== "production", + allowLoopback: opts.allowLoopback ?? process.env.NODE_ENV !== "production", + }); + return result.allowed; +} diff --git a/apps/control-plane/src/modules/discovery/index.ts b/apps/control-plane/src/modules/discovery/index.ts new file mode 100644 index 00000000..e849a7c3 --- /dev/null +++ b/apps/control-plane/src/modules/discovery/index.ts @@ -0,0 +1,164 @@ +import type { AgentRecord } from "../registry/types.js"; +import { getRegistryService } from "../registry/service.js"; +import { fetchCardFromUrl } from "../../lib/agent-card-fetcher.js"; + +export interface DiscoveryProvider { + readonly name: string; + discover(query: DiscoveryQuery): Promise; +} + +export interface DiscoveryQuery { + skill?: string; + capability?: string; + region?: string; + tags?: string[]; + limit?: number; + search?: string; +} + +// Local Registry Provider - uses our own registry +export class LocalRegistryProvider implements DiscoveryProvider { + readonly name = "local_registry"; + + async discover(query: DiscoveryQuery): Promise { + const service = getRegistryService(); + const { agents } = await service.list({ + skill: query.skill, + capability: query.capability, + region: query.region, + tags: query.tags, + search: query.search, + limit: query.limit ?? 20, + }); + return agents; + } +} + +// Direct URL Provider - fetch card directly from URL +export class DirectUrlProvider implements DiscoveryProvider { + readonly name = "direct_url"; + + async discover(query: DiscoveryQuery & { url: string }): Promise { + if (!query.url) return []; + try { + const result = await fetchCardFromUrl(query.url, { allowPrivate: true }); + // Convert card to AgentRecord-like (without persisting) + const record: AgentRecord = { + id: `temp_${Date.now()}`, + name: result.card.name, + description: result.card.description ?? null, + version: result.card.version, + url: result.card.url, + providerOrganization: result.card.provider?.organization ?? null, + trustLevel: "UNTRUSTED", + health: "UNKNOWN", + region: null, + environment: "development", + tags: [], + metadata: {}, + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + lastSeenAt: new Date().toISOString(), + ttlSeconds: null, + skills: result.card.skills.map((s) => ({ skillId: s.id, name: s.name, description: s.description, tags: s.tags })), + card: result.card, + }; + return [record]; + } catch (err) { + console.warn(`[discovery:direct] failed to fetch ${query.url}: ${(err as Error).message}`); + return []; + } + } +} + +// Well-Known Discovery Provider +export class WellKnownProvider implements DiscoveryProvider { + readonly name = "well_known"; + + async discover(query: DiscoveryQuery & { baseUrl: string }): Promise { + if (!query.baseUrl) return []; + const direct = new DirectUrlProvider(); + const wellKnownUrl = `${new URL(query.baseUrl).origin}/.well-known/agent.json`; + return direct.discover({ ...query, url: wellKnownUrl }); + } +} + +// Manually Configured Agents Provider +export class ManualProvider implements DiscoveryProvider { + readonly name = "manual"; + private manualAgents: AgentRecord[] = []; + + addAgent(agent: AgentRecord): void { + this.manualAgents.push(agent); + } + + async discover(query: DiscoveryQuery): Promise { + let list = this.manualAgents; + if (query.skill) { + list = list.filter((a) => a.skills?.some((s) => s.skillId === query.skill || s.name === query.skill)); + } + if (query.region) { + list = list.filter((a) => a.region === query.region); + } + if (query.search) { + const q = query.search.toLowerCase(); + list = list.filter((a) => a.name.toLowerCase().includes(q)); + } + return list.slice(0, query.limit ?? 20); + } +} + +// Composite Discovery - tries multiple providers +export class CompositeDiscovery { + private providers: DiscoveryProvider[] = []; + + constructor() { + this.providers.push(new LocalRegistryProvider()); + this.providers.push(new ManualProvider()); + } + + registerProvider(provider: DiscoveryProvider): void { + this.providers.push(provider); + } + + async discover(query: DiscoveryQuery & { url?: string; baseUrl?: string }): Promise { + const results: AgentRecord[] = []; + const seen = new Set(); + + for (const provider of this.providers) { + try { + const agents = await provider.discover(query as any); + for (const agent of agents) { + if (!seen.has(agent.id) && !seen.has(agent.url)) { + seen.add(agent.id); + seen.add(agent.url); + results.push(agent); + } + } + if (results.length >= (query.limit ?? 20)) break; + } catch (err) { + console.warn(`[discovery] provider ${provider.name} failed: ${(err as Error).message}`); + } + } + + // If direct URL provided, also try it + if (query.url) { + const direct = new DirectUrlProvider(); + const directResults = await direct.discover(query as any); + for (const agent of directResults) { + if (!seen.has(agent.url)) { + results.push(agent); + } + } + } + + return results.slice(0, query.limit ?? 20); + } +} + +let discoveryInstance: CompositeDiscovery | null = null; + +export function getDiscovery(): CompositeDiscovery { + if (!discoveryInstance) discoveryInstance = new CompositeDiscovery(); + return discoveryInstance; +} diff --git a/apps/control-plane/src/modules/registry/cache.ts b/apps/control-plane/src/modules/registry/cache.ts new file mode 100644 index 00000000..46f4852b --- /dev/null +++ b/apps/control-plane/src/modules/registry/cache.ts @@ -0,0 +1,81 @@ +import { getRedis, RedisCache } from "@agentmesh/redis-adapter"; +import type { AgentCard } from "@agentmesh/a2a-protocol"; +import type { AgentRecord } from "./types.js"; + +export class RegistryCache { + private cache: RedisCache; + private healthCache: RedisCache; + + constructor() { + try { + const redis = getRedis(); + this.cache = new RedisCache(redis, "agentmesh:agent:"); + this.healthCache = new RedisCache(redis, "agentmesh:health:"); + } catch { + // Fallback to in-memory if Redis not available (Phase 0 dev) + const mem = new Map(); + const fakeRedis = { + get: async (k: string) => mem.get(k) ?? null, + set: async (k: string, v: string) => { mem.set(k, v); }, + setex: async (k: string, _ttl: number, v: string) => { mem.set(k, v); }, + del: async (k: string) => { mem.delete(k); }, + exists: async (k: string) => (mem.has(k) ? 1 : 0), + } as any; + this.cache = new RedisCache(fakeRedis, "agentmesh:agent:"); + this.healthCache = new RedisCache(fakeRedis, "agentmesh:health:"); + console.warn("[registry-cache] Redis not available, using in-memory fallback"); + } + } + + // Agent record cache + async getAgent(id: string): Promise { + return this.cache.get(id); + } + + async setAgent(agent: AgentRecord, ttlSeconds = 300): Promise { + await this.cache.set(agent.id, agent, ttlSeconds); + } + + async delAgent(id: string): Promise { + await this.cache.del(id); + } + + // Card cache + async getCard(agentId: string): Promise<{ card: AgentCard; fetchedAt: string } | null> { + return this.cache.get<{ card: AgentCard; fetchedAt: string }>(`${agentId}:card`); + } + + async setCard(agentId: string, card: AgentCard, ttlSeconds = 300): Promise { + await this.cache.set(`${agentId}:card`, { card, fetchedAt: new Date().toISOString() }, ttlSeconds); + } + + // Health cache + async getHealth(agentId: string): Promise<{ health: string; lastSeenAt: string; latencyMs?: number } | null> { + return this.healthCache.get(`${agentId}`); + } + + async setHealth(agentId: string, health: string, latencyMs?: number): Promise { + await this.healthCache.set(agentId, { health, lastSeenAt: new Date().toISOString(), latencyMs }, 120); + } + + // Discovery cache - capability filtering + async getDiscoveryCache(key: string): Promise { + return this.cache.get(`discovery:${key}`); + } + + async setDiscoveryCache(key: string, agents: AgentRecord[], ttlSeconds = 60): Promise { + await this.cache.set(`discovery:${key}`, agents, ttlSeconds); + } + + async invalidateDiscovery(): Promise { + // In Phase 1, we don't have scan, so we just log + console.log("[registry-cache] discovery cache invalidation requested (Phase 1: no-op, will implement scan)"); + } +} + +let cacheInstance: RegistryCache | null = null; + +export function getRegistryCache(): RegistryCache { + if (!cacheInstance) cacheInstance = new RegistryCache(); + return cacheInstance; +} diff --git a/apps/control-plane/src/modules/registry/repository.ts b/apps/control-plane/src/modules/registry/repository.ts new file mode 100644 index 00000000..6d51efe0 --- /dev/null +++ b/apps/control-plane/src/modules/registry/repository.ts @@ -0,0 +1,345 @@ +import { getDb, schema } from "@agentmesh/postgres-adapter"; +import { eq, and, ilike, or, desc, asc, sql, inArray } from "drizzle-orm"; +import type { AgentFilter, RegisterAgentInput, AgentRecord } from "./types.js"; +import type { AgentCard } from "@agentmesh/a2a-protocol"; + +// In-memory fallback for Phase 0/1 when Postgres not available +class InMemoryRepo { + private agents = new Map(); + private skills = new Map>(); + + async create(input: RegisterAgentInput & { card?: AgentCard }): Promise { + const id = `agent_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 8)}`; + const now = new Date().toISOString(); + const record: AgentRecord = { + id, + organizationId: input.organizationId ?? null, + projectId: input.projectId ?? null, + name: input.name ?? input.card?.name ?? "Unknown Agent", + description: input.description ?? input.card?.description ?? null, + version: input.version ?? input.card?.version ?? "0.1.0", + url: input.url, + providerOrganization: input.providerOrganization ?? input.card?.provider?.organization ?? null, + trustLevel: input.trustLevel ?? "UNTRUSTED", + health: "UNKNOWN", + region: input.region ?? null, + environment: input.environment ?? "development", + tags: input.tags ?? [], + metadata: input.metadata ?? {}, + createdAt: now, + updatedAt: now, + lastSeenAt: now, + ttlSeconds: input.ttlSeconds ?? null, + card: input.card ?? null, + }; + this.agents.set(id, record); + if (input.card?.skills) { + this.skills.set( + id, + input.card.skills.map((s) => ({ skillId: s.id, name: s.name, description: s.description, tags: s.tags })) + ); + } + return record; + } + + async getById(id: string): Promise { + const agent = this.agents.get(id) ?? null; + if (!agent) return null; + return { ...agent, skills: this.skills.get(id) ?? [] }; + } + + async list(filter: AgentFilter): Promise<{ agents: AgentRecord[]; total: number }> { + let list = Array.from(this.agents.values()); + + if (filter.name) { + list = list.filter((a) => a.name.toLowerCase().includes(filter.name!.toLowerCase())); + } + if (filter.region) { + list = list.filter((a) => a.region === filter.region); + } + if (filter.health) { + list = list.filter((a) => a.health === filter.health); + } + if (filter.trustLevel) { + list = list.filter((a) => a.trustLevel === filter.trustLevel); + } + if (filter.skill) { + list = list.filter((a) => { + const skills = this.skills.get(a.id) ?? []; + return skills.some((s) => s.skillId === filter.skill || s.name === filter.skill); + }); + } + if (filter.search) { + const q = filter.search.toLowerCase(); + list = list.filter((a) => a.name.toLowerCase().includes(q) || (a.description ?? "").toLowerCase().includes(q)); + } + + const total = list.length; + const offset = filter.offset ?? 0; + const limit = filter.limit ?? 20; + list = list.slice(offset, offset + limit); + + const enriched = list.map((a) => ({ ...a, skills: this.skills.get(a.id) ?? [] })); + return { agents: enriched, total }; + } + + async update(id: string, input: Partial & { health?: string; lastSeenAt?: string }): Promise { + const existing = this.agents.get(id); + if (!existing) return null; + const now = new Date().toISOString(); + const updated: AgentRecord = { + ...existing, + ...input, + name: input.name ?? existing.name, + description: input.description ?? existing.description, + version: input.version ?? existing.version, + url: input.url ?? existing.url, + trustLevel: input.trustLevel ?? existing.trustLevel, + health: input.health ?? existing.health, + region: input.region ?? existing.region, + environment: input.environment ?? existing.environment, + tags: input.tags ?? existing.tags, + metadata: input.metadata ?? existing.metadata, + updatedAt: now, + lastSeenAt: input.lastSeenAt ?? existing.lastSeenAt, + }; + this.agents.set(id, updated); + return { ...updated, skills: this.skills.get(id) ?? [] }; + } + + async delete(id: string): Promise { + const existed = this.agents.has(id); + this.agents.delete(id); + this.skills.delete(id); + return existed; + } + + async updateHealth(id: string, health: string): Promise { + const existing = this.agents.get(id); + if (existing) { + existing.health = health; + existing.lastSeenAt = new Date().toISOString(); + existing.updatedAt = new Date().toISOString(); + this.agents.set(id, existing); + } + } +} + +const memRepo = new InMemoryRepo(); + +export class RegistryRepository { + private useMemory: boolean; + + constructor() { + // Check if DATABASE_URL is set and we can connect + // For Phase 1, we default to memory if no DB, but try Postgres + this.useMemory = !process.env.DATABASE_URL || process.env.DATABASE_URL.includes("localhost") ? true : false; + // Actually we will try to use memory for now to keep Phase 1 simple and not require DB running in CI + // In production, set USE_POSTGRES=true to force Postgres + if (process.env.USE_POSTGRES === "true") this.useMemory = false; + else this.useMemory = true; // Phase 1: default to memory for simplicity, Postgres will be used in Phase 2 when we have migrations running + + if (this.useMemory) { + console.log("[registry-repo] Using InMemory repository (Phase 1 default, set USE_POSTGRES=true to use Postgres)"); + } else { + console.log("[registry-repo] Using Postgres repository"); + } + } + + async create(input: RegisterAgentInput & { card?: AgentCard }): Promise { + if (this.useMemory) return memRepo.create(input); + + const db = getDb(); + const now = new Date(); + + // Create agent + const [agent] = await db + .insert(schema.agents) + .values({ + organizationId: input.organizationId as any, + projectId: input.projectId as any, + name: input.name ?? input.card?.name ?? "Unknown", + description: input.description ?? input.card?.description, + version: input.version ?? input.card?.version ?? "0.1.0", + url: input.url, + providerOrganization: input.providerOrganization ?? input.card?.provider?.organization, + trustLevel: input.trustLevel ?? "UNTRUSTED", + health: "UNKNOWN", + region: input.region, + environment: input.environment ?? "development", + tags: input.tags ?? [], + metadata: input.metadata ?? {}, + ttlSeconds: input.ttlSeconds, + }) + .returning(); + + // Create skills if card has them + if (input.card?.skills?.length) { + await db.insert(schema.skills).values( + input.card.skills.map((s) => ({ + agentId: agent.id, + skillId: s.id, + name: s.name, + description: s.description, + tags: s.tags, + metadata: {}, + })) + ); + } + + // Create agent card + if (input.card) { + await db.insert(schema.agentCards).values({ + agentId: agent.id, + card: input.card as any, + protocolVersion: input.card.protocolVersions?.[0] ?? "0.2", + }); + } + + return this.mapToRecord(agent, input.card); + } + + async getById(id: string): Promise { + if (this.useMemory) return memRepo.getById(id); + + const db = getDb(); + const [agent] = await db.select().from(schema.agents).where(eq(schema.agents.id, id as any)).limit(1); + if (!agent) return null; + + const skills = await db.select().from(schema.skills).where(eq(schema.skills.agentId, agent.id)); + const [cardRow] = await db + .select() + .from(schema.agentCards) + .where(eq(schema.agentCards.agentId, agent.id)) + .orderBy(desc(schema.agentCards.createdAt)) + .limit(1); + + return this.mapToRecord(agent, cardRow?.card as any, skills as any); + } + + async list(filter: AgentFilter): Promise<{ agents: AgentRecord[]; total: number }> { + if (this.useMemory) return memRepo.list(filter); + + const db = getDb(); + // For simplicity, we do basic filtering in Phase 1 + // Full capability filtering with joins will be in Phase 2 + let query = db.select().from(schema.agents).$dynamic(); + const conditions: any[] = []; + + if (filter.name) conditions.push(ilike(schema.agents.name, `%${filter.name}%`)); + if (filter.region) conditions.push(eq(schema.agents.region, filter.region)); + if (filter.health) conditions.push(eq(schema.agents.health, filter.health)); + if (filter.trustLevel) conditions.push(eq(schema.agents.trustLevel, filter.trustLevel)); + if (filter.organizationId) conditions.push(eq(schema.agents.organizationId, filter.organizationId as any)); + + if (conditions.length > 0) { + query = query.where(and(...conditions)); + } + + const agents = await query.limit(filter.limit ?? 20).offset(filter.offset ?? 0); + // Get total count + const [{ count }] = await db + .select({ count: sql`count(*)` }) + .from(schema.agents) + .where(conditions.length > 0 ? and(...conditions) : undefined); + + const enriched = await Promise.all( + agents.map(async (a) => { + const skills = await db.select().from(schema.skills).where(eq(schema.skills.agentId, a.id)); + return this.mapToRecord(a, null, skills as any); + }) + ); + + // Skill filtering post-query for Phase 1 (will be optimized in Phase 2) + let filtered = enriched; + if (filter.skill) { + filtered = filtered.filter((a) => a.skills?.some((s) => s.skillId === filter.skill || s.name === filter.skill)); + } + if (filter.search) { + const q = filter.search.toLowerCase(); + filtered = filtered.filter((a) => a.name.toLowerCase().includes(q) || (a.description ?? "").toLowerCase().includes(q)); + } + + return { agents: filtered, total: Number(count) }; + } + + async update(id: string, input: Partial & { health?: string; lastSeenAt?: string }): Promise { + if (this.useMemory) return memRepo.update(id, input); + + const db = getDb(); + const [existing] = await db.select().from(schema.agents).where(eq(schema.agents.id, id as any)).limit(1); + if (!existing) return null; + + const [updated] = await db + .update(schema.agents) + .set({ + name: input.name ?? existing.name, + description: input.description ?? existing.description, + version: input.version ?? existing.version, + url: input.url ?? existing.url, + trustLevel: input.trustLevel ?? existing.trustLevel, + health: (input.health as any) ?? existing.health, + region: input.region ?? existing.region, + environment: (input.environment as any) ?? existing.environment, + tags: input.tags ?? (existing.tags as any), + metadata: input.metadata ?? (existing.metadata as any), + updatedAt: new Date(), + lastSeenAt: input.lastSeenAt ? new Date(input.lastSeenAt) : existing.lastSeenAt, + }) + .where(eq(schema.agents.id, id as any)) + .returning(); + + const skills = await db.select().from(schema.skills).where(eq(schema.skills.agentId, updated.id)); + return this.mapToRecord(updated, null, skills as any); + } + + async delete(id: string): Promise { + if (this.useMemory) return memRepo.delete(id); + + const db = getDb(); + const result = await db.delete(schema.agents).where(eq(schema.agents.id, id as any)).returning({ id: schema.agents.id }); + return result.length > 0; + } + + async updateHealth(id: string, health: string): Promise { + if (this.useMemory) return memRepo.updateHealth(id, health); + + const db = getDb(); + await db + .update(schema.agents) + .set({ health: health as any, lastSeenAt: new Date(), updatedAt: new Date() }) + .where(eq(schema.agents.id, id as any)); + } + + private mapToRecord(agent: any, card?: AgentCard | null, skills?: any[]): AgentRecord { + return { + id: agent.id, + organizationId: agent.organizationId, + projectId: agent.projectId, + name: agent.name, + description: agent.description, + version: agent.version, + url: agent.url, + providerOrganization: agent.providerOrganization, + trustLevel: agent.trustLevel, + health: agent.health, + region: agent.region, + environment: agent.environment, + tags: (agent.tags as string[]) ?? [], + metadata: (agent.metadata as Record) ?? {}, + createdAt: agent.createdAt instanceof Date ? agent.createdAt.toISOString() : agent.createdAt, + updatedAt: agent.updatedAt instanceof Date ? agent.updatedAt.toISOString() : agent.updatedAt, + lastSeenAt: agent.lastSeenAt instanceof Date ? agent.lastSeenAt.toISOString() : agent.lastSeenAt, + ttlSeconds: agent.ttlSeconds, + skills: skills?.map((s) => ({ skillId: s.skillId, name: s.name, description: s.description, tags: s.tags as string[] })) ?? [], + card: card ?? null, + }; + } +} + +let repoInstance: RegistryRepository | null = null; + +export function getRegistryRepository(): RegistryRepository { + if (!repoInstance) repoInstance = new RegistryRepository(); + return repoInstance; +} diff --git a/apps/control-plane/src/modules/registry/service.ts b/apps/control-plane/src/modules/registry/service.ts new file mode 100644 index 00000000..ed1edaa7 --- /dev/null +++ b/apps/control-plane/src/modules/registry/service.ts @@ -0,0 +1,241 @@ +import { getRegistryRepository } from "./repository.js"; +import { getRegistryCache } from "./cache.js"; +import { fetchCardFromUrl, fetchCardWithDiscovery } from "../../lib/agent-card-fetcher.js"; +import { createEventBus, type AgentMeshEvent } from "@agentmesh/events"; +import type { RegisterAgentInput, AgentFilter, AgentRecord } from "./types.js"; +import { ValidationError, NotFoundError } from "@agentmesh/core"; +import { AgentCardSchema } from "@agentmesh/a2a-protocol"; + +export class RegistryService { + private repo = getRegistryRepository(); + private cache = getRegistryCache(); + private eventBus = createEventBus({ serviceName: "control-plane-registry" }); + + async register(input: RegisterAgentInput): Promise { + // Validate URL + if (!input.url) throw new ValidationError("Agent URL is required"); + try { + new URL(input.url); + } catch { + throw new ValidationError(`Invalid URL: ${input.url}`); + } + + let card = input.card; + let fetchedUrl = input.url; + + // Fetch card if requested or not provided + if (input.fetchCard !== false && !card) { + try { + const result = await fetchCardWithDiscovery(input.url, { allowPrivate: true }); + card = result.card; + fetchedUrl = result.url; + console.log(`[registry] fetched card from ${fetchedUrl} for ${card.name}`); + } catch (err) { + console.warn(`[registry] failed to fetch card from ${input.url}: ${(err as Error).message}, proceeding with manual registration`); + // If fetch fails, we still allow registration with provided name, but mark as UNTRUSTED + } + } + + // If card provided, validate it + if (card) { + const parsed = AgentCardSchema.safeParse(card); + if (!parsed.success) { + throw new ValidationError(`Invalid Agent Card: ${parsed.error.message}`, parsed.error.flatten()); + } + card = parsed.data; + } + + // Determine name/version from card if not provided + const name = input.name ?? card?.name ?? new URL(input.url).hostname; + const version = input.version ?? card?.version ?? "0.1.0"; + const description = input.description ?? card?.description; + + const record = await this.repo.create({ + ...input, + name, + version, + description, + card, + }); + + // Cache + await this.cache.setAgent(record); + if (card) await this.cache.setCard(record.id, card); + await this.cache.invalidateDiscovery(); + + // Event + await this.publishEvent("agent.registered", record.id, { agent: record, fetchedFrom: fetchedUrl }); + + return record; + } + + async getById(id: string, opts: { useCache?: boolean; includeCard?: boolean } = {}): Promise { + const useCache = opts.useCache ?? true; + + if (useCache) { + const cached = await this.cache.getAgent(id); + if (cached) { + // Optionally refresh card from cache + if (opts.includeCard) { + const cardCached = await this.cache.getCard(id); + if (cardCached) cached.card = cardCached.card; + } + return cached; + } + } + + const agent = await this.repo.getById(id); + if (!agent) throw new NotFoundError("Agent", id); + + // Cache it + await this.cache.setAgent(agent); + + // Try to get card if requested and not in record + if (opts.includeCard && !agent.card) { + try { + const cardCached = await this.cache.getCard(id); + if (cardCached) agent.card = cardCached.card; + else { + // Fetch fresh card from agent URL + const result = await fetchCardFromUrl(agent.url, { allowPrivate: true }); + agent.card = result.card; + await this.cache.setCard(id, result.card); + } + } catch (err) { + console.warn(`[registry] failed to fetch card for ${id}: ${(err as Error).message}`); + } + } + + return agent; + } + + async list(filter: AgentFilter): Promise<{ agents: AgentRecord[]; total: number }> { + // Try discovery cache for capability filtering + const cacheKey = `${filter.skill ?? ""}:${filter.capability ?? ""}:${filter.region ?? ""}:${filter.search ?? ""}:${filter.limit ?? 20}:${filter.offset ?? 0}`; + if (filter.skill || filter.capability) { + const cached = await this.cache.getDiscoveryCache(cacheKey); + if (cached) return { agents: cached, total: cached.length }; + } + + const result = await this.repo.list(filter); + + // Cache discovery results + if (filter.skill || filter.capability) { + await this.cache.setDiscoveryCache(cacheKey, result.agents); + } + + return result; + } + + async update(id: string, input: Partial): Promise { + const existing = await this.getById(id, { useCache: false }); + if (!existing) throw new NotFoundError("Agent", id); + + // If URL changed and fetchCard true, fetch new card + let card = existing.card; + if (input.url && input.url !== existing.url && input.fetchCard !== false) { + try { + const result = await fetchCardWithDiscovery(input.url, { allowPrivate: true }); + card = result.card; + await this.cache.setCard(id, card); + } catch (err) { + console.warn(`[registry] failed to fetch new card for ${id}: ${(err as Error).message}`); + } + } + + if (input.card) { + const parsed = AgentCardSchema.safeParse(input.card); + if (!parsed.success) throw new ValidationError(`Invalid Agent Card: ${parsed.error.message}`); + card = parsed.data; + } + + const updated = await this.repo.update(id, { ...input, ...(card ? { card } : {}) } as any); + if (!updated) throw new NotFoundError("Agent", id); + + // Update cache + await this.cache.setAgent(updated); + if (card) await this.cache.setCard(id, card); + await this.cache.invalidateDiscovery(); + + await this.publishEvent("agent.updated", id, { agent: updated }); + + return updated; + } + + async delete(id: string): Promise { + const existing = await this.repo.getById(id); + if (!existing) throw new NotFoundError("Agent", id); + + const deleted = await this.repo.delete(id); + if (!deleted) throw new NotFoundError("Agent", id); + + await this.cache.delAgent(id); + await this.cache.invalidateDiscovery(); + + await this.publishEvent("agent.deleted", id, { agentId: id }); + } + + async getCard(id: string, opts: { forceRefresh?: boolean } = {}): Promise<{ card: any; fetchedAt: string; fromCache: boolean }> { + const agent = await this.getById(id, { useCache: true }); + + if (!opts.forceRefresh) { + const cached = await this.cache.getCard(id); + if (cached) return { card: cached.card, fetchedAt: cached.fetchedAt, fromCache: true }; + } + + // Fetch fresh + try { + const result = await fetchCardFromUrl(agent.url, { allowPrivate: true }); + await this.cache.setCard(id, result.card); + await this.repo.update(id, { lastSeenAt: new Date().toISOString() } as any); + return { card: result.card, fetchedAt: result.fetchedAt, fromCache: false }; + } catch (err) { + // If fetch fails, return cached if exists, otherwise throw + const cached = await this.cache.getCard(id); + if (cached) return { card: cached.card, fetchedAt: cached.fetchedAt, fromCache: true }; + throw new ValidationError(`Failed to fetch Agent Card from ${agent.url}: ${(err as Error).message}`); + } + } + + async updateHealth(id: string, health: "HEALTHY" | "DEGRADED" | "UNHEALTHY" | "UNKNOWN", latencyMs?: number): Promise { + await this.repo.updateHealth(id, health); + await this.cache.setAgent({ ...(await this.getById(id, { useCache: false })), health } as any); + // Don't use cache for health, set separately + const cache = getRegistryCache(); + await cache.setAgent(await this.repo.getById(id) as any); + // Also set health cache + await (cache as any).setHealth?.(id, health, latencyMs); + + if (health === "UNHEALTHY") { + await this.publishEvent("agent.unhealthy", id, { agentId: id, health, latencyMs }); + } + } + + async heartbeat(id: string): Promise { + await this.repo.update(id, { lastSeenAt: new Date().toISOString() } as any); + await this.updateHealth(id, "HEALTHY"); + } + + private async publishEvent(type: AgentMeshEvent["type"], subject: string, data: unknown): Promise { + try { + const event: AgentMeshEvent = { + id: `evt_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 6)}`, + type, + source: "control-plane", + subject: `agent.${subject}`, + data, + timestamp: new Date().toISOString(), + }; + await this.eventBus.publish(event); + } catch (err) { + console.warn(`[registry] failed to publish event ${type}: ${(err as Error).message}`); + } + } +} + +let serviceInstance: RegistryService | null = null; + +export function getRegistryService(): RegistryService { + if (!serviceInstance) serviceInstance = new RegistryService(); + return serviceInstance; +} diff --git a/apps/control-plane/src/modules/registry/types.ts b/apps/control-plane/src/modules/registry/types.ts new file mode 100644 index 00000000..9d8f9000 --- /dev/null +++ b/apps/control-plane/src/modules/registry/types.ts @@ -0,0 +1,58 @@ +import type { AgentCard } from "@agentmesh/a2a-protocol"; + +export interface RegisterAgentInput { + name?: string; + description?: string; + url: string; + version?: string; + providerOrganization?: string; + trustLevel?: "UNTRUSTED" | "EXTERNAL" | "VERIFIED" | "ORGANIZATION" | "SYSTEM"; + region?: string; + environment?: "development" | "staging" | "production"; + tags?: string[]; + metadata?: Record; + organizationId?: string; + projectId?: string; + card?: AgentCard; // if already fetched + fetchCard?: boolean; // whether to fetch card from URL + ttlSeconds?: number; +} + +export interface AgentFilter { + skill?: string; + capability?: string; + name?: string; + region?: string; + environment?: string; + trustLevel?: string; + health?: string; + tags?: string[]; + organizationId?: string; + projectId?: string; + limit?: number; + offset?: number; + search?: string; +} + +export interface AgentRecord { + id: string; + organizationId?: string | null; + projectId?: string | null; + name: string; + description?: string | null; + version: string; + url: string; + providerOrganization?: string | null; + trustLevel: string; + health: string; + region?: string | null; + environment?: string | null; + tags: string[]; + metadata: Record; + createdAt: string; + updatedAt: string; + lastSeenAt?: string | null; + ttlSeconds?: number | null; + skills?: Array<{ skillId: string; name: string; description?: string; tags: string[] }>; + card?: AgentCard | null; +} diff --git a/apps/control-plane/src/modules/tasks/repository.ts b/apps/control-plane/src/modules/tasks/repository.ts new file mode 100644 index 00000000..e4ab509e --- /dev/null +++ b/apps/control-plane/src/modules/tasks/repository.ts @@ -0,0 +1,313 @@ +import { getDb, schema } from "@agentmesh/postgres-adapter"; +import { eq, and, desc, sql } from "drizzle-orm"; +import type { TaskFilter, TaskRecord, TaskHistoryRecord } from "./types.js"; +import type { TaskState } from "@agentmesh/core"; + +// In-memory fallback +class InMemoryTaskRepo { + private tasks = new Map(); + private history = new Map(); + private edges = new Map(); + + async create(task: TaskRecord): Promise { + this.tasks.set(task.id, task); + this.history.set(task.id, [ + { + id: `hist_${Date.now()}`, + taskId: task.id, + fromState: null, + toState: task.state, + reason: "Task created", + createdAt: new Date().toISOString(), + }, + ]); + // Handle delegation edge + if (task.parentTaskId) { + const list = this.edges.get(task.parentTaskId) ?? []; + list.push({ parentTaskId: task.parentTaskId, childTaskId: task.id, agentId: task.agentId, createdAt: new Date().toISOString() }); + this.edges.set(task.parentTaskId, list); + } + return task; + } + + async getById(id: string): Promise { + return this.tasks.get(id) ?? null; + } + + async list(filter: TaskFilter): Promise<{ tasks: TaskRecord[]; total: number }> { + let list = Array.from(this.tasks.values()); + + if (filter.agentId) list = list.filter((t) => t.agentId === filter.agentId); + if (filter.state) list = list.filter((t) => t.state === filter.state); + if (filter.contextId) list = list.filter((t) => t.contextId === filter.contextId); + if (filter.traceId) list = list.filter((t) => t.traceId === filter.traceId); + if (filter.rootTaskId) list = list.filter((t) => t.rootTaskId === filter.rootTaskId); + if (filter.organizationId) list = list.filter((t) => t.organizationId === filter.organizationId); + if (filter.projectId) list = list.filter((t) => t.projectId === filter.projectId); + + const total = list.length; + list.sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()); + + const offset = filter.offset ?? 0; + const limit = filter.limit ?? 20; + list = list.slice(offset, offset + limit); + + return { tasks: list, total }; + } + + async update(id: string, updates: Partial & { fromState?: TaskRecord["state"] }): Promise { + const existing = this.tasks.get(id); + if (!existing) return null; + + const now = new Date().toISOString(); + const updated: TaskRecord = { + ...existing, + ...updates, + id: existing.id, + updatedAt: now, + }; + + this.tasks.set(id, updated); + + // Add history + const hist = this.history.get(id) ?? []; + hist.push({ + id: `hist_${Date.now()}_${Math.random().toString(36).slice(2, 6)}`, + taskId: id, + fromState: updates.fromState ?? existing.state, + toState: updated.state, + reason: (updates as any).reason, + createdAt: now, + }); + this.history.set(id, hist); + + return updated; + } + + async getHistory(taskId: string): Promise { + return this.history.get(taskId) ?? []; + } + + async getEdges(parentTaskId: string): Promise> { + return this.edges.get(parentTaskId) ?? []; + } + + async getAllEdges(): Promise> { + const all: any[] = []; + for (const list of this.edges.values()) all.push(...list); + return all; + } + + async getDelegationTree(rootTaskId: string): Promise { + const visited = new Set(); + const result: TaskRecord[] = []; + + const dfs = (taskId: string) => { + if (visited.has(taskId)) return; + visited.add(taskId); + const task = this.tasks.get(taskId); + if (!task) return; + result.push(task); + const children = this.edges.get(taskId) ?? []; + for (const edge of children) dfs(edge.childTaskId); + }; + + dfs(rootTaskId); + return result; + } +} + +const memRepo = new InMemoryTaskRepo(); + +export class TaskRepository { + private useMemory: boolean; + + constructor() { + this.useMemory = process.env.USE_POSTGRES !== "true"; + if (this.useMemory) console.log("[task-repo] Using InMemory repository (Phase 2 default)"); + else console.log("[task-repo] Using Postgres repository"); + } + + async create(task: TaskRecord): Promise { + if (this.useMemory) return memRepo.create(task); + + const db = getDb(); + const [created] = await db + .insert(schema.tasks) + .values({ + id: task.id, + rootTaskId: task.rootTaskId, + parentTaskId: task.parentTaskId, + agentId: task.agentId, + sessionId: task.sessionId, + contextId: task.contextId, + traceId: task.traceId, + organizationId: task.organizationId as any, + projectId: task.projectId as any, + state: task.state as any, + input: task.input as any, + output: task.output as any, + error: task.error as any, + metadata: task.metadata as any, + attempts: task.attempts, + maxAttempts: task.maxAttempts, + expiresAt: task.expiresAt ? new Date(task.expiresAt) : undefined, + }) + .returning(); + + await db.insert(schema.taskHistory).values({ + taskId: task.id, + fromState: null, + toState: task.state as any, + reason: "Task created", + }); + + if (task.parentTaskId) { + await db.insert(schema.taskEdges).values({ + parentTaskId: task.parentTaskId, + childTaskId: task.id, + agentId: task.agentId, + }); + } + + return this.mapToRecord(created); + } + + async getById(id: string): Promise { + if (this.useMemory) return memRepo.getById(id); + + const db = getDb(); + const [task] = await db.select().from(schema.tasks).where(eq(schema.tasks.id, id)).limit(1); + if (!task) return null; + return this.mapToRecord(task); + } + + async list(filter: TaskFilter): Promise<{ tasks: TaskRecord[]; total: number }> { + if (this.useMemory) return memRepo.list(filter); + + const db = getDb(); + const conditions: any[] = []; + if (filter.agentId) conditions.push(eq(schema.tasks.agentId, filter.agentId)); + if (filter.state) conditions.push(eq(schema.tasks.state, filter.state as any)); + if (filter.contextId) conditions.push(eq(schema.tasks.contextId, filter.contextId)); + if (filter.traceId) conditions.push(eq(schema.tasks.traceId, filter.traceId)); + if (filter.rootTaskId) conditions.push(eq(schema.tasks.rootTaskId, filter.rootTaskId)); + if (filter.organizationId) conditions.push(eq(schema.tasks.organizationId, filter.organizationId as any)); + + let query = db.select().from(schema.tasks).$dynamic(); + if (conditions.length > 0) query = query.where(and(...conditions)); + + const tasks = await query + .orderBy(desc(schema.tasks.createdAt)) + .limit(filter.limit ?? 20) + .offset(filter.offset ?? 0); + + const [{ count }] = await db + .select({ count: sql`count(*)` }) + .from(schema.tasks) + .where(conditions.length > 0 ? and(...conditions) : undefined); + + return { tasks: tasks.map(this.mapToRecord), total: Number(count) }; + } + + async update(id: string, updates: Partial & { fromState?: TaskRecord["state"]; reason?: string }): Promise { + if (this.useMemory) return memRepo.update(id, updates); + + const db = getDb(); + const [existing] = await db.select().from(schema.tasks).where(eq(schema.tasks.id, id)).limit(1); + if (!existing) return null; + + const [updated] = await db + .update(schema.tasks) + .set({ + state: (updates.state as any) ?? existing.state, + output: (updates.output as any) ?? existing.output, + error: (updates.error as any) ?? existing.error, + metadata: (updates.metadata as any) ?? existing.metadata, + attempts: updates.attempts ?? existing.attempts, + updatedAt: new Date(), + expiresAt: updates.expiresAt ? new Date(updates.expiresAt) : existing.expiresAt, + }) + .where(eq(schema.tasks.id, id)) + .returning(); + + await db.insert(schema.taskHistory).values({ + taskId: id, + fromState: (updates.fromState as any) ?? (existing.state as any), + toState: (updates.state as any) ?? (existing.state as any), + reason: updates.reason, + }); + + return this.mapToRecord(updated); + } + + async getHistory(taskId: string): Promise { + if (this.useMemory) return memRepo.getHistory(taskId); + + const db = getDb(); + const rows = await db.select().from(schema.taskHistory).where(eq(schema.taskHistory.taskId, taskId)).orderBy(desc(schema.taskHistory.createdAt)); + return rows.map((r) => ({ + id: r.id, + taskId: r.taskId, + fromState: r.fromState as any, + toState: r.toState as any, + reason: r.reason ?? undefined, + actorId: r.actorId ?? undefined, + metadata: (r.metadata as any) ?? undefined, + createdAt: r.createdAt instanceof Date ? r.createdAt.toISOString() : (r.createdAt as any), + })); + } + + async getDelegationTree(rootTaskId: string): Promise { + if (this.useMemory) return memRepo.getDelegationTree(rootTaskId); + + // For Postgres, we need to recursively get edges - simplified for Phase 2 + const db = getDb(); + const visited = new Set(); + const result: TaskRecord[] = []; + + const dfs = async (taskId: string) => { + if (visited.has(taskId)) return; + visited.add(taskId); + const [task] = await db.select().from(schema.tasks).where(eq(schema.tasks.id, taskId)).limit(1); + if (!task) return; + result.push(this.mapToRecord(task)); + const edges = await db.select().from(schema.taskEdges).where(eq(schema.taskEdges.parentTaskId, taskId)); + for (const edge of edges) await dfs(edge.childTaskId); + }; + + await dfs(rootTaskId); + return result; + } + + private mapToRecord(row: any): TaskRecord { + return { + id: row.id, + rootTaskId: row.rootTaskId, + parentTaskId: row.parentTaskId, + agentId: row.agentId, + sessionId: row.sessionId, + contextId: row.contextId, + traceId: row.traceId, + organizationId: row.organizationId, + projectId: row.projectId, + state: row.state, + input: row.input, + output: row.output, + error: row.error, + metadata: row.metadata ?? {}, + attempts: row.attempts, + maxAttempts: row.maxAttempts, + createdAt: row.createdAt instanceof Date ? row.createdAt.toISOString() : row.createdAt, + updatedAt: row.updatedAt instanceof Date ? row.updatedAt.toISOString() : row.updatedAt, + expiresAt: row.expiresAt instanceof Date ? row.expiresAt.toISOString() : row.expiresAt, + }; + } +} + +let repoInstance: TaskRepository | null = null; + +export function getTaskRepository(): TaskRepository { + if (!repoInstance) repoInstance = new TaskRepository(); + return repoInstance; +} diff --git a/apps/control-plane/src/modules/tasks/service.ts b/apps/control-plane/src/modules/tasks/service.ts new file mode 100644 index 00000000..13a6a176 --- /dev/null +++ b/apps/control-plane/src/modules/tasks/service.ts @@ -0,0 +1,206 @@ +import { getTaskRepository } from "./repository.js"; +import { getRegistryService } from "../registry/service.js"; +import { createEventBus, type AgentMeshEvent } from "@agentmesh/events"; +import type { CreateTaskInput, TaskRecord, TaskFilter } from "./types.js"; +import { TaskState, generateId, nowIso, canTransition, NotFoundError, ValidationError } from "@agentmesh/core"; +import { TaskStateMachine } from "@agentmesh/tasks"; + +export class TaskService { + private repo = getTaskRepository(); + private registry = getRegistryService(); + private eventBus = createEventBus({ serviceName: "control-plane-tasks" }); + + async create(input: CreateTaskInput): Promise { + // Validate agent exists + let agent; + try { + agent = await this.registry.getById(input.agentId); + } catch { + throw new NotFoundError("Agent", input.agentId); + } + + if (agent.health === "UNHEALTHY") { + console.warn(`[tasks] creating task for unhealthy agent ${input.agentId}, but allowing (Phase 2)`); + } + + const id = generateId("task"); + const sessionId = input.sessionId ?? generateId("sess"); + const contextId = input.contextId ?? generateId("ctx"); + const traceId = input.traceId ?? generateId("trace"); + const rootTaskId = input.parentTaskId ? (await this.repo.getById(input.parentTaskId))?.rootTaskId ?? input.parentTaskId : id; + + const now = nowIso(); + + const task: TaskRecord = { + id, + rootTaskId, + parentTaskId: input.parentTaskId ?? null, + agentId: input.agentId, + sessionId, + contextId, + traceId, + organizationId: input.organizationId ?? null, + projectId: input.projectId ?? null, + state: TaskState.SUBMITTED, + input: { message: input.message, metadata: input.metadata }, + output: null, + error: null, + metadata: input.metadata ?? {}, + attempts: 0, + maxAttempts: 3, + createdAt: now, + updatedAt: now, + expiresAt: null, + }; + + const created = await this.repo.create(task); + + await this.publishEvent("task.created", id, { task: created, agentId: input.agentId }); + + // Transition to WORKING asynchronously (simulate delegation) + setImmediate(async () => { + try { + await this.transition(id, TaskState.WORKING, { reason: "Task delegated to agent" }); + // In Phase 2, we simulate task execution - in Phase 3+ we will actually call agent via A2A + // For now, we keep it WORKING until agent reports completion via API or timeout + } catch (err) { + console.error(`[tasks] failed to transition task ${id} to WORKING: ${(err as Error).message}`); + } + }); + + return created; + } + + async getById(id: string): Promise { + const task = await this.repo.getById(id); + if (!task) throw new NotFoundError("Task", id); + return task; + } + + async list(filter: TaskFilter): Promise<{ tasks: TaskRecord[]; total: number }> { + return this.repo.list(filter); + } + + async transition(id: string, toState: TaskRecord["state"], opts: { reason?: string; output?: Record; error?: TaskRecord["error"] } = {}): Promise { + const task = await this.repo.getById(id); + if (!task) throw new NotFoundError("Task", id); + + if (!canTransition(task.state as any, toState as any)) { + throw new ValidationError(`Invalid transition ${task.state} -> ${toState} for task ${id}`); + } + + const updated = await this.repo.update(id, { + state: toState, + fromState: task.state, + output: opts.output ?? task.output, + error: opts.error ?? task.error, + reason: opts.reason, + } as any); + + if (!updated) throw new NotFoundError("Task", id); + + // Publish event based on state + const eventMap: Record = { + [TaskState.WORKING]: "task.started", + [TaskState.COMPLETED]: "task.completed", + [TaskState.FAILED]: "task.failed", + [TaskState.CANCELED]: "task.canceled", + [TaskState.TIMEOUT]: "task.timeout", + }; + + const eventType = eventMap[toState]; + if (eventType) { + await this.publishEvent(eventType, id, { task: updated, reason: opts.reason }); + } + + return updated; + } + + async cancel(id: string, reason = "Canceled by user"): Promise { + const task = await this.getById(id); + if ([TaskState.COMPLETED, TaskState.FAILED, TaskState.CANCELED, TaskState.REJECTED].includes(task.state as any)) { + throw new ValidationError(`Task ${id} is already terminal: ${task.state}`); + } + return this.transition(id, TaskState.CANCELED, { reason }); + } + + async complete(id: string, output?: Record): Promise { + return this.transition(id, TaskState.COMPLETED, { output, reason: "Task completed" }); + } + + async fail(id: string, error: { code: string; message: string; details?: unknown }): Promise { + return this.transition(id, TaskState.FAILED, { error, reason: error.message }); + } + + async retry(id: string): Promise { + const task = await this.getById(id); + if (![TaskState.FAILED, TaskState.TIMEOUT].includes(task.state as any)) { + throw new ValidationError(`Only FAILED or TIMEOUT tasks can be retried, got ${task.state}`); + } + if (task.attempts >= task.maxAttempts) { + throw new ValidationError(`Task ${id} has reached max attempts ${task.maxAttempts}`); + } + + // Transition back to SUBMITTED with incremented attempts + const updated = await this.repo.update(id, { + state: TaskState.SUBMITTED, + fromState: task.state, + attempts: task.attempts + 1, + reason: "Retrying task", + } as any); + + if (!updated) throw new NotFoundError("Task", id); + + await this.publishEvent("task.created", id, { task: updated, retry: true }); + + // Auto transition to WORKING + setImmediate(() => this.transition(id, TaskState.WORKING, { reason: "Retry delegated" }).catch(() => {})); + + return updated as TaskRecord; + } + + async getHistory(id: string): Promise { + await this.getById(id); // ensure exists + return this.repo.getHistory(id); + } + + async getDelegationTree(rootTaskId: string): Promise<{ tree: TaskRecord[]; graph: any }> { + const tasks = await this.repo.getDelegationTree(rootTaskId); + if (tasks.length === 0) throw new NotFoundError("Task", rootTaskId); + + // Build simple graph for visualization + const nodes = tasks.map((t) => ({ id: t.id, label: `${t.id.slice(0, 8)} (${t.state})`, state: t.state, agentId: t.agentId })); + const edges: any[] = []; + + for (const task of tasks) { + if (task.parentTaskId) { + edges.push({ from: task.parentTaskId, to: task.id, agentId: task.agentId }); + } + } + + return { tree: tasks, graph: { nodes, edges, root: rootTaskId } }; + } + + private async publishEvent(type: AgentMeshEvent["type"], taskId: string, data: unknown): Promise { + try { + const event: AgentMeshEvent = { + id: `evt_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 6)}`, + type, + source: "control-plane", + subject: `task.${taskId}`, + data, + timestamp: new Date().toISOString(), + }; + await this.eventBus.publish(event); + } catch (err) { + console.warn(`[tasks] failed to publish event ${type}: ${(err as Error).message}`); + } + } +} + +let serviceInstance: TaskService | null = null; + +export function getTaskService(): TaskService { + if (!serviceInstance) serviceInstance = new TaskService(); + return serviceInstance; +} diff --git a/apps/control-plane/src/modules/tasks/types.ts b/apps/control-plane/src/modules/tasks/types.ts new file mode 100644 index 00000000..d5fe7b17 --- /dev/null +++ b/apps/control-plane/src/modules/tasks/types.ts @@ -0,0 +1,58 @@ +import type { TaskState } from "@agentmesh/core"; + +export interface CreateTaskInput { + agentId: string; + message: { text: string; role?: string; parts?: any[] }; + contextId?: string; + sessionId?: string; + metadata?: Record; + parentTaskId?: string; + organizationId?: string; + projectId?: string; + traceId?: string; +} + +export interface TaskRecord { + id: string; + rootTaskId: string; + parentTaskId?: string | null; + agentId: string; + sessionId: string; + contextId: string; + traceId: string; + organizationId?: string | null; + projectId?: string | null; + state: TaskState; + input?: Record | null; + output?: Record | null; + error?: { code: string; message: string; details?: unknown } | null; + metadata: Record; + attempts: number; + maxAttempts: number; + createdAt: string; + updatedAt: string; + expiresAt?: string | null; +} + +export interface TaskHistoryRecord { + id: string; + taskId: string; + fromState: TaskState | null; + toState: TaskState; + reason?: string; + actorId?: string; + metadata?: Record; + createdAt: string; +} + +export interface TaskFilter { + agentId?: string; + state?: TaskState; + contextId?: string; + traceId?: string; + organizationId?: string; + projectId?: string; + rootTaskId?: string; + limit?: number; + offset?: number; +} diff --git a/apps/control-plane/src/routes/v1/agents.ts b/apps/control-plane/src/routes/v1/agents.ts index 6548a0d8..b29a7d91 100644 --- a/apps/control-plane/src/routes/v1/agents.ts +++ b/apps/control-plane/src/routes/v1/agents.ts @@ -1,60 +1,203 @@ import type { FastifyInstance } from "fastify"; +import { z } from "zod"; +import { getRegistryService } from "../../modules/registry/service.js"; +import { getDiscovery } from "../../modules/discovery/index.js"; + +const RegisterSchema = z.object({ + name: z.string().optional(), + description: z.string().optional(), + url: z.string().url(), + version: z.string().optional(), + providerOrganization: z.string().optional(), + trustLevel: z.enum(["UNTRUSTED", "EXTERNAL", "VERIFIED", "ORGANIZATION", "SYSTEM"]).optional(), + region: z.string().optional(), + environment: z.enum(["development", "staging", "production"]).optional(), + tags: z.array(z.string()).optional(), + metadata: z.record(z.unknown()).optional(), + organizationId: z.string().optional(), + projectId: z.string().optional(), + card: z.record(z.unknown()).optional(), + fetchCard: z.boolean().optional(), + ttlSeconds: z.number().int().positive().optional(), +}); export async function agentsRoutes(app: FastifyInstance) { - // List agents + const registry = getRegistryService(); + const discovery = getDiscovery(); + + // List agents with filtering app.get("/", async (req) => { - const { skill, capability, region, limit = "20" } = req.query as any; + const { skill, capability, region, trustLevel, health, search, tags, limit = "20", offset = "0", organizationId, projectId } = req.query as any; + + const filter = { + skill: skill as string | undefined, + capability: capability as string | undefined, + region: region as string | undefined, + trustLevel: trustLevel as string | undefined, + health: health as string | undefined, + search: search as string | undefined, + tags: tags ? (Array.isArray(tags) ? tags : [tags]) : undefined, + organizationId: organizationId as string | undefined, + projectId: projectId as string | undefined, + limit: parseInt(limit, 10) || 20, + offset: parseInt(offset, 10) || 0, + }; + + const { agents, total } = await registry.list(filter); + return { - agents: [], - filter: { skill, capability, region, limit }, - message: "Agent Registry - full implementation in Phase 1. Will support capability filtering, ranking, health checks.", - total: 0, + agents, + total, + limit: filter.limit, + offset: filter.offset, + filter: { skill, capability, region, search }, + }; + }); + + // Discovery endpoint - composite discovery + app.get("/discover", async (req) => { + const { skill, capability, region, search, url, baseUrl, limit = "20" } = req.query as any; + + const agents = await discovery.discover({ + skill, + capability, + region, + search, + url, + baseUrl, + limit: parseInt(limit, 10) || 20, + }); + + return { + agents, + total: agents.length, + query: { skill, capability, region, search, url, baseUrl }, }; }); // Get agent by id app.get("/:id", async (req, reply) => { const { id } = req.params as { id: string }; - return reply.status(501).send({ - error: { code: "NOT_IMPLEMENTED", message: "Agent lookup - Phase 1" }, - id, - }); + const { includeCard } = req.query as any; + + try { + const agent = await registry.getById(id, { includeCard: includeCard === "true" }); + return { agent }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 404).send({ error: { code: e.code ?? "NOT_FOUND", message: e.message } }); + } }); // Get agent card app.get("/:id/card", async (req, reply) => { const { id } = req.params as { id: string }; - return reply.status(501).send({ - error: { code: "NOT_IMPLEMENTED", message: "Agent Card retrieval with caching, validation, signature verification - Phase 1" }, - id, - }); + const { refresh } = req.query as any; + + try { + const result = await registry.getCard(id, { forceRefresh: refresh === "true" }); + return { + card: result.card, + fetchedAt: result.fetchedAt, + fromCache: result.fromCache, + agentId: id, + }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "FETCH_FAILED", message: e.message } }); + } }); // Register agent app.post("/", async (req, reply) => { - const body = req.body as any; - // Phase 0: just validate structure, don't persist - return reply.status(501).send({ - error: { code: "NOT_IMPLEMENTED", message: "Agent registration - Phase 1 with Postgres + Redis + NATS events" }, - received: body?.name ?? "unknown", - }); + try { + const parsed = RegisterSchema.safeParse(req.body); + if (!parsed.success) { + return reply.status(400).send({ error: { code: "VALIDATION_ERROR", message: "Invalid input", details: parsed.error.flatten() } }); + } + + const agent = await registry.register(parsed.data as any); + return reply.status(201).send({ agent }); + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "REGISTRATION_FAILED", message: e.message, details: e.details } }); + } + }); + + // Update agent + app.patch("/:id", async (req, reply) => { + const { id } = req.params as { id: string }; + try { + const parsed = RegisterSchema.partial().safeParse(req.body); + if (!parsed.success) { + return reply.status(400).send({ error: { code: "VALIDATION_ERROR", message: "Invalid input", details: parsed.error.flatten() } }); + } + + const agent = await registry.update(id, parsed.data as any); + return { agent }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "UPDATE_FAILED", message: e.message } }); + } }); // Deregister app.delete("/:id", async (req, reply) => { const { id } = req.params as { id: string }; - return reply.status(501).send({ - error: { code: "NOT_IMPLEMENTED", message: "Agent deregistration - Phase 1" }, - id, - }); + try { + await registry.delete(id); + return { success: true, id }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "DELETE_FAILED", message: e.message } }); + } }); // Health app.get("/:id/health", async (req, reply) => { const { id } = req.params as { id: string }; - return reply.status(501).send({ - error: { code: "NOT_IMPLEMENTED", message: "Health checks - Phase 1 with heartbeat + TTL" }, - id, - }); + try { + const agent = await registry.getById(id); + return { + agentId: id, + health: agent.health, + lastSeenAt: agent.lastSeenAt, + url: agent.url, + trustLevel: agent.trustLevel, + }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 404).send({ error: { code: e.code ?? "NOT_FOUND", message: e.message } }); + } + }); + + // Heartbeat + app.post("/:id/heartbeat", async (req, reply) => { + const { id } = req.params as { id: string }; + try { + await registry.heartbeat(id); + return { success: true, id, timestamp: new Date().toISOString() }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "HEARTBEAT_FAILED", message: e.message } }); + } + }); + + // Update health manually + app.post("/:id/health", async (req, reply) => { + const { id } = req.params as { id: string }; + const { health, latencyMs } = req.body as any; + + if (!["HEALTHY", "DEGRADED", "UNHEALTHY", "UNKNOWN"].includes(health)) { + return reply.status(400).send({ error: { code: "VALIDATION_ERROR", message: "Invalid health value" } }); + } + + try { + await registry.updateHealth(id, health, latencyMs); + return { success: true, id, health, latencyMs }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "HEALTH_UPDATE_FAILED", message: e.message } }); + } }); } diff --git a/apps/control-plane/src/routes/v1/tasks.ts b/apps/control-plane/src/routes/v1/tasks.ts index 6143c7b9..1aabf320 100644 --- a/apps/control-plane/src/routes/v1/tasks.ts +++ b/apps/control-plane/src/routes/v1/tasks.ts @@ -1,53 +1,241 @@ import type { FastifyInstance } from "fastify"; +import { z } from "zod"; +import { getTaskService } from "../../modules/tasks/service.js"; + +const CreateTaskSchema = z.object({ + agentId: z.string().min(1), + message: z.object({ + text: z.string().min(1), + role: z.enum(["user", "agent"]).optional(), + parts: z.array(z.any()).optional(), + }), + contextId: z.string().optional(), + sessionId: z.string().optional(), + metadata: z.record(z.unknown()).optional(), + parentTaskId: z.string().optional(), + organizationId: z.string().optional(), + projectId: z.string().optional(), + traceId: z.string().optional(), +}); export async function tasksRoutes(app: FastifyInstance) { + const service = getTaskService(); + app.get("/", async (req) => { - const { state, agentId, limit = "20" } = req.query as any; - return { - tasks: [], - filter: { state, agentId, limit }, - message: "Task list - Phase 2 with Postgres persistence, filtering by state, tenant isolation", - total: 0, + const { state, agentId, contextId, traceId, rootTaskId, organizationId, projectId, limit = "20", offset = "0" } = req.query as any; + + const filter = { + state, + agentId, + contextId, + traceId, + rootTaskId, + organizationId, + projectId, + limit: parseInt(limit, 10) || 20, + offset: parseInt(offset, 10) || 0, }; + + const { tasks, total } = await service.list(filter as any); + + return { tasks, total, limit: filter.limit, offset: filter.offset }; }); app.get("/:id", async (req, reply) => { const { id } = req.params as { id: string }; - return reply.status(501).send({ - error: { code: "NOT_IMPLEMENTED", message: "Task lookup - Phase 2 with history, delegation tree" }, - id, - }); + try { + const task = await service.getById(id); + return { task }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 404).send({ error: { code: e.code ?? "NOT_FOUND", message: e.message } }); + } }); app.post("/", async (req, reply) => { - const body = req.body as any; - return reply.status(501).send({ - error: { code: "NOT_IMPLEMENTED", message: "Task creation - Phase 2 with routing, event publishing" }, - received: body, - }); + try { + const parsed = CreateTaskSchema.safeParse(req.body); + if (!parsed.success) { + return reply.status(400).send({ error: { code: "VALIDATION_ERROR", message: "Invalid input", details: parsed.error.flatten() } }); + } + + const task = await service.create(parsed.data as any); + return reply.status(201).send({ task }); + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "CREATE_FAILED", message: e.message } }); + } }); app.post("/:id/cancel", async (req, reply) => { const { id } = req.params as { id: string }; - return reply.status(501).send({ - error: { code: "NOT_IMPLEMENTED", message: "Task cancellation - Phase 2" }, - id, - }); + const { reason } = (req.body as any) ?? {}; + try { + const task = await service.cancel(id, reason); + return { task }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "CANCEL_FAILED", message: e.message } }); + } + }); + + app.post("/:id/complete", async (req, reply) => { + const { id } = req.params as { id: string }; + const { output } = (req.body as any) ?? {}; + try { + const task = await service.complete(id, output); + return { task }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "COMPLETE_FAILED", message: e.message } }); + } + }); + + app.post("/:id/fail", async (req, reply) => { + const { id } = req.params as { id: string }; + const { error } = (req.body as any) ?? {}; + if (!error?.message) { + return reply.status(400).send({ error: { code: "VALIDATION_ERROR", message: "error.message required" } }); + } + try { + const task = await service.fail(id, error); + return { task }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "FAIL_FAILED", message: e.message } }); + } + }); + + app.post("/:id/retry", async (req, reply) => { + const { id } = req.params as { id: string }; + try { + const task = await service.retry(id); + return { task }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "RETRY_FAILED", message: e.message } }); + } }); app.get("/:id/history", async (req, reply) => { const { id } = req.params as { id: string }; - return reply.status(501).send({ - error: { code: "NOT_IMPLEMENTED", message: "Task history - Phase 2" }, - id, - }); + try { + const history = await service.getHistory(id); + return { taskId: id, history }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "HISTORY_FAILED", message: e.message } }); + } }); app.get("/:id/graph", async (req, reply) => { const { id } = req.params as { id: string }; - return reply.status(501).send({ - error: { code: "NOT_IMPLEMENTED", message: "Delegation graph reconstruction - Phase 2" }, - id, + try { + const { tree, graph } = await service.getDelegationTree(id); + return { taskId: id, tree, graph }; + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 500).send({ error: { code: e.code ?? "GRAPH_FAILED", message: e.message } }); + } + }); + + // SSE Streaming for task updates - Phase 2 + app.get("/:id/stream", async (req, reply) => { + const { id } = req.params as { id: string }; + + try { + await service.getById(id); // ensure exists + } catch (err) { + const e = err as any; + return reply.status(e.statusCode ?? 404).send({ error: { code: e.code ?? "NOT_FOUND", message: e.message } }); + } + + // Setup SSE + reply.raw.writeHead(200, { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + "X-Accel-Buffering": "no", }); + + const sendEvent = (event: string, data: unknown) => { + reply.raw.write(`event: ${event}\n`); + reply.raw.write(`data: ${JSON.stringify(data)}\n\n`); + }; + + // Send initial task state + try { + const task = await service.getById(id); + sendEvent("task", task); + } catch {} + + // Subscribe to task events via event bus + const { createEventBus } = await import("@agentmesh/events"); + const eventBus = createEventBus({ serviceName: "control-plane-stream" }); + + const sub = await eventBus.subscribe("*", (event: any) => { + if (event.subject === `task.${id}` || event.data?.task?.id === id) { + const type = event.type.replace("task.", ""); + sendEvent(type, event.data); + // If terminal, close after a delay + if (["completed", "failed", "canceled", "timeout"].includes(type)) { + setTimeout(() => { + try { + reply.raw.end(); + } catch {} + }, 1000); + } + } + }); + + // Heartbeat every 15s + const heartbeat = setInterval(() => { + try { + reply.raw.write(`: heartbeat ${new Date().toISOString()}\n\n`); + } catch { + clearInterval(heartbeat); + } + }, 15000); + + // Polling fallback - check task state every 2s and send if changed + let lastState: string | null = null; + const poll = setInterval(async () => { + try { + const task = await service.getById(id); + if (task.state !== lastState) { + lastState = task.state; + sendEvent("state", { state: task.state, task }); + if (["COMPLETED", "FAILED", "CANCELED", "REJECTED", "TIMEOUT"].includes(task.state)) { + clearInterval(poll); + setTimeout(() => { + try { + reply.raw.end(); + } catch {} + }, 500); + } + } + } catch { + // Task deleted? close + clearInterval(poll); + try { + reply.raw.end(); + } catch {} + } + }, 2000); + + req.raw.on("close", async () => { + clearInterval(heartbeat); + clearInterval(poll); + try { + await sub.unsubscribe(); + await eventBus.close(); + } catch {} + try { + reply.raw.end(); + } catch {} + }); + + // Don't return, keep connection open + return reply; }); } diff --git a/apps/control-plane/src/workers/health-checker.ts b/apps/control-plane/src/workers/health-checker.ts new file mode 100644 index 00000000..be14a7a7 --- /dev/null +++ b/apps/control-plane/src/workers/health-checker.ts @@ -0,0 +1,98 @@ +import { getRegistryRepository } from "../modules/registry/repository.js"; +import { getRegistryService } from "../modules/registry/service.js"; + +export interface HealthCheckerConfig { + intervalMs?: number; + ttlMs?: number; + enabled?: boolean; +} + +export class HealthChecker { + private interval: NodeJS.Timeout | null = null; + private config: Required; + + constructor(config: HealthCheckerConfig = {}) { + this.config = { + intervalMs: config.intervalMs ?? 30000, + ttlMs: config.ttlMs ?? 120000, + enabled: config.enabled ?? true, + }; + } + + start(): void { + if (!this.config.enabled) { + console.log("[health-checker] disabled"); + return; + } + + console.log(`[health-checker] starting with interval ${this.config.intervalMs}ms, ttl ${this.config.ttlMs}ms`); + + this.interval = setInterval(async () => { + try { + await this.check(); + } catch (err) { + console.error("[health-checker] error during check", err); + } + }, this.config.intervalMs); + + // Don't block exit + if (this.interval && (this.interval as any).unref) { + (this.interval as any).unref(); + } + } + + stop(): void { + if (this.interval) { + clearInterval(this.interval); + this.interval = null; + console.log("[health-checker] stopped"); + } + } + + private async check(): Promise { + const repo = getRegistryRepository(); + const service = getRegistryService(); + + // List all agents (in-memory or postgres) + const { agents } = await repo.list({ limit: 1000 }); + + const now = Date.now(); + let unhealthyCount = 0; + let healthyCount = 0; + + for (const agent of agents) { + const lastSeen = agent.lastSeenAt ? new Date(agent.lastSeenAt).getTime() : 0; + const elapsed = now - lastSeen; + + // If TTL is set and elapsed > TTL, mark unhealthy + // If no TTL, use global TTL + const ttl = agent.ttlSeconds ? agent.ttlSeconds * 1000 : this.config.ttlMs; + + if (elapsed > ttl) { + if (agent.health !== "UNHEALTHY") { + console.log(`[health-checker] marking agent ${agent.id} (${agent.name}) as UNHEALTHY - last seen ${Math.round(elapsed / 1000)}s ago`); + await service.updateHealth(agent.id, "UNHEALTHY"); + unhealthyCount++; + } + } else if (elapsed < ttl / 2) { + // If recently seen and currently unhealthy, mark healthy + if (agent.health === "UNHEALTHY") { + console.log(`[health-checker] marking agent ${agent.id} (${agent.name}) as HEALTHY - recovered`); + await service.updateHealth(agent.id, "HEALTHY"); + healthyCount++; + } + } + } + + if (unhealthyCount > 0 || healthyCount > 0) { + console.log(`[health-checker] check done: ${unhealthyCount} -> UNHEALTHY, ${healthyCount} -> HEALTHY, total ${agents.length}`); + } + } +} + +let checkerInstance: HealthChecker | null = null; + +export function getHealthChecker(): HealthChecker { + if (!checkerInstance) checkerInstance = new HealthChecker(); + return checkerInstance; +} diff --git a/apps/gateway/src/modules/a2a-proxy.ts b/apps/gateway/src/modules/a2a-proxy.ts new file mode 100644 index 00000000..8da5a7de --- /dev/null +++ b/apps/gateway/src/modules/a2a-proxy.ts @@ -0,0 +1,131 @@ +import { validateUrlForSSRF } from "@agentmesh/core"; + +export interface ProxyOptions { + timeoutMs?: number; + retries?: number; + allowPrivate?: boolean; +} + +export interface ProxyRequest { + url: string; + method?: string; + headers?: Record; + body?: unknown; + traceId?: string; +} + +export interface ProxyResponse { + status: number; + headers: Record; + body: unknown; + latencyMs: number; +} + +export class A2AProxy { + async forward(req: ProxyRequest, opts: ProxyOptions = {}): Promise { + const timeout = opts.timeoutMs ?? 30000; + const url = req.url; + + // SSRF protection + const ssrfCheck = validateUrlForSSRF(url, { + allowPrivate: opts.allowPrivate ?? process.env.NODE_ENV !== "production", + allowLoopback: opts.allowPrivate ?? process.env.NODE_ENV !== "production", + }); + if (!ssrfCheck.allowed) { + throw new Error(`SSRF blocked: ${ssrfCheck.reason} - ${url}`); + } + + const controller = new AbortController(); + const timeoutId = setTimeout(() => controller.abort(), timeout); + const start = Date.now(); + + try { + const headers: Record = { + "Content-Type": "application/json", + Accept: "application/json", + "User-Agent": "AgentMesh-Gateway/0.1.0", + ...(req.headers ?? {}), + }; + + if (req.traceId) { + headers["X-Trace-Id"] = req.traceId; + headers["X-Request-Id"] = req.traceId; + } + + // Remove hop-by-hop headers + delete (headers as any)["connection"]; + delete (headers as any)["keep-alive"]; + delete (headers as any)["transfer-encoding"]; + + const res = await fetch(url, { + method: req.method ?? "POST", + headers, + body: req.body ? JSON.stringify(req.body) : undefined, + signal: controller.signal, + }); + + const latencyMs = Date.now() - start; + + let body: unknown; + const contentType = res.headers.get("content-type") ?? ""; + if (contentType.includes("application/json") || contentType.includes("+json")) { + try { + body = await res.json(); + } catch { + body = await res.text(); + } + } else { + body = await res.text(); + } + + const responseHeaders: Record = {}; + res.headers.forEach((value, key) => { + // Filter sensitive headers + if (!["set-cookie", "www-authenticate"].includes(key.toLowerCase())) { + responseHeaders[key] = value; + } + }); + + return { + status: res.status, + headers: responseHeaders, + body, + latencyMs, + }; + } catch (err) { + if ((err as any).name === "AbortError") { + throw new Error(`Proxy timeout after ${timeout}ms to ${url}`); + } + throw err; + } finally { + clearTimeout(timeoutId); + } + } + + async forwardWithRetry(req: ProxyRequest, opts: ProxyOptions = {}): Promise { + const retries = opts.retries ?? 0; + let lastErr: Error | null = null; + + for (let attempt = 0; attempt <= retries; attempt++) { + try { + return await this.forward(req, opts); + } catch (err) { + lastErr = err as Error; + if (attempt < retries) { + const backoff = Math.min(1000 * 2 ** attempt, 5000); + await new Promise((r) => setTimeout(r, backoff)); + console.warn(`[a2a-proxy] retry ${attempt + 1}/${retries} for ${req.url}: ${(err as Error).message}`); + } + } + } + + throw lastErr ?? new Error("Proxy failed after retries"); + } +} + +let proxyInstance: A2AProxy | null = null; + +export function getA2AProxy(): A2AProxy { + if (!proxyInstance) proxyInstance = new A2AProxy(); + return proxyInstance; +} diff --git a/apps/gateway/src/modules/registry-client.ts b/apps/gateway/src/modules/registry-client.ts new file mode 100644 index 00000000..545e7858 --- /dev/null +++ b/apps/gateway/src/modules/registry-client.ts @@ -0,0 +1,108 @@ +import { getRedis, RedisCache } from "@agentmesh/redis-adapter"; + +export type AgentRecord = any; + +// For Phase 1, gateway reads from same Redis cache as control-plane, plus fallback to control-plane HTTP +export interface RegistryClientConfig { + controlPlaneUrl?: string; + redisUrl?: string; + cacheTtlSeconds?: number; +} + +export class RegistryClient { + private cache: RedisCache; + private controlPlaneUrl: string; + private cacheTtl: number; + + constructor(config: RegistryClientConfig = {}) { + this.controlPlaneUrl = config.controlPlaneUrl ?? process.env.CONTROL_PLANE_URL ?? "http://localhost:3002"; + this.cacheTtl = config.cacheTtlSeconds ?? 60; + + try { + const redis = getRedis(); + this.cache = new RedisCache(redis, "agentmesh:agent:"); + } catch { + const mem = new Map(); + const fakeRedis = { + get: async (k: string) => mem.get(k) ?? null, + set: async (k: string, v: string) => { mem.set(k, v); }, + setex: async (k: string, _ttl: number, v: string) => { mem.set(k, v); }, + del: async (k: string) => { mem.delete(k); }, + exists: async (k: string) => (mem.has(k) ? 1 : 0), + } as any; + this.cache = new RedisCache(fakeRedis, "agentmesh:agent:"); + console.warn("[registry-client] Redis not available, using in-memory"); + } + } + + async getAgent(id: string): Promise { + // Try cache first + const cached = await this.cache.get(id); + if (cached) return cached as any; + + // Try control-plane HTTP + try { + const res = await fetch(`${this.controlPlaneUrl}/v1/agents/${id}`, { signal: AbortSignal.timeout(5000) }); + if (res.ok) { + const data = (await res.json()) as any; + const agent = data.agent; + if (agent) { + await this.cache.set(id, agent, this.cacheTtl); + return agent; + } + } + } catch (err) { + console.warn(`[registry-client] failed to fetch agent ${id} from control-plane: ${(err as Error).message}`); + } + + return null; + } + + async listAgents(filter: { skill?: string; capability?: string; region?: string; limit?: number } = {}): Promise { + const params = new URLSearchParams(); + if (filter.skill) params.set("skill", filter.skill); + if (filter.capability) params.set("capability", filter.capability); + if (filter.region) params.set("region", filter.region); + if (filter.limit) params.set("limit", String(filter.limit)); + + try { + const res = await fetch(`${this.controlPlaneUrl}/v1/agents?${params.toString()}`, { signal: AbortSignal.timeout(5000) }); + if (res.ok) { + const data = (await res.json()) as any; + return data.agents ?? []; + } + } catch (err) { + console.warn(`[registry-client] failed to list agents: ${(err as Error).message}`); + } + + return []; + } + + async discover(filter: { skill?: string; capability?: string; region?: string; search?: string; limit?: number }): Promise { + const params = new URLSearchParams(); + if (filter.skill) params.set("skill", filter.skill); + if (filter.capability) params.set("capability", filter.capability); + if (filter.region) params.set("region", filter.region); + if (filter.search) params.set("search", filter.search); + if (filter.limit) params.set("limit", String(filter.limit)); + + try { + const res = await fetch(`${this.controlPlaneUrl}/v1/agents/discover?${params.toString()}`, { signal: AbortSignal.timeout(5000) }); + if (res.ok) { + const data = (await res.json()) as any; + return data.agents ?? []; + } + } catch (err) { + console.warn(`[registry-client] discover failed: ${(err as Error).message}`); + } + + return []; + } +} + +let clientInstance: RegistryClient | null = null; + +export function getRegistryClient(): RegistryClient { + if (!clientInstance) clientInstance = new RegistryClient(); + return clientInstance; +} diff --git a/apps/gateway/src/routes/v1/index.ts b/apps/gateway/src/routes/v1/index.ts index 1101e754..3924ae60 100644 --- a/apps/gateway/src/routes/v1/index.ts +++ b/apps/gateway/src/routes/v1/index.ts @@ -1,13 +1,19 @@ import type { FastifyInstance } from "fastify"; +import { getRegistryClient } from "../../modules/registry-client.js"; +import { getA2AProxy } from "../../modules/a2a-proxy.js"; +import { tasksRoutes } from "./tasks.js"; export async function v1Routes(app: FastifyInstance) { + const registry = getRegistryClient(); + const proxy = getA2AProxy(); + app.get("/health", async () => { return { status: "ok", service: "gateway", version: app.config.VERSION, dataPlane: "active", - controlPlane: "cached-config (Phase 0)", + controlPlane: "cached-config (Phase 2: Redis + Control Plane HTTP + Tasks)", routingStrategies: app.routingEngine.listStrategies(), timestamp: new Date().toISOString(), }; @@ -16,7 +22,7 @@ export async function v1Routes(app: FastifyInstance) { app.get("/routes", async () => { return { routes: [], - message: "Routing table - full implementation in Phase 2", + message: "Routing table - Phase 2: engine integrated with task creation", }; }); @@ -27,33 +33,149 @@ export async function v1Routes(app: FastifyInstance) { version: app.config.VERSION, capabilities: { routing: ["round_robin", "least_loaded", "latency_aware", "weighted", "capability_match"], - streaming: "sse (Phase 2)", + streaming: "sse (Phase 2: implemented)", loadBalancing: "health-aware, latency-aware (Phase 2)", rateLimiting: "per org/project/agent/skill (Phase 3)", + discovery: ["local_registry", "direct_url", "well_known", "manual"], + a2aProxy: "Phase 1: JSON-RPC forwarding with SSRF protection", + tasks: "Phase 2: creation with routing, SSE streaming, delegation graph", }, - controlPlane: app.config.NATS_URL ? "connected via NATS (Phase 1+)" : "in-memory (Phase 0)", + controlPlane: app.config.NATS_URL ? "connected via NATS (Phase 2)" : "in-memory", }; }); - // Placeholder for A2A proxy - Phase 1+ - app.all("/a2a/*", async (req, reply) => { - return reply.status(501).send({ - error: { - code: "NOT_IMPLEMENTED", - message: "A2A Proxy will be implemented in Phase 1. This is Phase 0 foundation.", - traceId: req.traceId, - }, + // Agent discovery via gateway (proxies to control-plane) + app.get("/agents", async (req) => { + const { skill, capability, region, search, limit = "20" } = req.query as any; + const agents = await registry.discover({ + skill, + capability, + region, + search, + limit: parseInt(limit, 10) || 20, }); + return { agents, total: agents.length, source: "gateway->control-plane" }; }); - // Placeholder for task streaming - Phase 2 - app.get("/tasks/:id/stream", async (req, reply) => { - return reply.status(501).send({ - error: { - code: "NOT_IMPLEMENTED", - message: "Task streaming via SSE will be implemented in Phase 2", - taskId: (req.params as any).id, - }, - }); + app.get("/agents/:id", async (req, reply) => { + const { id } = req.params as { id: string }; + const agent = await registry.getAgent(id); + if (!agent) { + return reply.status(404).send({ error: { code: "AGENT_NOT_FOUND", message: `Agent not found: ${id}` } }); + } + return { agent }; }); + + // A2A Proxy - Forward JSON-RPC to agent + app.post("/agents/:id/invoke", async (req, reply) => { + const { id } = req.params as { id: string }; + const agent = await registry.getAgent(id); + if (!agent) { + return reply.status(404).send({ error: { code: "AGENT_NOT_FOUND", message: `Agent not found: ${id}` } }); + } + + const body = req.body as any; + if (!body || !body.method) { + return reply.status(400).send({ error: { code: "VALIDATION_ERROR", message: "Invalid JSON-RPC request, missing method" } }); + } + + try { + const result = await proxy.forward( + { + url: agent.url, + method: "POST", + body, + traceId: req.traceId, + headers: { + "X-AgentMesh-Agent-Id": id, + "X-AgentMesh-Trace-Id": req.traceId, + }, + }, + { timeoutMs: app.config.REQUEST_TIMEOUT_MS, retries: 0, allowPrivate: true } + ); + + try { + await app.eventBus.publish({ + id: `evt_${Date.now()}`, + type: "message.sent", + source: "gateway", + subject: `agent.${id}`, + data: { agentId: id, method: body.method, status: result.status, latencyMs: result.latencyMs }, + timestamp: new Date().toISOString(), + traceId: req.traceId, + }); + } catch {} + + return reply.status(result.status).send(result.body); + } catch (err) { + const e = err as Error; + app.log.error({ err: e, agentId: id, traceId: req.traceId }, "A2A proxy failed"); + return reply.status(502).send({ error: { code: "PROXY_FAILED", message: e.message, agentId: id, traceId: req.traceId } }); + } + }); + + // Generic A2A proxy - /v1/a2a/:id/* -> forwards to agent + app.all("/a2a/:id/*", async (req, reply) => { + const { id } = req.params as { id: string; "*": string }; + const agent = await registry.getAgent(id); + if (!agent) { + return reply.status(404).send({ error: { code: "AGENT_NOT_FOUND", message: `Agent not found: ${id}` } }); + } + + const extraPath = (req.params as any)["*"] ?? ""; + const targetUrl = `${agent.url.replace(/\/$/, "")}/${extraPath}`.replace(/\/$/, ""); + const queryString = req.url.includes("?") ? req.url.split("?")[1] : ""; + const finalUrl = queryString ? `${targetUrl}?${queryString}` : targetUrl; + + try { + const result = await proxy.forward( + { + url: finalUrl, + method: req.method, + body: req.method !== "GET" && req.method !== "HEAD" ? req.body : undefined, + traceId: req.traceId, + }, + { timeoutMs: app.config.REQUEST_TIMEOUT_MS, allowPrivate: true } + ); + + for (const [k, v] of Object.entries(result.headers)) { + if (!["content-length", "transfer-encoding"].includes(k.toLowerCase())) { + reply.header(k, v); + } + } + + return reply.status(result.status).send(result.body); + } catch (err) { + const e = err as Error; + return reply.status(502).send({ error: { code: "PROXY_FAILED", message: e.message, agentId: id } }); + } + }); + + // Routing preview - Phase 2 with real routing + app.post("/route", async (req) => { + const { skill, capability, region, strategy = "capability_match" } = req.body as any; + const candidates = await registry.discover({ skill, capability, region, limit: 50 }); + + if (candidates.length === 0) { + return { agent: null, candidates: [], message: "No agents found for criteria" }; + } + + const routingCandidates = candidates.map((a) => ({ + agent: a as any, + health: (a as any).health ?? "UNKNOWN", + score: 1, + })); + + const selected = app.routingEngine.route(routingCandidates as any, { skill, capability, region }, strategy); + + return { + agent: selected?.agent ?? null, + candidates: candidates.slice(0, 10), + strategy, + total: candidates.length, + }; + }); + + // Register tasks routes - Phase 2 + await app.register(tasksRoutes, { prefix: "/tasks" }); } diff --git a/apps/gateway/src/routes/v1/tasks.ts b/apps/gateway/src/routes/v1/tasks.ts new file mode 100644 index 00000000..6e5fe3be --- /dev/null +++ b/apps/gateway/src/routes/v1/tasks.ts @@ -0,0 +1,159 @@ +import type { FastifyInstance } from "fastify"; + +export async function tasksRoutes(app: FastifyInstance) { + const controlPlaneUrl = process.env.CONTROL_PLANE_URL ?? "http://localhost:3002"; + + // Helper to proxy to control-plane + async function proxyToControlPlane(path: string, opts: { method?: string; body?: unknown; headers?: Record } = {}) { + const url = `${controlPlaneUrl}${path}`; + const res = await fetch(url, { + method: opts.method ?? "GET", + headers: { "Content-Type": "application/json", ...(opts.headers ?? {}) }, + body: opts.body ? JSON.stringify(opts.body) : undefined, + signal: AbortSignal.timeout(10000), + }); + const data = await res.json().catch(() => ({})); + return { status: res.status, data }; + } + + // List tasks + app.get("/", async (req, reply) => { + const query = req.url.includes("?") ? req.url.split("?")[1] : ""; + const { status, data } = await proxyToControlPlane(`/v1/tasks?${query}`); + return reply.status(status).send(data); + }); + + // Get task + app.get("/:id", async (req, reply) => { + const { id } = req.params as { id: string }; + const { status, data } = await proxyToControlPlane(`/v1/tasks/${id}`); + return reply.status(status).send(data); + }); + + // Create task - with routing + app.post("/", async (req, reply) => { + const body = req.body as any; + if (!body?.agentId && (body?.skill || body?.capability)) { + // If no agentId but skill provided, route to find agent + try { + const { getRegistryClient } = await import("../../modules/registry-client.js"); + const registry = getRegistryClient(); + const candidates = await registry.discover({ + skill: body.skill, + capability: body.capability, + region: body.region, + limit: 10, + }); + + if (candidates.length === 0) { + return reply.status(404).send({ error: { code: "NO_AGENT_FOUND", message: `No agent found for skill=${body.skill} capability=${body.capability}` } }); + } + + // Use routing engine + const routingCandidates = candidates.map((a: any) => ({ agent: a, health: a.health ?? "UNKNOWN", score: 1 })); + const selected = app.routingEngine.route(routingCandidates as any, { skill: body.skill, capability: body.capability, region: body.region }, body.strategy ?? "capability_match"); + + if (!selected) { + return reply.status(404).send({ error: { code: "ROUTING_FAILED", message: "Routing failed to select agent" } }); + } + + body.agentId = selected.agent.id; + app.log.info({ routedTo: body.agentId, skill: body.skill, strategy: body.strategy }, "task routed"); + } catch (err) { + app.log.error({ err }, "routing failed"); + return reply.status(500).send({ error: { code: "ROUTING_FAILED", message: (err as Error).message } }); + } + } + + const { status, data } = await proxyToControlPlane("/v1/tasks", { method: "POST", body }); + return reply.status(status).send(data); + }); + + // Cancel + app.post("/:id/cancel", async (req, reply) => { + const { id } = req.params as { id: string }; + const { status, data } = await proxyToControlPlane(`/v1/tasks/${id}/cancel`, { method: "POST", body: req.body }); + return reply.status(status).send(data); + }); + + // Complete (for agents to report) + app.post("/:id/complete", async (req, reply) => { + const { id } = req.params as { id: string }; + const { status, data } = await proxyToControlPlane(`/v1/tasks/${id}/complete`, { method: "POST", body: req.body }); + return reply.status(status).send(data); + }); + + // Fail + app.post("/:id/fail", async (req, reply) => { + const { id } = req.params as { id: string }; + const { status, data } = await proxyToControlPlane(`/v1/tasks/${id}/fail`, { method: "POST", body: req.body }); + return reply.status(status).send(data); + }); + + // Retry + app.post("/:id/retry", async (req, reply) => { + const { id } = req.params as { id: string }; + const { status, data } = await proxyToControlPlane(`/v1/tasks/${id}/retry`, { method: "POST" }); + return reply.status(status).send(data); + }); + + // History + app.get("/:id/history", async (req, reply) => { + const { id } = req.params as { id: string }; + const { status, data } = await proxyToControlPlane(`/v1/tasks/${id}/history`); + return reply.status(status).send(data); + }); + + // Graph + app.get("/:id/graph", async (req, reply) => { + const { id } = req.params as { id: string }; + const { status, data } = await proxyToControlPlane(`/v1/tasks/${id}/graph`); + return reply.status(status).send(data); + }); + + // Streaming - proxy SSE + app.get("/:id/stream", async (req, reply) => { + const { id } = req.params as { id: string }; + const url = `${controlPlaneUrl}/v1/tasks/${id}/stream`; + + try { + const res = await fetch(url, { + headers: { Accept: "text/event-stream" }, + }); + + if (!res.ok || !res.body) { + return reply.status(res.status).send({ error: { code: "STREAM_FAILED", message: `Failed to connect to task stream: ${res.status}` } }); + } + + reply.raw.writeHead(200, { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + "X-Accel-Buffering": "no", + }); + + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + + // Forward stream + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + const chunk = decoder.decode(value, { stream: true }); + reply.raw.write(chunk); + } + } catch (err) { + app.log.warn({ err, taskId: id }, "stream forwarding error"); + } finally { + try { + reply.raw.end(); + } catch {} + } + + return reply; + } catch (err) { + return reply.status(502).send({ error: { code: "STREAM_PROXY_FAILED", message: (err as Error).message } }); + } + }); +} diff --git a/packages/a2a-protocol/src/fetcher.ts b/packages/a2a-protocol/src/fetcher.ts new file mode 100644 index 00000000..d00407c5 --- /dev/null +++ b/packages/a2a-protocol/src/fetcher.ts @@ -0,0 +1,113 @@ +import { AgentCardSchema, type AgentCard, WELL_KNOWN_AGENT_CARD_PATH } from "./index.js"; + +export interface FetchCardOptions { + timeoutMs?: number; + allowPrivate?: boolean; // allow private IPs in dev + allowLoopback?: boolean; + headers?: Record; + validateCard?: boolean; +} + +// Simple SSRF check - duplicated to avoid circular dep, but core will be used in services +function isPrivateHost(hostname: string): boolean { + const lower = hostname.toLowerCase(); + if (lower === "localhost" || lower === "127.0.0.1" || lower === "::1") return true; + if (lower === "169.254.169.254") return true; + if (/^10\./.test(lower)) return true; + if (/^192\.168\./.test(lower)) return true; + if (/^172\.(1[6-9]|2\d|3[0-1])\./.test(lower)) return true; + return false; +} + +function validateUrl(urlString: string, opts: FetchCardOptions): void { + const url = new URL(urlString); + if (!["http:", "https:"].includes(url.protocol)) { + throw new Error(`Only http/https allowed, got ${url.protocol}`); + } + if (!opts.allowPrivate && isPrivateHost(url.hostname) && !opts.allowLoopback) { + // In production, block. In dev, allow if flag set + if (process.env.NODE_ENV === "production") { + throw new Error(`SSRF blocked: private host ${url.hostname}`); + } + // In dev, log warning but allow if allowPrivate or allowLoopback + if (!opts.allowPrivate && !opts.allowLoopback) { + throw new Error(`SSRF blocked: private host ${url.hostname} (use allowPrivate in dev)`); + } + } +} + +export async function fetchAgentCard(url: string, opts: FetchCardOptions = {}): Promise<{ card: AgentCard; raw: unknown; url: string; fetchedAt: string }> { + const timeout = opts.timeoutMs ?? 10000; + validateUrl(url, opts); + + const controller = new AbortController(); + const timeoutId = setTimeout(() => controller.abort(), timeout); + + try { + const res = await fetch(url, { + signal: controller.signal, + headers: { + Accept: "application/json", + "User-Agent": "AgentMesh/0.1.0 (A2A Card Fetcher)", + ...opts.headers, + }, + }); + + if (!res.ok) { + throw new Error(`Failed to fetch Agent Card: ${res.status} ${res.statusText} from ${url}`); + } + + const contentType = res.headers.get("content-type") ?? ""; + if (!contentType.includes("application/json") && !contentType.includes("text/json") && !contentType.includes("+json")) { + // Some agents may return with text/plain, so we try anyway but warn + console.warn(`[a2a] unexpected content-type ${contentType} for ${url}`); + } + + const raw = await res.json(); + const parsed = opts.validateCard !== false ? AgentCardSchema.parse(raw) : (raw as AgentCard); + + return { + card: parsed, + raw, + url, + fetchedAt: new Date().toISOString(), + }; + } catch (err) { + if ((err as any).name === "AbortError") { + throw new Error(`Timeout fetching Agent Card from ${url} after ${timeout}ms`); + } + throw err; + } finally { + clearTimeout(timeoutId); + } +} + +export async function fetchWellKnownCard(baseUrl: string, opts: FetchCardOptions = {}): Promise<{ card: AgentCard; raw: unknown; url: string; fetchedAt: string }> { + const base = new URL(baseUrl); + const wellKnownUrl = `${base.origin}${WELL_KNOWN_AGENT_CARD_PATH}`; + return fetchAgentCard(wellKnownUrl, opts); +} + +export async function fetchCardWithFallbacks(urls: string[], opts: FetchCardOptions = {}): Promise<{ card: AgentCard; raw: unknown; url: string; fetchedAt: string }> { + let lastError: Error | null = null; + for (const url of urls) { + try { + return await fetchAgentCard(url, opts); + } catch (err) { + lastError = err as Error; + console.warn(`[a2a] failed to fetch ${url}: ${(err as Error).message}`); + } + } + throw lastError ?? new Error("No URLs provided for card fetching"); +} + +export function buildDiscoveryUrls(baseUrl: string): string[] { + const base = new URL(baseUrl); + return [ + baseUrl, // direct + `${base.origin}${WELL_KNOWN_AGENT_CARD_PATH}`, // well-known + `${base.origin}/.well-known/a2a.json`, // alternative + `${base.origin}/agent.json`, // common alternative + `${base.origin}/.well-known/agent-card.json`, + ]; +} diff --git a/packages/a2a-protocol/src/index.ts b/packages/a2a-protocol/src/index.ts index 21a33668..1d6d50f0 100644 --- a/packages/a2a-protocol/src/index.ts +++ b/packages/a2a-protocol/src/index.ts @@ -216,3 +216,5 @@ export function negotiateProtocolVersion(requested: string[], supported: readonl } return null; } + +export * from "./fetcher.js"; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index dc82d0d5..67ff5b89 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -182,3 +182,5 @@ export function isExpired(expiresAt: string | Date): boolean { const d = typeof expiresAt === "string" ? new Date(expiresAt) : expiresAt; return d.getTime() < Date.now(); } + +export * from "./security.js"; diff --git a/packages/core/src/security.ts b/packages/core/src/security.ts new file mode 100644 index 00000000..3ddba03e --- /dev/null +++ b/packages/core/src/security.ts @@ -0,0 +1,130 @@ +/** + * SSRF Protection - Production-grade URL validation + * Blocks private IPs, localhost, metadata endpoints, etc. + */ + +const PRIVATE_IP_RANGES = [ + /^127\./, // loopback + /^10\./, // private 10.0.0.0/8 + /^192\.168\./, // private 192.168.0.0/16 + /^172\.(1[6-9]|2\d|3[0-1])\./, // private 172.16.0.0/12 + /^169\.254\./, // link-local + AWS metadata + /^0\.0\.0\.0/, + /^::1$/, + /^fc00:/, // unique local + /^fe80:/, // link-local +]; + +const BLOCKED_HOSTNAMES = new Set([ + "localhost", + "metadata.google.internal", + "metadata.google", + "instance-data", + "169.254.169.254", +]); + +const BLOCKED_SUFFIXES = [".internal", ".local", ".localhost"]; + +export interface SSRFOptions { + allowPrivate?: boolean; // allow private IPs in development + allowLoopback?: boolean; + allowedHosts?: string[]; // explicit allowlist + blockedHosts?: string[]; // explicit blocklist +} + +export function isPrivateIP(ip: string): boolean { + return PRIVATE_IP_RANGES.some((re) => re.test(ip)); +} + +export function isBlockedHostname(hostname: string, opts: SSRFOptions = {}): boolean { + const lower = hostname.toLowerCase(); + + if (opts.allowedHosts?.includes(lower)) return false; + if (opts.blockedHosts?.includes(lower)) return true; + + if (BLOCKED_HOSTNAMES.has(lower)) return true; + + for (const suffix of BLOCKED_SUFFIXES) { + if (lower.endsWith(suffix)) return true; + } + + // Check if IP and private + if (/^\d+\.\d+\.\d+\.\d+$/.test(lower) || lower.includes(":")) { + if (isPrivateIP(lower)) { + if (lower.startsWith("127.") && opts.allowLoopback) return false; + if (opts.allowPrivate) return false; + return true; + } + } + + // localhost + if (lower === "localhost" || lower === "127.0.0.1" || lower === "::1") { + if (opts.allowLoopback) return false; + return true; + } + + // AWS metadata + if (lower === "169.254.169.254") return true; + + return false; +} + +export function validateUrlForSSRF(urlString: string, opts: SSRFOptions = {}): { allowed: boolean; reason?: string } { + let url: URL; + try { + url = new URL(urlString); + } catch { + return { allowed: false, reason: "Invalid URL format" }; + } + + // Only allow http/https + if (!["http:", "https:"].includes(url.protocol)) { + return { allowed: false, reason: `Protocol ${url.protocol} not allowed, only http/https` }; + } + + // Block credentials in URL + if (url.username || url.password) { + return { allowed: false, reason: "URL with credentials not allowed" }; + } + + // Check hostname + if (isBlockedHostname(url.hostname, opts)) { + return { allowed: false, reason: `Hostname ${url.hostname} is blocked (SSRF protection)` }; + } + + // Block non-standard ports that are commonly used for internal services + const port = url.port ? parseInt(url.port, 10) : url.protocol === "https:" ? 443 : 80; + const blockedPorts = [22, 25, 3306, 5432, 6379, 11211, 27017, 9200]; + if (blockedPorts.includes(port) && !opts.allowPrivate) { + return { allowed: false, reason: `Port ${port} is blocked` }; + } + + return { allowed: true }; +} + +export function assertUrlAllowed(urlString: string, opts: SSRFOptions = {}): void { + const result = validateUrlForSSRF(urlString, opts); + if (!result.allowed) { + throw new Error(`SSRF blocked: ${result.reason} - ${urlString}`); + } +} + +// Webhook URL validation - stricter +export function validateWebhookUrl(urlString: string, opts: SSRFOptions = {}): { allowed: boolean; reason?: string } { + const base = validateUrlForSSRF(urlString, opts); + if (!base.allowed) return base; + + const url = new URL(urlString); + + // Webhooks must be https in production + if (process.env.NODE_ENV === "production" && url.protocol !== "https:") { + return { allowed: false, reason: "Webhook URL must be https in production" }; + } + + // No fragments + if (url.hash) { + return { allowed: false, reason: "Webhook URL must not contain fragment" }; + } + + return { allowed: true }; +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e950d08a..b3e6ea31 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -144,6 +144,9 @@ importers: '@fastify/sensible': specifier: ^6.0.0 version: 6.0.5 + drizzle-orm: + specifier: ^0.35.3 + version: 0.35.3(@libsql/client-wasm@0.18.0)(@opentelemetry/api@1.9.1)(@types/pg@8.6.1)(postgres@3.4.9) fastify: specifier: ^5.3.2 version: 5.12.5