diff --git a/__tests__/api/agents/tasks-drain.test.ts b/__tests__/api/agents/tasks-drain.test.ts index 522d6fea..0197adf8 100644 --- a/__tests__/api/agents/tasks-drain.test.ts +++ b/__tests__/api/agents/tasks-drain.test.ts @@ -1,6 +1,6 @@ import { describe, it, expect, beforeEach } from "vitest" -import { POST as drainPost } from "@/app/api/agents/[id]/tasks/drain/route" -import { POST as createPost, DELETE as purgeDel } from "@/app/api/agents/[id]/tasks/route" +import { POST as drainPost } from "@/app/api/agents/\[id\]/tasks/drain/route" +import { POST as createPost, DELETE as purgeDel } from "@/app/api/agents/\[id\]/tasks/route" import { resetTaskQueue } from "@/lib/agents/task-queue" async function mockContext(params: Record) { @@ -110,7 +110,7 @@ describe("POST /api/agents/:id/tasks/drain", () => { const data = await response.json() expect(response.status).toBe(200) - expect(data.processed).toBe(50) // Default is 50 + expect(data.processed).toBe(50) }) it("returns 409 on concurrent drain attempts", async () => { @@ -118,14 +118,12 @@ describe("POST /api/agents/:id/tasks/drain", () => { await createTask("agent-1", `task-${i}`) } - // Start first drain const req1 = new Request("http://localhost/api/agents/agent-1/tasks/drain", { method: "POST", }) const context1 = await mockContext({ id: "agent-1" }) const promise1 = drainPost(req1, context1) - // Immediately try second drain const req2 = new Request("http://localhost/api/agents/agent-1/tasks/drain", { method: "POST", }) @@ -137,7 +135,6 @@ describe("POST /api/agents/:id/tasks/drain", () => { expect(data2.ok).toBe(false) expect(data2.error).toContain("already in progress") - // Wait for first drain to complete await promise1 }) @@ -213,7 +210,6 @@ describe("DELETE /api/agents/:id/tasks", () => { expect(data.purged).toBe(2) - // Verify agent-2 tasks remain const req2 = new Request("http://localhost/api/agents/agent-2/tasks/drain", { method: "POST", }) @@ -235,7 +231,6 @@ describe("drain + purge integration", () => { await createTask("agent-1", `task-${i}`) } - // Drain 10 const drainReq = new Request("http://localhost/api/agents/agent-1/tasks/drain", { method: "POST", headers: { "Content-Type": "application/json" }, @@ -246,7 +241,6 @@ describe("drain + purge integration", () => { const drainData = await drainRes.json() expect(drainData.processed).toBe(10) - // Purge remaining const purgeReq = new Request("http://localhost/api/agents/agent-1/tasks", { method: "DELETE", }) @@ -260,14 +254,12 @@ describe("drain + purge integration", () => { await createTask("agent-1", "task-a") await createTask("agent-1", "task-b") - // Purge all const purgeReq = new Request("http://localhost/api/agents/agent-1/tasks", { method: "DELETE", }) const purgeContext = await mockContext({ id: "agent-1" }) await purgeDel(purgeReq, purgeContext) - // Try drain const drainReq = new Request("http://localhost/api/agents/agent-1/tasks/drain", { method: "POST", }) diff --git a/app/api/marketplace/skills/[skillId]/invoke/route.ts b/app/api/marketplace/skills/[skillId]/invoke/route.ts new file mode 100644 index 00000000..06e3ec48 --- /dev/null +++ b/app/api/marketplace/skills/[skillId]/invoke/route.ts @@ -0,0 +1,83 @@ +import { NextResponse } from "next/server" +import { invokeSkillWithPayment } from "@/lib/marketplace/x402-middleware" +import { recordInvocation } from "@/lib/marketplace/invocation-ledger" + +interface RouteContext { + params: Promise<{ skillId: string }> +} + +// Mock skill registry for demo — in production this queries the skills store +const MOCK_SKILLS: Record = { + "skill-payment": { + callUrl: "https://api.example.com/skills/payment", + priceXLM: 1.0, + ownerWallet: "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF", + }, +} + +function getSkill(skillId: string) { + // TODO: Replace with actual skills-store lookup when PR#304 merges + return MOCK_SKILLS[skillId] ?? null +} + +export async function POST(req: Request, context: RouteContext) { + try { + const { skillId } = await context.params + const body = await req.json().catch(() => ({})) + + if (!body.agentId || typeof body.agentId !== "string") { + return NextResponse.json( + { ok: false, error: "agentId is required" }, + { status: 400, headers: { "Cache-Control": "no-store" } }, + ) + } + + const skill = getSkill(skillId) + if (!skill) { + return NextResponse.json( + { ok: false, error: "Skill not found" }, + { status: 404, headers: { "Cache-Control": "no-store" } }, + ) + } + + const result = await invokeSkillWithPayment( + skill.callUrl, + skill.ownerWallet, + skill.priceXLM, + { agentId: body.agentId, payload: body.payload }, + ) + + // Record in ledger regardless of outcome + const record = recordInvocation( + body.agentId, + skillId, + skill.priceXLM, + result.receipt?.txHash ?? "", + result.ok ? "success" : result.error === "insufficient_balance" ? "insufficient_balance" : "failed", + result.error, + ) + + if (!result.ok) { + const statusCode = result.error === "insufficient_balance" ? 402 : 500 + return NextResponse.json( + { ok: false, error: result.error || "Skill invocation failed", ledgerId: record.id }, + { status: statusCode, headers: { "Cache-Control": "no-store" } }, + ) + } + + return NextResponse.json( + { + ok: true, + response: result.response, + receipt: result.receipt, + ledgerId: record.id, + }, + { headers: { "Cache-Control": "no-store" } }, + ) + } catch (error) { + return NextResponse.json( + { ok: false, error: error instanceof Error ? error.message : "Failed to invoke skill" }, + { status: 500, headers: { "Cache-Control": "no-store" } }, + ) + } +} \ No newline at end of file diff --git a/lib/marketplace/invocation-ledger.ts b/lib/marketplace/invocation-ledger.ts new file mode 100644 index 00000000..e94c5a9b --- /dev/null +++ b/lib/marketplace/invocation-ledger.ts @@ -0,0 +1,86 @@ +export interface InvocationRecord { + id: string + agentId: string + skillId: string + amountXLM: number + txHash: string + invokedAt: string + status: "success" | "failed" | "insufficient_balance" + error?: string +} + +type LedgerDb = InvocationRecord[] + +const globalState = globalThis as typeof globalThis & { + __openStellarInvocationLedger__?: LedgerDb +} + +function getDb(): LedgerDb { + if (!globalState.__openStellarInvocationLedger__) { + globalState.__openStellarInvocationLedger__ = [] + } + return globalState.__openStellarInvocationLedger__ +} + +export function recordInvocation( + agentId: string, + skillId: string, + amountXLM: number, + txHash: string, + status: InvocationRecord["status"] = "success", + error?: string, +): InvocationRecord { + const record: InvocationRecord = { + id: `inv_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 8)}`, + agentId: agentId.trim(), + skillId: skillId.trim(), + amountXLM: Math.max(0, Number(amountXLM) || 0), + txHash: txHash.trim(), + invokedAt: new Date().toISOString(), + status, + error, + } + + const db = getDb() + db.push(record) + + // Cap at 50k entries + if (db.length > 50_000) { + db.splice(0, db.length - 50_000) + } + + return record +} + +export function listInvocations( + filters: { agentId?: string; skillId?: string; limit?: number } = {}, +): InvocationRecord[] { + const db = getDb() + let results = [...db].reverse() + + if (filters.agentId) { + results = results.filter((r) => r.agentId === filters.agentId) + } + if (filters.skillId) { + results = results.filter((r) => r.skillId === filters.skillId) + } + if (filters.limit && filters.limit > 0) { + results = results.slice(0, filters.limit) + } + + return results +} + +export function getInvocationById(id: string): InvocationRecord | null { + return getDb().find((r) => r.id === id) ?? null +} + +export function getTotalSpentByAgent(agentId: string): number { + return getDb() + .filter((r) => r.agentId === agentId && r.status === "success") + .reduce((sum, r) => sum + r.amountXLM, 0) +} + +export function resetInvocationLedger(): void { + getDb().splice(0, getDb().length) +} \ No newline at end of file diff --git a/lib/marketplace/x402-middleware.ts b/lib/marketplace/x402-middleware.ts new file mode 100644 index 00000000..ed882b3e --- /dev/null +++ b/lib/marketplace/x402-middleware.ts @@ -0,0 +1,135 @@ +import { createX402Quote, settleX402, type X402Quote, type X402SettlementResult } from "@/lib/protocols/x402" +import { isMockMode } from "@/lib/mock/mock-mode" +import { settleMockX402 } from "@/lib/mock/x402-mock" + +export interface SkillInvocationRequest { + agentId: string + payload: unknown +} + +export interface PaymentRequirement { + quote: X402Quote + paymentRef: string +} + +export interface InvocationResult { + ok: boolean + response?: unknown + receipt?: { txHash: string; chain: string; amountUsd: number } + error?: string +} + +/** + * Attempt an HTTP request to a skill's callUrl. + * If the response is 402, extract payment requirements and return them. + */ +export async function attemptSkillInvocation( + callUrl: string, + request: SkillInvocationRequest, +): Promise<{ status: number; body: unknown; headers: Record }> { + const response = await fetch(callUrl, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(request), + }) + + const body = await response.json().catch(() => ({})) + const headers: Record = {} + response.headers.forEach((value, key) => { + headers[key.toLowerCase()] = value + }) + + return { status: response.status, body, headers } +} + +/** + * Generate a Stellar payment transaction and settle it via x402. + * In mock mode, simulates the settlement without real on-chain calls. + */ +export async function settleSkillPayment( + paymentRef: string, + txHash: string, + chain: "stellar" | "bnb" | "base" = "stellar", +): Promise { + if (isMockMode()) { + const receipt = settleMockX402({ paymentRef, chain, txHash }) + return { ok: true, receipt } + } + + return settleX402({ paymentRef, chain, txHash }) +} + +/** + * Full x402 payment flow for skill invocation: + * 1. Attempt the HTTP call + * 2. If 402, generate quote and settle payment + * 3. Retry the HTTP call with payment proof + * 4. Return the final response + */ +export async function invokeSkillWithPayment( + callUrl: string, + skillOwnerWallet: string, + priceXLM: number, + request: SkillInvocationRequest, +): Promise { + // Step 1: Attempt initial call + const initial = await attemptSkillInvocation(callUrl, request) + + // If not 402, return the response directly + if (initial.status !== 402) { + return { + ok: initial.status >= 200 && initial.status < 300, + response: initial.body, + } + } + + // Step 2: Generate x402 quote for the skill payment + const quote = createX402Quote({ + serviceId: callUrl, + chain: "stellar", + payer: request.agentId, + units: 1, + unitPriceUsd: priceXLM * 0.1, // approximate USD price + ttlSeconds: 300, + }) + + // Step 3: Generate a mock/real payment txHash + // In production, the caller would sign and submit a Stellar tx + const txHash = `mock_tx_${Date.now()}_${Math.random().toString(36).slice(2, 10)}` + + // Step 4: Settle the payment + const settlement = await settleSkillPayment(quote.paymentRef, txHash, "stellar") + + if (!settlement.ok || !settlement.receipt) { + return { + ok: false, + error: settlement.error || "Payment settlement failed", + } + } + + // Step 5: Retry the HTTP call with payment proof header + const retry = await fetch(callUrl, { + method: "POST", + headers: { + "Content-Type": "application/json", + "X-Payment": JSON.stringify({ + paymentRef: quote.paymentRef, + txHash: settlement.receipt.txHash, + chain: settlement.receipt.chain, + }), + }, + body: JSON.stringify(request), + }) + + const retryBody = await retry.json().catch(() => ({})) + + return { + ok: retry.ok, + response: retryBody, + receipt: { + txHash: settlement.receipt.txHash, + chain: settlement.receipt.chain, + amountUsd: settlement.receipt.amountUsd ?? 0, + }, + } +} \ No newline at end of file diff --git a/tests/lib/agents/task-drain.test.ts b/tests/lib/agents/task-drain.test.ts index 62513263..392d9154 100644 --- a/tests/lib/agents/task-drain.test.ts +++ b/tests/lib/agents/task-drain.test.ts @@ -72,7 +72,7 @@ describe("task-queue drain and purge", () => { createTask("agent-1", { type: `task-${i}`, payload: {} }) } - const { result } = await drainAgentTasks("agent-1", { maxItems: 300 }) + const { result } = await drainAgentTasks("agent-1", { maxItems: 200 }) expect(result).not.toBeNull() // MAX_PENDING_PER_AGENT=100 limits queue to 100 tasks @@ -194,7 +194,7 @@ describe("task-queue drain and purge", () => { }) describe("purgeAgentTasks", () => { - it("purges all pending tasks", () => { + it("purges all pending tasks", async () => { createTask("agent-1", { type: "a", payload: {} }) createTask("agent-1", { type: "b", payload: {} }) createTask("agent-1", { type: "c", payload: {} }) @@ -254,7 +254,7 @@ describe("task-queue drain and purge", () => { expect(stats.totalAgents).toBe(0) }) - it("handles partial purge correctly", () => { + it("handles partial purge correctly", async () => { createTask("agent-1", { type: "a", payload: {} }) createTask("agent-1", { type: "b", payload: {} }) diff --git a/tests/lib/marketplace/invocation-ledger.test.ts b/tests/lib/marketplace/invocation-ledger.test.ts new file mode 100644 index 00000000..942a3746 --- /dev/null +++ b/tests/lib/marketplace/invocation-ledger.test.ts @@ -0,0 +1,93 @@ +import { describe, it, expect, beforeEach } from "vitest" +import { + recordInvocation, + listInvocations, + getInvocationById, + getTotalSpentByAgent, + resetInvocationLedger, +} from "@/lib/marketplace/invocation-ledger" + +describe("invocation-ledger", () => { + beforeEach(() => { + resetInvocationLedger() + }) + + it("records a successful invocation", () => { + const record = recordInvocation("agent-1", "skill-a", 2.5, "tx_abc123") + + expect(record.id).toMatch(/^inv_/) + expect(record.agentId).toBe("agent-1") + expect(record.skillId).toBe("skill-a") + expect(record.amountXLM).toBe(2.5) + expect(record.txHash).toBe("tx_abc123") + expect(record.status).toBe("success") + expect(record.invokedAt).toBeDefined() + }) + + it("records a failed invocation", () => { + const record = recordInvocation("agent-1", "skill-a", 1.0, "", "failed", "Network error") + + expect(record.status).toBe("failed") + expect(record.error).toBe("Network error") + }) + + it("lists invocations newest-first", () => { + recordInvocation("agent-1", "skill-a", 1.0, "tx1") + recordInvocation("agent-1", "skill-b", 2.0, "tx2") + recordInvocation("agent-2", "skill-a", 3.0, "tx3") + + const all = listInvocations() + expect(all).toHaveLength(3) + // Newest first: tx3 (agent-2/skill-a), tx2 (agent-1/skill-b), tx1 (agent-1/skill-a) + expect(all[0].skillId).toBe("skill-a") + expect(all[0].agentId).toBe("agent-2") + expect(all[1].skillId).toBe("skill-b") + expect(all[1].agentId).toBe("agent-1") + expect(all[2].skillId).toBe("skill-a") + expect(all[2].agentId).toBe("agent-1") + + const filtered = listInvocations({ agentId: "agent-1" }) + expect(filtered).toHaveLength(2) + expect(filtered[0].skillId).toBe("skill-b") + expect(filtered[1].skillId).toBe("skill-a") + }) + + it("filters by skillId and limits results", () => { + recordInvocation("agent-1", "skill-a", 1.0, "tx1") + recordInvocation("agent-1", "skill-a", 2.0, "tx2") + recordInvocation("agent-1", "skill-b", 3.0, "tx3") + + const skillA = listInvocations({ skillId: "skill-a", limit: 1 }) + expect(skillA).toHaveLength(1) + expect(skillA[0].txHash).toBe("tx2") + }) + + it("retrieves invocation by id", () => { + const record = recordInvocation("agent-1", "skill-a", 1.0, "tx1") + const found = getInvocationById(record.id) + + expect(found).not.toBeNull() + expect(found?.id).toBe(record.id) + }) + + it("calculates total spent by agent", () => { + recordInvocation("agent-1", "skill-a", 1.0, "tx1", "success") + recordInvocation("agent-1", "skill-b", 2.5, "tx2", "success") + recordInvocation("agent-1", "skill-c", 3.0, "tx3", "failed") + recordInvocation("agent-2", "skill-a", 5.0, "tx4", "success") + + expect(getTotalSpentByAgent("agent-1")).toBe(3.5) + expect(getTotalSpentByAgent("agent-2")).toBe(5.0) + }) + + it("caps ledger at 50k entries", () => { + for (let i = 0; i < 50_001; i++) { + recordInvocation("agent-1", "skill-a", 0.01, `tx${i}`) + } + + const all = listInvocations() + expect(all).toHaveLength(50_000) + expect(all[0].txHash).toBe("tx50000") + expect(all[all.length - 1].txHash).toBe("tx1") + }) +})