From 1cc11557440691984f13c90b08c6bd7c61ffa1d3 Mon Sep 17 00:00:00 2001 From: qianghan Date: Sun, 19 Jul 2026 16:41:25 -0700 Subject: [PATCH] =?UTF-8?q?feat(capabilities):=20PR-3=20=E2=80=94=20discov?= =?UTF-8?q?ery=E2=86=92registry=20sync=20(zero-code=20onboarding=20G2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reads Capability Descriptors (PR-1) that a provider publishes in their orchestrator's /discovery advertisement and runs them through the shipped cap-lifecycle: validate → dedup by semantic_key → NEW identity registers into the universal overlay (so the agent can describe/list it), EXISTING same-name is an idempotent refresh, DIFFERENT name is a synonym and is SKIPPED + reported. This turns "provider does work" into "it just works" — no hand-run cap_add, no hardcoded per-cap map. - lib/capabilities/discovery-sync.ts: PURE projection + planner (descriptorToEntry, synthesizeUsage from the io contract, planDiscoverySync, extractDescriptors — tolerant of array / {runners} / {orchestrators} shapes, ignores legacy no-descriptor entries) + thin I/O (applyDiscoverySync, fetchDiscoveryDescriptors). - app/api/cron/capability-sync/route.ts: DEFAULT-OFF (CAP_DISCOVERY_SYNC_ENABLED), CRON_SECRET-guarded, best-effort per URL. SAFETY: additive (new module + default-off route, no edits to existing paths); fail-open (bad descriptor/URL skipped, never thrown); writes only the operator-governed universal overlay. Golden green; 12 unit tests. Plan: LIVE-RUNNER-CAP-ADOPTION-PROPOSAL.html §4 (PR-3). Reuses PR-1 descriptor schema + the shipped registry-overlay + cap_add dedup semantics. Co-Authored-By: Claude Opus 4.8 --- app/api/cron/capability-sync/route.ts | 54 +++++ lib/capabilities/discovery-sync.ts | 232 +++++++++++++++++++ tests/unit/capability-discovery-sync.test.ts | 150 ++++++++++++ 3 files changed, 436 insertions(+) create mode 100644 app/api/cron/capability-sync/route.ts create mode 100644 lib/capabilities/discovery-sync.ts create mode 100644 tests/unit/capability-discovery-sync.test.ts diff --git a/app/api/cron/capability-sync/route.ts b/app/api/cron/capability-sync/route.ts new file mode 100644 index 00000000..9b025649 --- /dev/null +++ b/app/api/cron/capability-sync/route.ts @@ -0,0 +1,54 @@ +/** + * Capability discovery→registry sync cron — zero-code onboarding PR-3 (gap G2). + * + * DEFAULT-OFF: inert unless CAP_DISCOVERY_SYNC_ENABLED=1, so merging this route + * changes nothing in production until deliberately enabled. When on, it fetches + * each orchestrator /discovery in CAP_DISCOVERY_URLS (comma-separated), extracts + * + validates Capability Descriptors (PR-1), and registers new identities into + * the universal overlay via the shipped lifecycle (dedup-guarded). Best-effort: + * a bad URL or descriptor is skipped, never fatal. + * + * Protected by CRON_SECRET when set (Vercel Cron sends it as a bearer). + */ +import { NextResponse } from "next/server"; +import { + fetchDiscoveryDescriptors, + applyDiscoverySync, +} from "@/lib/capabilities/discovery-sync"; +import type { CapabilityDescriptor } from "@/lib/capabilities/descriptor"; + +export const dynamic = "force-dynamic"; + +function authorized(req: Request): boolean { + const secret = process.env.CRON_SECRET; + if (!secret) return true; // no secret configured → allow (staging convenience) + const got = req.headers.get("authorization") ?? ""; + return got === `Bearer ${secret}`; +} + +export async function GET(req: Request) { + if (process.env.CAP_DISCOVERY_SYNC_ENABLED !== "1") { + return NextResponse.json({ enabled: false }); + } + if (!authorized(req)) { + return NextResponse.json({ error: "unauthorized" }, { status: 401 }); + } + + const urls = (process.env.CAP_DISCOVERY_URLS ?? "") + .split(",") + .map((u) => u.trim()) + .filter(Boolean); + + const all: CapabilityDescriptor[] = []; + let invalid = 0; + const perUrl: Array<{ url: string; valid: number; invalid: number }> = []; + for (const url of urls) { + const { valid, invalid: bad } = await fetchDiscoveryDescriptors(url); + all.push(...valid); + invalid += bad.length; + perUrl.push({ url, valid: valid.length, invalid: bad.length }); + } + + const report = await applyDiscoverySync(all, invalid); + return NextResponse.json({ enabled: true, sources: perUrl, ...report }); +} diff --git a/lib/capabilities/discovery-sync.ts b/lib/capabilities/discovery-sync.ts new file mode 100644 index 00000000..0203a50e --- /dev/null +++ b/lib/capabilities/discovery-sync.ts @@ -0,0 +1,232 @@ +/** + * Discovery → registry sync — zero-code onboarding PR-3 (plan + * LIVE-RUNNER-CAP-ADOPTION-PROPOSAL.html §4, gap G2). + * + * SERVER-ONLY (transitively imports @vercel/blob via registry-overlay). Reads + * Capability Descriptors (PR-1) a provider publishes in their orchestrator's + * `/discovery` advertisement and runs them through the shipped cap-lifecycle: + * + * validate (PR-1) → dedup by semantic_key → NEW identity ⇒ register it into the + * universal overlay (so the agent can describe/list it); EXISTING identity same + * name ⇒ idempotent refresh; DIFFERENT name ⇒ synonym, SKIP + report. + * + * This is what turns "provider does work" into "it just works": no hand-run + * cap_add, no hardcoded per-cap map. The projection + planner are PURE (no I/O), + * so they unit-test without a network or Blob. The apply/fetch wrappers are thin. + * + * SAFETY: additive (new module + a default-OFF cron route); fail-open (a bad + * descriptor is skipped, never thrown); writes only the operator-governed + * universal overlay. Merging changes nothing until CAP_DISCOVERY_SYNC_ENABLED=1. + * NOTE: promoting a discovered cap to UNIVERSAL scope is governed by the sync + * being operator-triggered; per-descriptor scope governance is a later PR. + */ +import { + validateDescriptor, + computeSemanticKey, + type CapabilityDescriptor, + type Capability, +} from "./descriptor"; +import type { RegistryEntry, OutputKind } from "./generate-registry"; +import type { UsageBlock, UsageInput } from "./usage"; +import { readRegistry, upsertOverlayEntry } from "./registry-overlay"; + +const MEDIA_KINDS = new Set(["image", "video", "audio", "3d"]); + +const ACTIONS_BY_OUTPUT: Record = { + image: ["generate", "restyle", "edit"], + video: ["animate", "generate"], + audio: ["tts", "music"], + "3d": ["generate"], + text: ["generate"], + tool: ["run"], +}; + +/** Canonical name from identity (mirrors cap_add §4.1): media ⇒ family[-variant]-modality; tool/mcp ⇒ family-modality. A descriptor may pin an explicit name (an existing identity). */ +export function canonicalName(cap: Capability): string { + if (cap.name) return cap.name; + const medialess = cap.kind === "tool" || cap.kind === "mcp"; + const parts = medialess + ? [cap.family, cap.modality] + : [cap.family, cap.variant, cap.modality]; + return parts.filter(Boolean).join("-").toLowerCase(); +} + +/** Synthesize the agent-usage block (describe_capability card) from the io contract. */ +export function synthesizeUsage(cap: Capability, name: string): UsageBlock { + const invokeVia = MEDIA_KINDS.has(cap.output_kind) ? "create_media" : "run_capability"; + const actions = ACTIONS_BY_OUTPUT[cap.output_kind] ?? ["generate"]; + + const inputs: Record = {}; + const exampleArgs: Record = {}; + for (const [param, spec] of Object.entries(cap.io.inputs)) { + const agentKey = spec.from ?? param; + inputs[agentKey] = { + type: spec.type ?? "string", + required: spec.required, + ...(spec.values ? { enum: [...spec.values] } : {}), + ...(spec.default !== undefined ? { default: spec.default } : {}), + note: `wire=${spec.wire}`, + }; + if (spec.required) exampleArgs[agentKey] = spec.default ?? `<${param}>`; + } + + const example = + invokeVia === "create_media" + ? { tool: "create_media", args: { action: actions[0], model_override: name, ...exampleArgs } } + : { tool: "run_capability", args: { capability: name, ...exampleArgs } }; + + return { + invoke_via: invokeVia, + actions, + inputs, + output: { kind: cap.io.output.kind }, + good_for: cap.good_for ?? `${cap.family} ${cap.modality}`, + ...(cap.avoid_for ? { avoid_for: cap.avoid_for } : {}), + example, + }; +} + +/** Project a validated descriptor into a universal RegistryEntry (active, live). */ +export function descriptorToEntry(cap: Capability): RegistryEntry { + const name = canonicalName(cap); + return { + name, + // Descriptor kinds are a superset (adds "live-runner"); the core union + // widening is a separate PR, so cast here — the overlay stores JSON and + // consumers that switch on kind fall back gracefully. + kind: cap.kind as RegistryEntry["kind"], + modality: cap.modality, + family: cap.family, + variant: cap.variant ?? null, + semantic_key: cap.semantic_key ?? computeSemanticKey(cap), + aliases: cap.aliases ?? [], + output_kind: cap.output_kind as OutputKind, + fallback_chain: null, + sla: null, + unit_kind: cap.offering.price.display_unit ?? null, + display_unit: cap.offering.price.display_unit ?? null, + display_price_usd: cap.offering.price.display_usd ?? null, + status: "active", + owner_scope: "universal", + usage: synthesizeUsage(cap, name), + }; +} + +export interface SyncPlan { + /** New/refreshed identities to upsert into the universal overlay. */ + register: RegistryEntry[]; + /** Descriptors refused because their key already belongs to another name. */ + synonymSkipped: Array<{ name: string; key: string; owner: string }>; +} + +/** + * PURE planner. Dedups descriptors against the existing registry view AND within + * the batch (mirrors cap_seed): a key owned by a DIFFERENT name is a synonym and + * is skipped; same name (or unseen) registers. + */ +export function planDiscoverySync( + descriptors: CapabilityDescriptor[], + existing: RegistryEntry[], +): SyncPlan { + const seen = new Map( + existing.map((e) => [ + e.semantic_key ?? + computeSemanticKey({ + kind: e.kind, + modality: e.modality ?? "", + family: e.family ?? "", + variant: e.variant ?? null, + }), + e.name, + ]), + ); + + const register: RegistryEntry[] = []; + const synonymSkipped: SyncPlan["synonymSkipped"] = []; + + for (const d of descriptors) { + const entry = descriptorToEntry(d.capability); + const owner = seen.get(entry.semantic_key); + if (owner && owner !== entry.name) { + synonymSkipped.push({ name: entry.name, key: entry.semantic_key, owner }); + continue; + } + register.push(entry); + seen.set(entry.semantic_key, entry.name); + } + return { register, synonymSkipped }; +} + +export interface SyncReport { + registered: string[]; + synonymSkipped: SyncPlan["synonymSkipped"]; + invalid: number; +} + +/** Apply a sync: read the universal view, plan, upsert. Best-effort (per-entry). */ +export async function applyDiscoverySync( + descriptors: CapabilityDescriptor[], + invalidCount = 0, +): Promise { + const existing = await readRegistry(); // universal view + const plan = planDiscoverySync(descriptors, existing); + const registered: string[] = []; + for (const entry of plan.register) { + try { + await upsertOverlayEntry(entry, { scope: "universal" }); + registered.push(entry.name); + } catch { + // best-effort: one Blob write failing must not abort the rest + } + } + return { registered, synonymSkipped: plan.synonymSkipped, invalid: invalidCount }; +} + +/** + * PURE extraction: pull + validate Capability Descriptors out of a parsed + * `/discovery` payload. Tolerant of shapes — a bare array, `{runners:[…]}`, or + * `{orchestrators:[{runners:[…]}]}`. Entries without a `capability` block (legacy + * app-examples advertisements) are simply ignored (back-compat). + */ +export function extractDescriptors(discovery: unknown): { + valid: CapabilityDescriptor[]; + invalid: Array<{ errors: string[] }>; +} { + const entries: unknown[] = Array.isArray(discovery) + ? discovery + : isRecord(discovery) && Array.isArray(discovery.runners) + ? discovery.runners + : isRecord(discovery) && Array.isArray(discovery.orchestrators) + ? discovery.orchestrators.flatMap((o) => + isRecord(o) && Array.isArray(o.runners) ? o.runners : [], + ) + : []; + + const valid: CapabilityDescriptor[] = []; + const invalid: Array<{ errors: string[] }> = []; + for (const e of entries) { + if (!isRecord(e) || e.capability === undefined) continue; // legacy / no descriptor + const res = validateDescriptor({ capability: e.capability }); + if (res.ok) valid.push(res.value); + else invalid.push({ errors: res.errors }); + } + return { valid, invalid }; +} + +/** Fetch one orchestrator's /discovery and extract descriptors. Fail-open. */ +export async function fetchDiscoveryDescriptors(url: string): Promise<{ + valid: CapabilityDescriptor[]; + invalid: Array<{ errors: string[] }>; +}> { + try { + const res = await fetch(url, { cache: "no-store" }); + if (!res.ok) return { valid: [], invalid: [] }; + return extractDescriptors(await res.json()); + } catch { + return { valid: [], invalid: [] }; + } +} + +function isRecord(v: unknown): v is Record { + return typeof v === "object" && v !== null; +} diff --git a/tests/unit/capability-discovery-sync.test.ts b/tests/unit/capability-discovery-sync.test.ts new file mode 100644 index 00000000..146f2783 --- /dev/null +++ b/tests/unit/capability-discovery-sync.test.ts @@ -0,0 +1,150 @@ +import { describe, it, expect } from "vitest"; +import { + extractDescriptors, + planDiscoverySync, + descriptorToEntry, + canonicalName, +} from "@/lib/capabilities/discovery-sync"; +import { parseDescriptor } from "@/lib/capabilities/descriptor"; +import type { RegistryEntry } from "@/lib/capabilities/generate-registry"; + +const screenAgentCap = { + kind: "live-runner", + family: "screen-agent", + modality: "video-understanding", + variant: null, + output_kind: "text", + good_for: "screen recording -> bug report", + io: { + endpoint: "/analyze", + inputs: { + video: { wire: "video_b64", from: "source_url", required: true }, + preset: { wire: "preset", type: "enum", values: ["bug-report", "agent-eval"], default: "bug-report" }, + }, + output: { kind: "text", fields: ["report_markdown", "summary"], primary: "report_markdown" }, + }, + offering: { + app: "emran/screen-agent", + mode: "single-shot", + capacity: 1, + price: { price_per_unit: 138889, pixels_per_unit: 1, unit: "WEI", display_usd: 0.000139, display_unit: "second" }, + }, +} as const; + +const fluxDevOffering = { + kind: "ai", + name: "flux-dev", + family: "flux", + variant: "dev", + modality: "t2i", + output_kind: "image", + io: { + endpoint: "/generate", + inputs: { prompt: { wire: "prompt", from: "prompt", required: true } }, + output: { kind: "url", fields: ["url"], primary: "url" }, + }, + offering: { app: "you/fal-app", mode: "single-shot", capacity: 4, price: { price_per_unit: 100, pixels_per_unit: 1, unit: "WEI" } }, +} as const; + +describe("discovery-sync (PR-3)", () => { + describe("canonicalName", () => { + it("derives family[-variant]-modality for media caps", () => { + expect(canonicalName({ ...(fluxDevOffering as any), name: undefined })).toBe("flux-dev-t2i"); + }); + it("honors an explicit name (existing identity)", () => { + expect(canonicalName(fluxDevOffering as any)).toBe("flux-dev"); + }); + it("derives family-modality for a new live-runner cap", () => { + expect(canonicalName(screenAgentCap as any)).toBe("screen-agent-video-understanding"); + }); + }); + + describe("descriptorToEntry", () => { + it("projects identity + price + a synthesized usage block", () => { + const entry = descriptorToEntry(screenAgentCap as any); + expect(entry.name).toBe("screen-agent-video-understanding"); + expect(entry.kind).toBe("live-runner"); + expect(entry.semantic_key).toBe("live-runner:video-understanding:screen-agent:-"); + expect(entry.status).toBe("active"); + expect(entry.owner_scope).toBe("universal"); + expect(entry.display_price_usd).toBe(0.000139); + // text output → run_capability, and the required input is keyed by `from` + expect(entry.usage?.invoke_via).toBe("run_capability"); + expect(entry.usage?.inputs.source_url?.required).toBe(true); + expect(entry.usage?.inputs.source_url?.note).toBe("wire=video_b64"); + }); + it("media output → create_media invoke_via", () => { + const entry = descriptorToEntry(fluxDevOffering as any); + expect(entry.usage?.invoke_via).toBe("create_media"); + }); + }); + + describe("planDiscoverySync", () => { + const existing: RegistryEntry[] = [ + { + name: "flux-dev", kind: "ai", modality: "t2i", family: "flux", variant: "dev", + semantic_key: "ai:t2i:flux:dev", aliases: [], output_kind: "image", + fallback_chain: null, sla: null, unit_kind: null, display_unit: null, + display_price_usd: null, status: "active", owner_scope: "universal", + }, + ]; + + it("registers a genuinely new identity", () => { + const plan = planDiscoverySync([parseDescriptor({ capability: screenAgentCap })], existing); + expect(plan.register.map((e) => e.name)).toEqual(["screen-agent-video-understanding"]); + expect(plan.synonymSkipped).toHaveLength(0); + }); + + it("registers a new PROVIDER of an existing identity when the name matches (idempotent refresh)", () => { + // flux-dev offering reuses the exact identity + name → not a synonym, allowed. + const plan = planDiscoverySync([parseDescriptor({ capability: fluxDevOffering })], existing); + expect(plan.register.map((e) => e.name)).toEqual(["flux-dev"]); + expect(plan.synonymSkipped).toHaveLength(0); + }); + + it("SKIPS a synonym: same key, different name", () => { + const synonym = { ...fluxDevOffering, name: "myco-flux" }; + const plan = planDiscoverySync([parseDescriptor({ capability: synonym })], existing); + expect(plan.register).toHaveLength(0); + expect(plan.synonymSkipped).toEqual([ + { name: "myco-flux", key: "ai:t2i:flux:dev", owner: "flux-dev" }, + ]); + }); + + it("dedups within the batch too", () => { + const a = parseDescriptor({ capability: screenAgentCap }); + const b = parseDescriptor({ capability: { ...screenAgentCap, name: "screen-agent-dupe" } }); + const plan = planDiscoverySync([a, b], []); + expect(plan.register).toHaveLength(1); + expect(plan.synonymSkipped).toHaveLength(1); + }); + }); + + describe("extractDescriptors", () => { + it("pulls valid descriptors, ignores legacy entries, reports invalid", () => { + const discovery = { + runners: [ + { app: "legacy/app", version: "1", price_info: {} }, // no capability block → ignored + { capability: screenAgentCap }, // valid + { capability: { kind: "ai", family: "x", modality: "t2i", output_kind: "image" } }, // invalid: no io/offering + ], + }; + const { valid, invalid } = extractDescriptors(discovery); + expect(valid).toHaveLength(1); + expect(valid[0].capability.offering.app).toBe("emran/screen-agent"); + expect(invalid).toHaveLength(1); + }); + + it("accepts a bare array and the orchestrators[] shape", () => { + expect(extractDescriptors([{ capability: screenAgentCap }]).valid).toHaveLength(1); + expect( + extractDescriptors({ orchestrators: [{ runners: [{ capability: screenAgentCap }] }] }).valid, + ).toHaveLength(1); + }); + + it("returns empty on junk input (fail-open)", () => { + expect(extractDescriptors(null).valid).toHaveLength(0); + expect(extractDescriptors("nope").valid).toHaveLength(0); + }); + }); +});