diff --git a/web/src/app/api/services/route.ts b/web/src/app/api/services/route.ts index e24a0a18..8dd1201e 100644 --- a/web/src/app/api/services/route.ts +++ b/web/src/app/api/services/route.ts @@ -3,6 +3,7 @@ import { db } from "@/db"; import { tlsTalos, tlsCommerceServices } from "@/db/schema"; import { and, desc, eq, ilike, lt, ne, or } from "drizzle-orm"; import { parseLimit } from "@/lib/parse-limit"; +import { fetchReputations } from "@/lib/reputation-ledger"; import { withTraceContext } from "@/lib/tracing"; // GET /api/services — Discover available services across all TALOS agents @@ -16,77 +17,125 @@ async function handleGet(request: NextRequest) { if (!parsedLimit.ok) return parsedLimit.response; const limit = parsedLimit.limit; - const conditions = [eq(tlsTalos.status, "Active")]; + const minScore = searchParams.has("minScore") ? Number(searchParams.get("minScore")) : undefined; + const minConfidence = searchParams.has("minConfidence") ? Number(searchParams.get("minConfidence")) : undefined; + const allowColdStart = searchParams.get("allowColdStart") === "true"; - // Exclude the requesting TALOS's own services - if (selfId) { - conditions.push(ne(tlsCommerceServices.talosId, selfId)); - } + let currentCursor = cursor; + const accumulated: any[] = []; + let exhausted = false; - // Filter by TALOS category (case-insensitive match in DB) - if (category) { - conditions.push(ilike(tlsTalos.category, category)); - } + // Loop until we fulfill the limit or exhaust the DB + while (accumulated.length < limit && !exhausted) { + const conditions = [eq(tlsTalos.status, "Active")]; - // Cursor condition (createdAt DESC with id tiebreaker) - if (cursor) { - const [cursorDate, cursorId] = cursor.split("|"); - if (cursorDate && cursorId) { - conditions.push( - or( - lt(tlsCommerceServices.createdAt, new Date(cursorDate)), - and( - eq(tlsCommerceServices.createdAt, new Date(cursorDate)), - lt(tlsCommerceServices.id, cursorId), - ), - )!, - ); + // Exclude the requesting TALOS's own services + if (selfId) { + conditions.push(ne(tlsCommerceServices.talosId, selfId)); + } + + // Filter by TALOS category (case-insensitive match in DB) + if (category) { + conditions.push(ilike(tlsTalos.category, category)); } - } - const services = await db - .select({ - id: tlsCommerceServices.id, - talosId: tlsCommerceServices.talosId, - talosName: tlsTalos.name, - talosCategory: tlsTalos.category, - serviceName: tlsCommerceServices.serviceName, - description: tlsCommerceServices.description, - price: tlsCommerceServices.price, - currency: tlsCommerceServices.currency, - chains: tlsCommerceServices.chains, - createdAt: tlsCommerceServices.createdAt, - }) - .from(tlsCommerceServices) - .innerJoin(tlsTalos, eq(tlsCommerceServices.talosId, tlsTalos.id)) - .where(conditions.length > 0 ? and(...conditions) : undefined) - .orderBy(desc(tlsCommerceServices.createdAt), desc(tlsCommerceServices.id)) - .limit(limit + 1); - - const hasMore = services.length > limit; - const page = hasMore ? services.slice(0, limit) : services; - - const results = page.map((s) => ({ - talosId: s.talosId, - talosName: s.talosName, - talosCategory: s.talosCategory, - serviceName: s.serviceName, - description: s.description, - price: Number(s.price), - currency: s.currency, - chains: s.chains, - })); + // Cursor condition (createdAt DESC with id tiebreaker) + if (currentCursor) { + const [cursorDate, cursorId] = currentCursor.split("|"); + if (cursorDate && cursorId) { + conditions.push( + or( + lt(tlsCommerceServices.createdAt, new Date(cursorDate)), + and( + eq(tlsCommerceServices.createdAt, new Date(cursorDate)), + lt(tlsCommerceServices.id, cursorId), + ), + )!, + ); + } + } + + const services = await db + .select({ + id: tlsCommerceServices.id, + talosId: tlsCommerceServices.talosId, + talosName: tlsTalos.name, + talosCategory: tlsTalos.category, + serviceName: tlsCommerceServices.serviceName, + description: tlsCommerceServices.description, + price: tlsCommerceServices.price, + currency: tlsCommerceServices.currency, + chains: tlsCommerceServices.chains, + createdAt: tlsCommerceServices.createdAt, + }) + .from(tlsCommerceServices) + .innerJoin(tlsTalos, eq(tlsCommerceServices.talosId, tlsTalos.id)) + .where(conditions.length > 0 ? and(...conditions) : undefined) + .orderBy(desc(tlsCommerceServices.createdAt), desc(tlsCommerceServices.id)) + .limit(limit * 2); + + if (services.length === 0) { + exhausted = true; + break; + } + + if (services.length < limit * 2) { + exhausted = true; + } + + const talosIds = Array.from(new Set(services.map((s) => s.talosId))); + const reputations = await fetchReputations(talosIds, new Date()); + + for (const service of services) { + let valid = true; + + if (minScore !== undefined || minConfidence !== undefined || allowColdStart) { + const rep = reputations.get(service.talosId); + if (rep) { + if (rep.evidence === "insufficient") { + if (!allowColdStart) valid = false; + } else { + if (minScore !== undefined && rep.score < minScore) valid = false; + if (minConfidence !== undefined && rep.confidence < minConfidence) valid = false; + } + } else { + // Cold start + if (!allowColdStart) valid = false; + } + } + + if (valid) { + accumulated.push({ + id: service.id, + talosId: service.talosId, + talosName: service.talosName, + talosCategory: service.talosCategory, + serviceName: service.serviceName, + description: service.description, + price: Number(service.price), + currency: service.currency, + chains: service.chains, + createdAt: service.createdAt, + }); + if (accumulated.length === limit) { + currentCursor = `${service.createdAt.toISOString()}|${service.id}`; + break; + } + } + currentCursor = `${service.createdAt.toISOString()}|${service.id}`; + } + } // Shuffle for diversity — agents see different services each cycle - for (let i = results.length - 1; i > 0; i--) { + for (let i = accumulated.length - 1; i > 0; i--) { const j = Math.floor(Math.random() * (i + 1)); - [results[i], results[j]] = [results[j], results[i]]; + [accumulated[i], accumulated[j]] = [accumulated[j], accumulated[i]]; } - const lastItem = page[page.length - 1]; - const nextCursor = hasMore && lastItem - ? `${lastItem.createdAt.toISOString()}|${lastItem.id}` - : null; + const nextCursor = (exhausted && accumulated.length < limit) ? null : currentCursor; + + // Remove cursor tracking fields from the output to match original payload structure + const results = accumulated.map(({ id, createdAt, ...rest }) => rest); return Response.json({ data: results, nextCursor }); } catch { diff --git a/web/src/app/api/talos/[id]/reputation/route.ts b/web/src/app/api/talos/[id]/reputation/route.ts index c7474232..4b6a7130 100644 --- a/web/src/app/api/talos/[id]/reputation/route.ts +++ b/web/src/app/api/talos/[id]/reputation/route.ts @@ -8,6 +8,7 @@ import { REPUTATION_SCORE_VERSION, reputationInputsSchema, ReputationJobInput, + MAX_JOB_AGE_DAYS, } from "@/lib/reputation"; export const dynamic = "force-dynamic"; @@ -39,12 +40,6 @@ const querySchema = z.object({ jobLimit: z.coerce.number().int().positive().max(10_000).optional(), }); -/** - * Maximum age of a job (days) that will be considered for scoring. - * Prevents ancient activity from skewing the score and bounds query - * cost at the DB layer (one indexed scan filtered by `createdAt`). - */ -const MAX_JOB_AGE_DAYS = 365; /** * GET /api/talos/:id/reputation diff --git a/web/src/app/api/talos/route.ts b/web/src/app/api/talos/route.ts index 80b853bb..af1cc953 100644 --- a/web/src/app/api/talos/route.ts +++ b/web/src/app/api/talos/route.ts @@ -8,6 +8,7 @@ import { createAgentKeypair, fundTestnetAccount, verifyStellarSignature } from " import { createTalosSchema, parseBody } from "@/lib/schemas"; import { parseLimit } from "@/lib/parse-limit"; import { TimeoutError, withTimeout } from "@/lib/timeout"; +import { fetchReputations } from "@/lib/reputation-ledger"; // GET /api/talos — List TALOS entries with cursor-based pagination export async function GET(request: NextRequest) { @@ -18,6 +19,10 @@ export async function GET(request: NextRequest) { if (!parsedLimit.ok) return parsedLimit.response; const limit = parsedLimit.limit; + const minScore = searchParams.has("minScore") ? Number(searchParams.get("minScore")) : undefined; + const minConfidence = searchParams.has("minConfidence") ? Number(searchParams.get("minConfidence")) : undefined; + const allowColdStart = searchParams.get("allowColdStart") === "true"; + // Add timeout for patron count query const patronCountQuery = db .select({ @@ -29,86 +34,125 @@ export async function GET(request: NextRequest) { .as("patronCount"); const patronCount = patronCountQuery; + let currentCursor = cursor; + const accumulated: any[] = []; + let exhausted = false; - const conditions = []; - if (cursor) { - const [cursorDate, cursorId] = cursor.split("|"); - if (cursorDate && cursorId) { - conditions.push( - or( - lt(tlsTalos.createdAt, new Date(cursorDate)), - and( - eq(tlsTalos.createdAt, new Date(cursorDate)), - lt(tlsTalos.id, cursorId), - ), - )!, - ); + // Loop until we fulfill the limit or exhaust the DB + while (accumulated.length < limit && !exhausted) { + const conditions = []; + if (currentCursor) { + const [cursorDate, cursorId] = currentCursor.split("|"); + if (cursorDate && cursorId) { + conditions.push( + or( + lt(tlsTalos.createdAt, new Date(cursorDate)), + and( + eq(tlsTalos.createdAt, new Date(cursorDate)), + lt(tlsTalos.id, cursorId), + ), + )!, + ); + } } - } - let entries; - try { - entries = await withTimeout( - db.select({ - id: tlsTalos.id, - onChainId: tlsTalos.onChainId, - agentName: tlsTalos.agentName, - name: tlsTalos.name, - category: tlsTalos.category, - description: tlsTalos.description, - status: tlsTalos.status, - stellarAssetCode: tlsTalos.stellarAssetCode, - pulsePrice: tlsTalos.pulsePrice, - totalSupply: tlsTalos.totalSupply, - creatorShare: tlsTalos.creatorShare, - investorShare: tlsTalos.investorShare, - treasuryShare: tlsTalos.treasuryShare, - persona: tlsTalos.persona, - targetAudience: tlsTalos.targetAudience, - channels: tlsTalos.channels, - toneVoice: tlsTalos.toneVoice, - approvalThreshold: tlsTalos.approvalThreshold, - gtmBudget: tlsTalos.gtmBudget, - minPatronPulse: tlsTalos.minPatronPulse, - agentOnline: tlsTalos.agentOnline, - agentLastSeen: tlsTalos.agentLastSeen, - walletPublicKey: tlsTalos.walletPublicKey, - creatorPublicKey: tlsTalos.creatorPublicKey, - investorPublicKey: tlsTalos.investorPublicKey, - treasuryPublicKey: tlsTalos.treasuryPublicKey, - createdAt: tlsTalos.createdAt, - updatedAt: tlsTalos.updatedAt, - patrons: patronCount.count, - }) - .from(tlsTalos) - .leftJoin(patronCount, eq(tlsTalos.id, patronCount.talosId)) - .where(conditions.length > 0 ? and(...conditions) : undefined) - .orderBy(desc(tlsTalos.createdAt), desc(tlsTalos.id)) - .limit(limit + 1), - 10_000, - "Talos list query timeout", - ); - } catch (error) { - if (error instanceof TimeoutError) { - return Response.json( - { error: "Query timeout. Please try again with a simpler query.", details: error.message }, - { status: 408 }, + let entries; + try { + entries = await withTimeout( + db.select({ + id: tlsTalos.id, + onChainId: tlsTalos.onChainId, + agentName: tlsTalos.agentName, + name: tlsTalos.name, + category: tlsTalos.category, + description: tlsTalos.description, + status: tlsTalos.status, + stellarAssetCode: tlsTalos.stellarAssetCode, + pulsePrice: tlsTalos.pulsePrice, + totalSupply: tlsTalos.totalSupply, + creatorShare: tlsTalos.creatorShare, + investorShare: tlsTalos.investorShare, + treasuryShare: tlsTalos.treasuryShare, + persona: tlsTalos.persona, + targetAudience: tlsTalos.targetAudience, + channels: tlsTalos.channels, + toneVoice: tlsTalos.toneVoice, + approvalThreshold: tlsTalos.approvalThreshold, + gtmBudget: tlsTalos.gtmBudget, + minPatronPulse: tlsTalos.minPatronPulse, + agentOnline: tlsTalos.agentOnline, + agentLastSeen: tlsTalos.agentLastSeen, + walletPublicKey: tlsTalos.walletPublicKey, + creatorPublicKey: tlsTalos.creatorPublicKey, + investorPublicKey: tlsTalos.investorPublicKey, + treasuryPublicKey: tlsTalos.treasuryPublicKey, + createdAt: tlsTalos.createdAt, + updatedAt: tlsTalos.updatedAt, + patrons: patronCount.count, + }) + .from(tlsTalos) + .leftJoin(patronCount, eq(tlsTalos.id, patronCount.talosId)) + .where(conditions.length > 0 ? and(...conditions) : undefined) + .orderBy(desc(tlsTalos.createdAt), desc(tlsTalos.id)) + .limit(limit * 2), // fetch chunk + 10_000, + "Talos list query timeout", ); + } catch (error) { + if (error instanceof TimeoutError) { + return Response.json( + { error: "Query timeout. Please try again with a simpler query.", details: error.message }, + { status: 408 }, + ); + } + console.error("Talos list query error:", error); + return Response.json({ error: "Internal server error" }, { status: 500 }); + } + + if (entries.length === 0) { + exhausted = true; + break; + } + + if (entries.length < limit * 2) { + exhausted = true; } - console.error("Talos list query error:", error); - return Response.json({ error: "Internal server error" }, { status: 500 }); - } - const hasMore = entries.length > limit; - const page = hasMore ? entries.slice(0, limit) : entries; - const data = page.map((c) => ({ ...c, patrons: c.patrons ?? 0 })); + const talosIds = entries.map((e) => e.id); + const reputations = await fetchReputations(talosIds, new Date()); + + for (const entry of entries) { + let valid = true; + + if (minScore !== undefined || minConfidence !== undefined || allowColdStart) { + const rep = reputations.get(entry.id); + if (rep) { + if (rep.evidence === "insufficient") { + if (!allowColdStart) valid = false; + } else { + if (minScore !== undefined && rep.score < minScore) valid = false; + if (minConfidence !== undefined && rep.confidence < minConfidence) valid = false; + } + } else { + // Cold start + if (!allowColdStart) valid = false; + } + } + + if (valid) { + accumulated.push({ ...entry, patrons: entry.patrons ?? 0 }); + if (accumulated.length === limit) { + currentCursor = `${entry.createdAt.toISOString()}|${entry.id}`; + break; + } + } + currentCursor = `${entry.createdAt.toISOString()}|${entry.id}`; + } + } - const lastItem = page[page.length - 1]; - const nextCursor = hasMore && lastItem - ? `${lastItem.createdAt.toISOString()}|${lastItem.id}` - : null; + const nextCursor = (exhausted && accumulated.length < limit) ? null : currentCursor; - return Response.json({ data, nextCursor }); + return Response.json({ data: accumulated, nextCursor }); } catch { return internalError(request); } diff --git a/web/src/lib/openapi.ts b/web/src/lib/openapi.ts index abaeed37..e63c9c08 100644 --- a/web/src/lib/openapi.ts +++ b/web/src/lib/openapi.ts @@ -1063,6 +1063,24 @@ Inter-agent commerce uses the Stellar x402 payment protocol: schema: { type: "integer", minimum: 1, maximum: 100, default: 50 }, description: "Max items per page (1–100)", }, + minScoreParam: { + name: "minScore", + in: "query", + schema: { type: "number", minimum: 0, maximum: 100 }, + description: "Planner policy: Minimum reputation score (0-100) required to include in the results.", + }, + minConfidenceParam: { + name: "minConfidence", + in: "query", + schema: { type: "number", minimum: 0, maximum: 1 }, + description: "Planner policy: Minimum reputation confidence (0.0-1.0) required to include in the results.", + }, + allowColdStartParam: { + name: "allowColdStart", + in: "query", + schema: { type: "boolean", default: false }, + description: "Planner policy: Include cold-start providers with 'insufficient' evidence even if they don't meet minScore/minConfidence.", + }, }, headers: { ApiVersion: { @@ -1167,6 +1185,9 @@ Inter-agent commerce uses the Stellar x402 payment protocol: parameters: [ { $ref: "#/components/parameters/cursorParam" }, { $ref: "#/components/parameters/limitParam" }, + { $ref: "#/components/parameters/minScoreParam" }, + { $ref: "#/components/parameters/minConfidenceParam" }, + { $ref: "#/components/parameters/allowColdStartParam" }, ], responses: { "200": { @@ -2390,6 +2411,9 @@ Use this for multi-chain payment completion flows that should trigger fulfillmen }, { $ref: "#/components/parameters/cursorParam" }, { $ref: "#/components/parameters/limitParam" }, + { $ref: "#/components/parameters/minScoreParam" }, + { $ref: "#/components/parameters/minConfidenceParam" }, + { $ref: "#/components/parameters/allowColdStartParam" }, ], responses: { "200": { diff --git a/web/src/lib/reputation-ledger.ts b/web/src/lib/reputation-ledger.ts index a9b4d022..14c241f4 100644 --- a/web/src/lib/reputation-ledger.ts +++ b/web/src/lib/reputation-ledger.ts @@ -1,6 +1,7 @@ import { db } from "@/db"; import { tlsCommerceJobs, tlsReputationInputs } from "@/db/schema"; -import { eq } from "drizzle-orm"; +import { eq, inArray, and, sql } from "drizzle-orm"; +import { computeReputation, MAX_JOB_AGE_DAYS, ReputationJobInput, ReputationScore, reputationInputsSchema } from "./reputation"; import type { PgTransaction } from "drizzle-orm/pg-core"; import type { ExtractTablesWithRelations } from "drizzle-orm"; import type { PostgresJsQueryResultHKT } from "drizzle-orm/postgres-js"; @@ -126,3 +127,62 @@ export async function rebuildReputationLedger(providerId?: string) { return { ingestedCount }; } + +/** + * Bulk fetch and compute reputations for a list of providers. + */ +export async function fetchReputations( + talosIds: string[], + now: Date +): Promise> { + const result = new Map(); + if (talosIds.length === 0) return result; + + const cutoff = new Date( + now.getTime() - MAX_JOB_AGE_DAYS * 24 * 60 * 60 * 1000 + ); + + const jobRows = await db + .select({ + talosId: tlsReputationInputs.talosId, + id: tlsReputationInputs.jobId, + status: tlsReputationInputs.status, + requesterTalosId: tlsReputationInputs.requesterTalosId, + createdAt: tlsReputationInputs.jobCreatedAt, + updatedAt: tlsReputationInputs.jobUpdatedAt, + hasResult: tlsReputationInputs.hasResult, + }) + .from(tlsReputationInputs) + .where( + and( + inArray(tlsReputationInputs.talosId, talosIds), + sql`${tlsReputationInputs.jobCreatedAt} >= ${cutoff}` + ) + ); + + const grouped = new Map(); + for (const row of jobRows) { + if (!grouped.has(row.talosId)) { + grouped.set(row.talosId, []); + } + grouped.get(row.talosId)!.push({ + id: row.id, + status: row.status ?? "unknown", + requesterTalosId: row.requesterTalosId, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + hasResult: row.hasResult, + }); + } + + for (const providerId of talosIds) { + const jobs = grouped.get(providerId) ?? []; + const inputs = reputationInputsSchema.parse({ + providerId, + jobs, + }); + result.set(providerId, computeReputation(inputs, { now })); + } + + return result; +} diff --git a/web/src/lib/reputation.ts b/web/src/lib/reputation.ts index 9aefad0d..ac76a420 100644 --- a/web/src/lib/reputation.ts +++ b/web/src/lib/reputation.ts @@ -32,6 +32,9 @@ export const REPUTATION_SCORE_VERSION = "1.0.0" as const; /** Half-life for the exponential decay weighting of past jobs (days). */ export const REPUTATION_HALF_LIFE_DAYS = 30; +/** Maximum age of a job (days) that will be considered for scoring. */ +export const MAX_JOB_AGE_DAYS = 365; + /** Latency budget considered "on-time" for fulfillment latency signal. */ export const ON_TIME_BUDGET_HOURS = 24; diff --git a/web/tests/fixtures/openapi.snapshot.json b/web/tests/fixtures/openapi.snapshot.json index 70b9432f..f34c3980 100644 --- a/web/tests/fixtures/openapi.snapshot.json +++ b/web/tests/fixtures/openapi.snapshot.json @@ -2367,6 +2367,35 @@ "default": 50 }, "description": "Max items per page (1–100)" + }, + "minScoreParam": { + "name": "minScore", + "in": "query", + "schema": { + "type": "number", + "minimum": 0, + "maximum": 100 + }, + "description": "Planner policy: Minimum reputation score (0-100) required to include in the results." + }, + "minConfidenceParam": { + "name": "minConfidence", + "in": "query", + "schema": { + "type": "number", + "minimum": 0, + "maximum": 1 + }, + "description": "Planner policy: Minimum reputation confidence (0.0-1.0) required to include in the results." + }, + "allowColdStartParam": { + "name": "allowColdStart", + "in": "query", + "schema": { + "type": "boolean", + "default": false + }, + "description": "Planner policy: Include cold-start providers with 'insufficient' evidence even if they don't meet minScore/minConfidence." } }, "headers": { @@ -2530,6 +2559,15 @@ }, { "$ref": "#/components/parameters/limitParam" + }, + { + "$ref": "#/components/parameters/minScoreParam" + }, + { + "$ref": "#/components/parameters/minConfidenceParam" + }, + { + "$ref": "#/components/parameters/allowColdStartParam" } ], "responses": { @@ -4373,6 +4411,15 @@ }, { "$ref": "#/components/parameters/limitParam" + }, + { + "$ref": "#/components/parameters/minScoreParam" + }, + { + "$ref": "#/components/parameters/minConfidenceParam" + }, + { + "$ref": "#/components/parameters/allowColdStartParam" } ], "responses": {