From e54eb55f2b147da0f475049182428cc0372fddb3 Mon Sep 17 00:00:00 2001 From: david Date: Thu, 27 Aug 2026 09:54:53 +0200 Subject: [PATCH] feat: add balance-aware provider resale agent --- buyer/src/sdk/Marketplace.ts | 135 ++++- buyer/src/sdk/types.ts | 15 + deploy/mainnet/deploy.sh | 2 +- ...r-compose.supplier-openrouter-reseller.yml | 79 +++ .../mainnet/openrouter-reseller.config.json | 21 + gateway/package.json | 3 +- gateway/src/cli/reseller-demo.ts | 197 +++++++ gateway/src/openai/chatCompletions.ts | 4 + gateway/src/openai/errors.ts | 16 + gateway/src/openai/validate.ts | 11 +- gateway/src/server.ts | 1 + packages/shared/src/tx/advert/postAdvert.ts | 15 +- packages/shared/src/tx/escrow/submit.ts | 4 + packages/shared/src/tx/index.ts | 4 + packages/shared/src/tx/inputBound.ts | 26 + packages/shared/src/tx/internal/liveCbor.ts | 15 +- pnpm-lock.yaml | 116 ++++ supplier/package.json | 7 +- supplier/src/cli/postAdvertFlow.ts | 4 + supplier/src/cli/setup-reseller.ts | 494 ++++++++++++++++++ supplier/src/config.ts | 153 ++++++ supplier/src/index.ts | 78 ++- supplier/src/jobRunner.ts | 116 +++- supplier/src/openai.ts | 11 + supplier/src/reseller/capacityGate.ts | 179 +++++++ supplier/src/reseller/dashboardPage.ts | 83 +++ supplier/src/reseller/evidenceStore.ts | 446 ++++++++++++++++ supplier/src/reseller/money.ts | 54 ++ supplier/src/reseller/openRouterCapacity.ts | 189 +++++++ supplier/src/reseller/provider.ts | 22 + supplier/src/reseller/runtime.ts | 331 ++++++++++++ supplier/src/reseller/setupConfig.ts | 176 +++++++ supplier/src/server.ts | 155 +++++- tests/fixtures/supplier-side/sample-config.ts | 1 + .../buyer-sdk-submit-prompt-reject.test.ts | 38 ++ tests/unit/reseller-capacity.test.ts | 151 ++++++ tests/unit/reseller-redaction.test.ts | 38 ++ tests/unit/reseller-setup-config.test.ts | 68 +++ tests/unit/supplier-config.test.ts | 59 +++ tests/unit/supplier-reseller-routes.test.ts | 78 +++ tests/unit/tx-submit.test.ts | 22 + 41 files changed, 3579 insertions(+), 38 deletions(-) create mode 100644 deploy/mainnet/docker-compose.supplier-openrouter-reseller.yml create mode 100644 deploy/mainnet/openrouter-reseller.config.json create mode 100644 gateway/src/cli/reseller-demo.ts create mode 100644 packages/shared/src/tx/inputBound.ts create mode 100644 supplier/src/cli/setup-reseller.ts create mode 100644 supplier/src/reseller/capacityGate.ts create mode 100644 supplier/src/reseller/dashboardPage.ts create mode 100644 supplier/src/reseller/evidenceStore.ts create mode 100644 supplier/src/reseller/money.ts create mode 100644 supplier/src/reseller/openRouterCapacity.ts create mode 100644 supplier/src/reseller/provider.ts create mode 100644 supplier/src/reseller/runtime.ts create mode 100644 supplier/src/reseller/setupConfig.ts create mode 100644 tests/unit/reseller-capacity.test.ts create mode 100644 tests/unit/reseller-redaction.test.ts create mode 100644 tests/unit/reseller-setup-config.test.ts create mode 100644 tests/unit/supplier-reseller-routes.test.ts diff --git a/buyer/src/sdk/Marketplace.ts b/buyer/src/sdk/Marketplace.ts index 727c429..e124c76 100644 --- a/buyer/src/sdk/Marketplace.ts +++ b/buyer/src/sdk/Marketplace.ts @@ -80,6 +80,8 @@ import { buildAcceptTx, buildReclaimTx, TxConstructionError, + chatInputTokenUpperBound, + BOUNDED_INPUT_DETAIL_MARKER, } from "@marketplace/shared/tx"; import { decodeAdvertDatum, decodeEscrowDatum, canonicalize } from "@marketplace/shared/cbor"; import type { AdvertDatum } from "@marketplace/shared/cbor"; @@ -88,6 +90,7 @@ import type { DiscoverSuppliersOptions, SubmitPromptOptions, SubmitPromptResult, + SupplierCapabilityView, SubmitTtsOptions, SubmitTtsResult, SubmitOcrOptions, @@ -110,6 +113,50 @@ import { MemoryTaskHistoryStore } from "./history.js"; import { HttpClient, HttpError } from "./httpClient.js"; const ZERO_SIGNATURE = "0".repeat(128); + +function validateSupplierCapability( + body: unknown, + advert: AdvertDatum, + advertRef: OutputReference, +): SupplierCapabilityView { + if (typeof body !== "object" || body === null || Array.isArray(body)) { + throw new TxConstructionError( + "supplier_preflight_failed", + "supplier /capability response must be an object", + ); + } + const capability = body as Record; + const expectedRef = `${advertRef.txHash}#${advertRef.index}`; + if ( + capability.advert_ref !== expectedRef || + capability.capability_id !== advert.capability_id || + capability.model !== advert.model || + capability.max_output_tokens !== advert.max_output_tokens || + capability.max_processing_ms !== advert.max_processing_ms || + capability.price_lovelace !== advert.price_lovelace.toString() || + capability.supplier_pkh !== advert.supplier_pkh || + typeof capability.pub_key_hex !== "string" + ) { + throw new TxConstructionError( + "supplier_preflight_failed", + "supplier /capability response does not match the on-chain advert", + ); + } + if ( + capability.max_input_tokens !== undefined && + ( + typeof capability.max_input_tokens !== "number" || + !Number.isSafeInteger(capability.max_input_tokens) || + capability.max_input_tokens <= 0 + ) + ) { + throw new TxConstructionError( + "supplier_preflight_failed", + "supplier max_input_tokens must be a positive safe integer", + ); + } + return capability as unknown as SupplierCapabilityView; +} const HEX64_RE = /^[0-9a-fA-F]{64}$/; const SIG_RE = /^[0-9a-fA-F]{128}$/; @@ -241,16 +288,23 @@ export class Marketplace extends EventEmitterBase { } return result.body as SupplierView[]; } - // ─── submitPrompt — happy path + every adversarial branch ────────────── async submitPrompt(opts: SubmitPromptOptions): Promise { - const { advertRef, messages, payment_lovelace, max_output_tokens } = opts; + const { + advertRef, + messages, + payment_lovelace, + max_output_tokens, + public_preview, + } = opts; + let supplierHttp: HttpClient | null = null; let advertDatum: AdvertDatum | null = null; let escrowOutputRef: OutputReference | null = null; let escrowRefStr = ""; let postedAtMs = Date.now(); + let deliverByMs = 0; const recordFailure = (reason: string): void => { this.historyStore.save({ @@ -279,6 +333,46 @@ export class Marketplace extends EventEmitterBase { } } + if (advertDatum) { + supplierHttp = new HttpClient({ + baseUrl: advertDatum.endpoint_url, + fetch: this.fetchImpl, + }); + if (advertDatum.detail_uri.endsWith(BOUNDED_INPUT_DETAIL_MARKER)) { + let capabilityResult; + try { + capabilityResult = await supplierHttp.getJson("/capability"); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw new TxConstructionError("supplier_preflight_failed", message); + } + if (!capabilityResult.ok || capabilityResult.parseError) { + throw new TxConstructionError( + "supplier_preflight_failed", + `supplier /capability returned HTTP ${capabilityResult.status}`, + ); + } + const capability = validateSupplierCapability( + capabilityResult.body, + advertDatum, + advertRef, + ); + if (capability.max_input_tokens === undefined) { + throw new TxConstructionError( + "supplier_preflight_failed", + "bounded-input supplier omitted max_input_tokens", + ); + } + const inputUnits = chatInputTokenUpperBound(messages); + if (inputUnits > capability.max_input_tokens) { + throw new TxConstructionError( + "input_cap_exceeded", + `input bound ${inputUnits} exceeds supplier cap ${capability.max_input_tokens}`, + ); + } + } + } + escrowResult = await buildPostEscrowTx({ chain: this.chain, buyerKey: this.walletKey, @@ -321,24 +415,33 @@ export class Marketplace extends EventEmitterBase { if (escrowUtxo && escrowUtxo.datumHex) { const ed = decodeEscrowDatum(escrowUtxo.datumHex); postedAtMs = ed.posted_at; + deliverByMs = ed.deliver_by; } } catch { /* posted_at is best-effort metadata */ } if (!advertDatum) { - // Should never happen — buildPostEscrowTx throws before this on missing/bad advert. + // Should never happen — buildPostEscrowTx rejects missing or bad adverts. const reason = "advert datum unavailable after escrow post"; recordFailure(reason); throw new ReceiptVerificationError(reason); } + if (!supplierHttp) { + const reason = "supplier client unavailable after escrow post"; + recordFailure(reason); + throw new ReceiptVerificationError(reason); + } + if (deliverByMs === 0) { + deliverByMs = + postedAtMs + advertDatum.max_processing_ms + 30_000; + } + const supplierCallTimeoutMs = Math.max( + 30_000, + deliverByMs - Date.now() + 90_000, + ); // ── 4. Call supplier /v1/chat/completions ───────────────────────── - const supplierBaseUrl = advertDatum.endpoint_url; - const supplierHttp = new HttpClient({ - baseUrl: supplierBaseUrl, - fetch: this.fetchImpl, - }); const requestedMax = max_output_tokens ?? advertDatum.max_output_tokens; const cappedMax = Math.min(requestedMax, advertDatum.max_output_tokens); @@ -347,11 +450,14 @@ export class Marketplace extends EventEmitterBase { messages, max_tokens: cappedMax, }; - let chatResult; try { chatResult = await supplierHttp.postJson("/v1/chat/completions", chatBody, { - headers: { "X-Escrow-Ref": escrowRefStr }, + headers: { + "X-Escrow-Ref": escrowRefStr, + ...(public_preview ? { "X-Vector-Public-Preview": "1" } : {}), + }, + timeoutMs: supplierCallTimeoutMs, }); } catch (err) { if (err instanceof HttpError) { @@ -385,8 +491,11 @@ export class Marketplace extends EventEmitterBase { throw sErr; } const POLL_INTERVAL_MS = 2_000; - const POLL_TIMEOUT_MS = 180_000; - const pollDeadline = Date.now() + POLL_TIMEOUT_MS; + const pollBudgetMs = Math.max( + 30_000, + deliverByMs + 90_000 - Date.now(), + ); + const pollDeadline = Date.now() + pollBudgetMs; let polled = chatResult; // First poll runs immediately so a quickly-completed job doesn't pay // the leading delay; subsequent polls wait POLL_INTERVAL_MS. @@ -423,7 +532,7 @@ export class Marketplace extends EventEmitterBase { if (polled.status !== 200) { const sErr = new SupplierError("timeout", { status: polled.status, - message: `supplier job ${jobId} not done within ${POLL_TIMEOUT_MS}ms`, + message: `supplier job ${jobId} not done within ${pollBudgetMs}ms`, }); recordFailure("timeout"); throw sErr; diff --git a/buyer/src/sdk/types.ts b/buyer/src/sdk/types.ts index a8c00bd..0407f0d 100644 --- a/buyer/src/sdk/types.ts +++ b/buyer/src/sdk/types.ts @@ -138,6 +138,19 @@ export interface SupplierView { created_slot: number; } +/** Live off-chain capability metadata published by a supplier endpoint. */ +export interface SupplierCapabilityView { + capability_id: string; + model: string; + max_output_tokens: number; + max_processing_ms: number; + price_lovelace: string; + advert_ref: string; + supplier_pkh: string; + pub_key_hex: string; + max_input_tokens?: number; +} + /** * DiscoverSuppliersOptions — optional filters for discoverSuppliers(). */ @@ -154,6 +167,8 @@ export interface SubmitPromptOptions { messages: import("@marketplace/shared/tx").ChatMessage[]; payment_lovelace: bigint; max_output_tokens?: number; + /** Explicit consent for the seller's public redacted demo preview. */ + public_preview?: boolean; } /** diff --git a/deploy/mainnet/deploy.sh b/deploy/mainnet/deploy.sh index 3ddf5b7..fc1e7ed 100755 --- a/deploy/mainnet/deploy.sh +++ b/deploy/mainnet/deploy.sh @@ -128,7 +128,7 @@ done drain_supplier() { # drain_supplier [ "$DRAIN_TIMEOUT_SECS" = "0" ] && return 0 local cid deadline status - cid=$(docker compose -f "$COMPOSE_DIR/$1" ps -q | head -1) + cid=$(docker compose -f "$COMPOSE_DIR/$1" ps -q supplier) [ -n "$cid" ] || return 0 deadline=$(( $(date +%s) + DRAIN_TIMEOUT_SECS )) while [ "$(date +%s)" -lt "$deadline" ]; do diff --git a/deploy/mainnet/docker-compose.supplier-openrouter-reseller.yml b/deploy/mainnet/docker-compose.supplier-openrouter-reseller.yml new file mode 100644 index 0000000..95d7868 --- /dev/null +++ b/deploy/mainnet/docker-compose.supplier-openrouter-reseller.yml @@ -0,0 +1,79 @@ +name: marketplace-mainnet-supplier-openrouter-reseller + +services: + postgres: + image: postgres:17-alpine + container_name: marketplace-mainnet-openrouter-reseller-postgres + restart: unless-stopped + env_file: + - path: /var/lib/agents-marketplace/openrouter-reseller-postgres.env + required: true + healthcheck: + test: ["CMD-SHELL", "pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB"] + interval: 5s + timeout: 5s + retries: 20 + start_period: 10s + volumes: + - openrouter-reseller-postgres:/var/lib/postgresql/data + networks: + - reseller-db + logging: + driver: json-file + options: + max-size: "10m" + max-file: "3" + + supplier: + build: + context: ../.. + dockerfile: supplier/Dockerfile + container_name: marketplace-mainnet-supplier-openrouter-reseller + restart: unless-stopped + depends_on: + postgres: + condition: service_healthy + env_file: + - path: /var/lib/agents-marketplace/openrouter-reseller.env + required: true + healthcheck: + test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null || exit 1"] + interval: 20s + timeout: 5s + retries: 3 + start_period: 20s + networks: + - marketplace-mainnet-net + - traefik-net + - reseller-db + labels: + - "traefik.enable=true" + - "traefik.docker.network=traefik-net" + - "traefik.http.routers.mp-openrouter-reseller-mainnet.entrypoints=web" + - "traefik.http.routers.mp-openrouter-reseller-mainnet.rule=Host(`mp-suppliers-openrouter-reseller.vector.apexfusion.org`)" + - "traefik.http.routers.mp-openrouter-reseller-mainnet.middlewares=mp-openrouter-reseller-mainnet-https" + - "traefik.http.middlewares.mp-openrouter-reseller-mainnet-https.redirectscheme.scheme=https" + - "traefik.http.middlewares.mp-openrouter-reseller-mainnet-https.redirectscheme.permanent=true" + - "traefik.http.routers.mp-openrouter-reseller-mainnet-tls.entrypoints=websecure" + - "traefik.http.routers.mp-openrouter-reseller-mainnet-tls.rule=Host(`mp-suppliers-openrouter-reseller.vector.apexfusion.org`)" + - "traefik.http.routers.mp-openrouter-reseller-mainnet-tls.tls.certresolver=letsencrypt" + - "traefik.http.services.mp-openrouter-reseller-mainnet.loadbalancer.server.port=8080" + logging: + driver: json-file + options: + max-size: "10m" + max-file: "3" + +volumes: + openrouter-reseller-postgres: + name: marketplace-mainnet-openrouter-reseller-postgres + +networks: + reseller-db: + internal: true + marketplace-mainnet-net: + external: true + name: marketplace-mainnet-net + traefik-net: + external: true + name: traefik-net diff --git a/deploy/mainnet/openrouter-reseller.config.json b/deploy/mainnet/openrouter-reseller.config.json new file mode 100644 index 0000000..401a167 --- /dev/null +++ b/deploy/mainnet/openrouter-reseller.config.json @@ -0,0 +1,21 @@ +{ + "networkId": 1, + "ogmiosUrl": "https://ogmios.vector.mainnet.apexfusion.org", + "endpointUrl": "https://mp-suppliers-openrouter-reseller.vector.apexfusion.org", + "marketplaceModel": "vector/deepseek-v4-flash-spot", + "openRouterModel": "deepseek/deepseek-v4-flash", + "maxInputTokens": 8192, + "maxOutputTokens": 512, + "maxProcessingMs": 600000, + "inferenceTimeoutMs": 180000, + "priceLovelace": "200000", + "supplierBondLovelace": "1000000", + "buyerBondLovelace": "1000000", + "reserveUsd": "1.00", + "pollIntervalMs": 30000, + "providerTimeoutMs": 10000, + "previewMaxChars": 160, + "minimumWalletLovelace": "20000000", + "fundingTimeoutMs": 3600000, + "composeFile": "deploy/mainnet/docker-compose.supplier-openrouter-reseller.yml" +} diff --git a/gateway/package.json b/gateway/package.json index 90b56bd..0bf124f 100644 --- a/gateway/package.json +++ b/gateway/package.json @@ -8,7 +8,8 @@ "start": "tsx src/index.ts", "dev": "tsx src/index.ts", "rotate-master-key": "tsx src/cli/rotate-master-key.ts", - "create-demo-key": "tsx src/cli/create-demo-key.ts" + "create-demo-key": "tsx src/cli/create-demo-key.ts", + "reseller-demo": "tsx src/cli/reseller-demo.ts" }, "dependencies": { "@marketplace/buyer": "workspace:*", diff --git a/gateway/src/cli/reseller-demo.ts b/gateway/src/cli/reseller-demo.ts new file mode 100644 index 0000000..0d8dc2b --- /dev/null +++ b/gateway/src/cli/reseller-demo.ts @@ -0,0 +1,197 @@ +import { resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +interface DemoOptions { + gatewayUrl: string; + dashboardUrl: string; + model: string; + prompt: string; + maxTokens: number; + timeoutMs: number; +} + +interface PublicJob { + escrow_ref: string; + status: "settled" | "failed"; + upstream_cost_usd: string | null; + ap3x_payout: string; + failure_reason: string | null; +} + +export function parseDemoArgs(argv: string[]): DemoOptions { + const values: Record = {}; + for (let index = 0; index < argv.length; index += 2) { + const flag = argv[index]; + const value = argv[index + 1]; + if (!flag?.startsWith("--") || value === undefined) { + throw new Error(`invalid demo argument near ${flag ?? "end"}`); + } + if ( + flag !== "--gateway-url" && + flag !== "--dashboard-url" && + flag !== "--model" && + flag !== "--prompt" && + flag !== "--max-tokens" && + flag !== "--timeout-ms" + ) { + throw new Error(`unknown demo argument: ${flag}`); + } + values[flag] = value; + } + const gatewayUrl = requiredUrl(values["--gateway-url"], "--gateway-url"); + const dashboardUrl = requiredUrl(values["--dashboard-url"], "--dashboard-url"); + const model = requiredString(values["--model"], "--model"); + const prompt = values["--prompt"] ?? "Reply with one sentence that confirms this marketplace job completed."; + const maxTokens = positiveInteger(values["--max-tokens"] ?? "64", "--max-tokens"); + const timeoutMs = positiveInteger(values["--timeout-ms"] ?? "900000", "--timeout-ms"); + return { + gatewayUrl: gatewayUrl.replace(/\/+$/, ""), + dashboardUrl: dashboardUrl.replace(/\/+$/, ""), + model, + prompt, + maxTokens, + timeoutMs, + }; +} + +export async function runResellerDemo( + options: DemoOptions, + gatewayApiKey: string, +): Promise> { + if (gatewayApiKey === "") throw new Error("GATEWAY_API_KEY is required"); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), options.timeoutMs); + try { + const response = await fetch( + `${options.gatewayUrl}/openai/v1/chat/completions`, + { + method: "POST", + headers: { + authorization: `Bearer ${gatewayApiKey}`, + "content-type": "application/json", + }, + body: JSON.stringify({ + model: options.model, + messages: [{ role: "user", content: options.prompt }], + max_tokens: options.maxTokens, + stream: false, + public_preview: true, + }), + signal: controller.signal, + }, + ); + const body = await response.json().catch(() => null) as Record | null; + if (!response.ok) { + throw new Error( + `gateway returned HTTP ${response.status}: ${JSON.stringify(body)}`, + ); + } + const escrowRef = response.headers.get("x-vector-escrow-ref"); + if (!escrowRef) throw new Error("gateway response omitted X-Vector-Escrow-Ref"); + const job = await waitForPublicJob( + options.dashboardUrl, + escrowRef, + options.timeoutMs, + controller.signal, + ); + if (job.status === "failed") { + throw new Error(`reseller job failed: ${job.failure_reason ?? "unknown"}`); + } + return { + completion_id: body?.id ?? null, + escrow_ref: escrowRef, + model: body?.model ?? options.model, + upstream_cost_usd: job.upstream_cost_usd, + ap3x_payout: job.ap3x_payout, + dashboard_url: options.dashboardUrl, + }; + } finally { + clearTimeout(timer); + } +} + +async function waitForPublicJob( + dashboardUrl: string, + escrowRef: string, + timeoutMs: number, + signal: AbortSignal, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const response = await fetch(`${dashboardUrl}/api/jobs?limit=100`, { + cache: "no-store", + signal, + }); + if (!response.ok) { + throw new Error(`dashboard jobs returned HTTP ${response.status}`); + } + const body = await response.json() as { jobs?: PublicJob[] }; + const job = body.jobs?.find((candidate) => candidate.escrow_ref === escrowRef); + if (job) return job; + await new Promise((resolveWait) => setTimeout(resolveWait, 2_000)); + } + throw new Error(`timed out waiting for dashboard settlement ${escrowRef}`); +} + +function requiredString(value: string | undefined, flag: string): string { + if (!value || value.trim() === "") throw new Error(`${flag} is required`); + return value.trim(); +} + +function requiredUrl(value: string | undefined, flag: string): string { + const raw = requiredString(value, flag); + let url: URL; + try { + url = new URL(raw); + } catch { + throw new Error(`${flag} must be a valid URL`); + } + const loopback = + url.hostname === "localhost" || + url.hostname === "127.0.0.1" || + url.hostname === "[::1]"; + if (url.protocol !== "https:" && !(url.protocol === "http:" && loopback)) { + throw new Error(`${flag} must use HTTPS except on loopback`); + } + if ( + url.username !== "" || + url.password !== "" || + url.search !== "" || + url.hash !== "" + ) { + throw new Error(`${flag} must not contain credentials, query, or fragment`); + } + return raw; +} + +function positiveInteger(value: string, flag: string): number { + if (!/^[1-9]\d*$/.test(value) || !Number.isSafeInteger(Number(value))) { + throw new Error(`${flag} must be a positive safe integer`); + } + return Number(value); +} + +export async function main( + argv: string[], + env: Record, +): Promise { + try { + const result = await runResellerDemo( + parseDemoArgs(argv), + env.GATEWAY_API_KEY ?? "", + ); + process.stdout.write(`${JSON.stringify(result, null, 2)}\n`); + return 0; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + process.stderr.write(`reseller demo failed: ${message}\n`); + return 1; + } +} + +const invoked = fileURLToPath(import.meta.url) === resolve(process.argv[1] ?? ""); +if (invoked) { + void main(process.argv.slice(2), process.env).then((code) => { + process.exitCode = code; + }); +} diff --git a/gateway/src/openai/chatCompletions.ts b/gateway/src/openai/chatCompletions.ts index 9493e11..e6ae98b 100644 --- a/gateway/src/openai/chatCompletions.ts +++ b/gateway/src/openai/chatCompletions.ts @@ -152,6 +152,7 @@ async function runOneShot( messages: parsed.messages, payment_lovelace: cand.priceLovelace, max_output_tokens: parsed.maxTokens, + public_preview: parsed.publicPreview, }); used = cand; break; @@ -167,6 +168,9 @@ async function runOneShot( // Settle: resolve the live Submitted ref, Accept, await confirmation. const originalRefStr = `${result.escrowRef.txHash}#${result.escrowRef.index}`; + if (!parsed.stream) { + res.setHeader("X-Vector-Escrow-Ref", originalRefStr); + } const submittedRef = await resolveSubmittedRef({ indexerUrl: config.indexerUrl, buyerPkh: ctx.walletKey.pubKeyHash, diff --git a/gateway/src/openai/errors.ts b/gateway/src/openai/errors.ts index e8687c6..db6c66b 100644 --- a/gateway/src/openai/errors.ts +++ b/gateway/src/openai/errors.ts @@ -71,6 +71,22 @@ export function toGatewayError(err: unknown): GatewayError { return new GatewayError(502, "server_error", "indexer_error", `indexer error: ${err.reason}`); } if (err instanceof TxConstructionError) { + if (err.reason === "input_cap_exceeded") { + return new GatewayError( + 400, + "invalid_request_error", + "input_cap_exceeded", + err.message, + ); + } + if (err.reason === "supplier_preflight_failed") { + return new GatewayError( + 503, + "server_error", + "supplier_preflight_failed", + err.message, + ); + } return new GatewayError(502, "server_error", "escrow_failed", `escrow build failed: ${err.reason}`); } const message = err instanceof Error ? err.message : String(err); diff --git a/gateway/src/openai/validate.ts b/gateway/src/openai/validate.ts index cf7ccdb..be27fc8 100644 --- a/gateway/src/openai/validate.ts +++ b/gateway/src/openai/validate.ts @@ -12,6 +12,7 @@ export interface ParsedChatRequest { stream: boolean; tools?: unknown[]; toolChoice?: unknown; + publicPreview: boolean; } const ROLES = new Set(["system", "user", "assistant"]); @@ -83,6 +84,14 @@ export function parseChatRequest(body: unknown, opts?: { allowTools?: boolean }) maxTokens = mt; } + if (b.public_preview !== undefined && typeof b.public_preview !== "boolean") { + throw badRequest( + "invalid_public_preview", + "`public_preview` must be a boolean when provided", + ); + } + const publicPreview = b.public_preview === true; + const stream = b.stream === true; - return { model, messages, maxTokens, stream, tools, toolChoice }; + return { model, messages, maxTokens, stream, tools, toolChoice, publicPreview }; } diff --git a/gateway/src/server.ts b/gateway/src/server.ts index aa4a5cf..fed00dd 100644 --- a/gateway/src/server.ts +++ b/gateway/src/server.ts @@ -41,6 +41,7 @@ function corsMiddleware(allowOrigins: string[]) { res.setHeader("Vary", "Origin"); res.setHeader("Access-Control-Allow-Methods", "GET, POST, OPTIONS"); res.setHeader("Access-Control-Allow-Headers", "Content-Type, Authorization"); + res.setHeader("Access-Control-Expose-Headers", "X-Vector-Escrow-Ref"); res.setHeader("Access-Control-Max-Age", "600"); } if (req.method === "OPTIONS") { diff --git a/packages/shared/src/tx/advert/postAdvert.ts b/packages/shared/src/tx/advert/postAdvert.ts index 61406cf..be0c865 100644 --- a/packages/shared/src/tx/advert/postAdvert.ts +++ b/packages/shared/src/tx/advert/postAdvert.ts @@ -28,6 +28,8 @@ export interface PostAdvertParams { walletKey: WalletKey; advertDatum: AdvertDatum; deposit_lovelace: bigint; + /** Durable journal hook invoked with signed CBOR before broadcast. */ + beforeSubmit?: (built: PostAdvertBuildResult) => Promise; } export async function buildPostAdvertTx( @@ -75,6 +77,7 @@ export async function buildPostAdvertTx( walletKey, advertDatum, deposit_lovelace, + beforeSubmit: params.beforeSubmit, }); } @@ -102,13 +105,15 @@ export async function buildPostAdvertTx( const txCborHex = encodeTxBody(body); const expectedTxHash = sha256Hex(txCborHex); - - // Submit so the mock seeds the new UTxO + spends inputs (none here). - await chain.submitTx(txCborHex); - - return { + const built: PostAdvertBuildResult = { txCborHex, expectedTxHash, advertOutputRef: { txHash: expectedTxHash, index: 0 }, }; + await params.beforeSubmit?.(built); + + // Submit so the mock seeds the new UTxO + spends inputs (none here). + await chain.submitTx(txCborHex); + + return built; } diff --git a/packages/shared/src/tx/escrow/submit.ts b/packages/shared/src/tx/escrow/submit.ts index 918e207..cc24486 100644 --- a/packages/shared/src/tx/escrow/submit.ts +++ b/packages/shared/src/tx/escrow/submit.ts @@ -26,6 +26,8 @@ export interface SubmitParams { supplierKey: WalletKey; escrowRef: OutputReference; receiptHash: string; + /** Durable journal hook invoked with signed CBOR before broadcast. */ + beforeSubmit?: (built: BuildResult) => Promise; } function isHex32(s: string): boolean { @@ -113,6 +115,7 @@ export async function buildSubmitTx( deliverBy: datum.deliver_by, tipMs, receiptHash: receiptHash.toLowerCase(), + beforeSubmit: params.beforeSubmit, }); } @@ -138,6 +141,7 @@ export async function buildSubmitTx( const txCborHex = encodeTxBody(body); const expectedTxHash = sha256Hex(txCborHex); + await params.beforeSubmit?.({ txCborHex, expectedTxHash }); await chain.submitTx(txCborHex); return { txCborHex, expectedTxHash }; diff --git a/packages/shared/src/tx/index.ts b/packages/shared/src/tx/index.ts index dd5b5c7..ef03e9e 100644 --- a/packages/shared/src/tx/index.ts +++ b/packages/shared/src/tx/index.ts @@ -5,6 +5,10 @@ export type { ChatMessage, ToolCall, WalletKey, BuildResult, PostAdvertBuildResult, PostEscrowBuildResult } from "./types.js"; export { TxConstructionError } from "./types.js"; export { normalizeChatMessage, chatMessagesEquivalent, toolCallsEqual } from "./chatMessage.js"; +export { + BOUNDED_INPUT_DETAIL_MARKER, + chatInputTokenUpperBound, +} from "./inputBound.js"; export { mockSlotToWallclockMs, mockWallclockMsToSlot, NETWORK_BUFFER_MS } from "./internal/constants.js"; export { detectCborBackend, type CborBackend } from "./internal/cborBackend.js"; export { escrowLockFloor, minAdaForEscrowDatum } from "./internal/minAdaFloor.js"; diff --git a/packages/shared/src/tx/inputBound.ts b/packages/shared/src/tx/inputBound.ts new file mode 100644 index 0000000..f63bcc1 --- /dev/null +++ b/packages/shared/src/tx/inputBound.ts @@ -0,0 +1,26 @@ +import type { ChatMessage } from "./types.js"; +import { canonicalize } from "../cbor/canonical.js"; + +/** Advert detail-URI marker for suppliers that require live input-cap preflight. */ +export const BOUNDED_INPUT_DETAIL_MARKER = "#vector-bounded-input-v1"; + +const CHAT_MESSAGE_OVERHEAD = 8; +const CHAT_REPLY_PRIMER = 3; +const encoder = new TextEncoder(); + +/** + * Conservative, tokenizer-independent upper bound for text chat input. + * + * Every UTF-8 byte counts as one token unit. The fixed per-message and reply + * overhead covers role separators and the provider chat template. Major + * byte-fallback tokenizers cannot emit more content tokens than input bytes. + */ +export function chatInputTokenUpperBound( + messages: readonly ChatMessage[], +): number { + return ( + CHAT_REPLY_PRIMER + + CHAT_MESSAGE_OVERHEAD * messages.length + + encoder.encode(canonicalize(messages)).byteLength + ); +} diff --git a/packages/shared/src/tx/internal/liveCbor.ts b/packages/shared/src/tx/internal/liveCbor.ts index 1410fbe..16f2a5e 100644 --- a/packages/shared/src/tx/internal/liveCbor.ts +++ b/packages/shared/src/tx/internal/liveCbor.ts @@ -101,6 +101,7 @@ export interface LiveSubmitParams { deliverBy: number; tipMs: number; receiptHash: string; + beforeSubmit?: (built: BuildResult) => Promise; } export interface LiveAcceptParams { @@ -136,6 +137,7 @@ export interface LivePostAdvertParams { walletKey: WalletKey; advertDatum: AdvertDatum; deposit_lovelace: bigint; + beforeSubmit?: (built: PostAdvertBuildResult) => Promise; } export interface LiveRetireAdvertParams { @@ -499,6 +501,7 @@ async function buildSpendOpenOrClaimedTx(opts: { validFromMs: number; validToMs: number; signerPkh: string; + beforeSubmit?: (built: BuildResult) => Promise; }): Promise { const provider = buildLucidProvider(opts.chain); const { lucid } = await createLucidContext(provider, opts.walletKey, { @@ -585,6 +588,7 @@ async function buildSpendOpenOrClaimedTx(opts: { const txCborHex = signed.toCBOR(); const expectedTxHash = signed.toHash(); + await opts.beforeSubmit?.({ txCborHex, expectedTxHash }); await opts.chain.submitTx(txCborHex); @@ -650,6 +654,7 @@ export async function buildLiveTxForSubmit(params: LiveSubmitParams): Promise=6.8.1'} + pg-cloudflare@1.4.0: + resolution: {integrity: sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==} + + pg-connection-string@2.14.0: + resolution: {integrity: sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==} + + pg-int8@1.0.1: + resolution: {integrity: sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==} + engines: {node: '>=4.0.0'} + + pg-pool@3.14.0: + resolution: {integrity: sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==} + peerDependencies: + pg: '>=8.0' + + pg-protocol@1.16.0: + resolution: {integrity: sha512-sILXutLVjCLjcDuOmvhX5e2Z4cS5qG/6Bu3VkpFwdf/633ElGLpEh9bgmuI5I4sqKqkifQiGyiCcx1HdtrK7tg==} + + pg-types@2.2.0: + resolution: {integrity: sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==} + engines: {node: '>=4'} + + pg@8.23.0: + resolution: {integrity: sha512-Ip2EQCngowJLGOfCwkFhPXU7/ljlhn6Rxlmy4XYfL2Y+vyRM59+8uR2xqRWKdYmbXmxCFOAmKxBuSUCdF34qLg==} + engines: {node: '>= 16.0.0'} + peerDependencies: + pg-native: '>=3.0.1' + peerDependenciesMeta: + pg-native: + optional: true + + pgpass@1.0.5: + resolution: {integrity: sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==} + picocolors@1.1.1: resolution: {integrity: sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA==} @@ -2430,6 +2473,22 @@ packages: resolution: {integrity: sha512-pMMHxBOZKFU6HgAZ4eyGnwXF/EvPGGqUr0MnZ5+99485wwW41kW91A4LOGxSHhgugZmSChL5AlElNdwlNgcnLQ==} engines: {node: ^10 || ^12 || >=14} + postgres-array@2.0.0: + resolution: {integrity: sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==} + engines: {node: '>=4'} + + postgres-bytea@1.0.1: + resolution: {integrity: sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==} + engines: {node: '>=0.10.0'} + + postgres-date@1.0.7: + resolution: {integrity: sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==} + engines: {node: '>=0.10.0'} + + postgres-interval@1.2.0: + resolution: {integrity: sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==} + engines: {node: '>=0.10.0'} + prebuild-install@7.1.3: resolution: {integrity: sha512-8Mf2cbV7x1cXPUILADGI3wuhfqWvtiLA1iclTDbFRZkgRQS0NqsPZphna9V+HyTEadheuPmjaJMsbzKQFOzLug==} engines: {node: '>=10'} @@ -2640,6 +2699,10 @@ packages: resolution: {integrity: sha512-UXWMKhLOwVKb728IUtQPXxfYU+usdybtUrK/8uGE8CQMvrhOpwvzDBwj0QhSL7MQc7vIsISBG8VQ8+IDQxpfQA==} engines: {node: '>=0.10.0'} + split2@4.2.0: + resolution: {integrity: sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==} + engines: {node: '>= 10.x'} + sprintf-js@1.1.3: resolution: {integrity: sha512-Oo+0REFV59/rz3gfJNKQiBlwfHaSESl1pcGyABQsnnIfWOFt6JNj5gCog2U6MLZ//IGYD+nA8nI+mTShREReaA==} @@ -3863,6 +3926,12 @@ snapshots: dependencies: '@types/node': 25.6.0 + '@types/pg@8.23.1': + dependencies: + '@types/node': 25.6.0 + pg-protocol: 1.16.0 + pg-types: 2.2.0 + '@types/prop-types@15.7.15': {} '@types/qs@6.15.0': {} @@ -4947,6 +5016,41 @@ snapshots: dependencies: node-ensure: 0.0.0 + pg-cloudflare@1.4.0: + optional: true + + pg-connection-string@2.14.0: {} + + pg-int8@1.0.1: {} + + pg-pool@3.14.0(pg@8.23.0): + dependencies: + pg: 8.23.0 + + pg-protocol@1.16.0: {} + + pg-types@2.2.0: + dependencies: + pg-int8: 1.0.1 + postgres-array: 2.0.0 + postgres-bytea: 1.0.1 + postgres-date: 1.0.7 + postgres-interval: 1.2.0 + + pg@8.23.0: + dependencies: + pg-connection-string: 2.14.0 + pg-pool: 3.14.0(pg@8.23.0) + pg-protocol: 1.16.0 + pg-types: 2.2.0 + pgpass: 1.0.5 + optionalDependencies: + pg-cloudflare: 1.4.0 + + pgpass@1.0.5: + dependencies: + split2: 4.2.0 + picocolors@1.1.1: {} picomatch@2.3.2: {} @@ -4997,6 +5101,16 @@ snapshots: picocolors: 1.1.1 source-map-js: 1.2.1 + postgres-array@2.0.0: {} + + postgres-bytea@1.0.1: {} + + postgres-date@1.0.7: {} + + postgres-interval@1.2.0: + dependencies: + xtend: 4.0.2 + prebuild-install@7.1.3: dependencies: detect-libc: 2.1.2 @@ -5282,6 +5396,8 @@ snapshots: source-map-js@1.2.1: {} + split2@4.2.0: {} + sprintf-js@1.1.3: {} stackback@0.0.2: {} diff --git a/supplier/package.json b/supplier/package.json index 36e11ef..b8c1873 100644 --- a/supplier/package.json +++ b/supplier/package.json @@ -9,17 +9,20 @@ "tx:post-advert": "tsx src/cli/post-advert.ts", "tx:retire-advert": "tsx src/cli/retire-advert.ts", "tx:publish-reference-scripts": "tsx src/cli/publish-reference-scripts.ts", - "tx:consolidate-wallet": "tsx src/cli/consolidate-wallet.ts" + "tx:consolidate-wallet": "tsx src/cli/consolidate-wallet.ts", + "reseller:setup": "tsx src/cli/setup-reseller.ts" }, "dependencies": { "@marketplace/shared": "workspace:*", "@noble/ed25519": "^2.1.0", "@noble/hashes": "^1.8.0", "bech32": "^2.0.0", - "express": "^4.21.2" + "express": "^4.21.2", + "pg": "^8.16.3" }, "devDependencies": { "@types/express": "^5.0.3", + "@types/pg": "^8.15.5", "@types/supertest": "^6.0.2", "supertest": "^7.0.0", "typescript": "^5.7.2" diff --git a/supplier/src/cli/postAdvertFlow.ts b/supplier/src/cli/postAdvertFlow.ts index a9f8272..41534b3 100644 --- a/supplier/src/cli/postAdvertFlow.ts +++ b/supplier/src/cli/postAdvertFlow.ts @@ -15,6 +15,7 @@ import { detectCborBackend, mockSlotToWallclockMs, TxConstructionError, + type PostAdvertBuildResult, type WalletKey, } from "@marketplace/shared/tx"; @@ -34,6 +35,8 @@ export interface PostAdvertFlowParams { * in that order. Defaults to console.log. */ log?: (line: string) => void; + /** Durable journal hook invoked with the signed transaction before broadcast. */ + beforeSubmit?: (built: PostAdvertBuildResult) => Promise; } export interface PostAdvertFlowResult { @@ -110,6 +113,7 @@ export async function runPostAdvert( walletKey, advertDatum: datumToSubmit, deposit_lovelace: datumToSubmit.supplier_bond_lovelace, + beforeSubmit: params.beforeSubmit, }); writeLog(`submitted tx ${built.expectedTxHash}`); diff --git a/supplier/src/cli/setup-reseller.ts b/supplier/src/cli/setup-reseller.ts new file mode 100644 index 0000000..b23048d --- /dev/null +++ b/supplier/src/cli/setup-reseller.ts @@ -0,0 +1,494 @@ +import { spawn } from "node:child_process"; +import { createHash, randomBytes } from "node:crypto"; +import { constants } from "node:fs"; +import { chmod, mkdir, open, rename } from "node:fs/promises"; +import { dirname, relative, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { LiveOgmiosProvider, type OutputReference } from "@marketplace/shared/chain"; +import { decodeAdvertDatum, type AdvertDatum } from "@marketplace/shared/cbor"; +import { + BOUNDED_INPUT_DETAIL_MARKER, + loadBlueprint, + type WalletKey, +} from "@marketplace/shared/tx"; +import { deriveKeypair } from "./gen-keypair.js"; +import { runPostAdvert } from "./postAdvertFlow.js"; +import { callOpenAi } from "../openai.js"; +import { OpenRouterCapacityProvider } from "../reseller/openRouterCapacity.js"; +import { + loadResellerSetupConfig, + type ResellerSetupConfig, +} from "../reseller/setupConfig.js"; + +interface CliOptions { + configPath: string; + startStack: boolean; +} + +interface SetupState extends Record { + SUPPLIER_PRIV_KEY_HEX: string; + SUPPLIER_PUB_KEY_HEX: string; + SUPPLIER_PKH: string; + SUPPLIER_ADDRESS: string; + POSTGRES_PASSWORD: string; + OPENAI_API_KEY: string; + ADVERT_REF: string; +} + +interface SetupResult { + supplierAddress: string; + supplierPkh: string; + advertRef: string; + endpointUrl: string; + dashboardUrl: string; + marketplaceModel: string; + providerModel: string; + stackStarted: boolean; +} + +const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), "../../.."); +const OPENROUTER_BASE_URL = "https://openrouter.ai/api"; +const SHA256_EMPTY = createHash("sha256").update("").digest("hex"); +const MAINNET_ZERO_TIME_MS = "1756485600000"; +const MAINNET_ESCROW_REF = + "c8d84c6d67ec67a1efe5e9c6c06d53020e05d1bb96d1c55ecb1eb7d5010c4d54#0"; +const MAINNET_ADVERT_REF = + "c8d84c6d67ec67a1efe5e9c6c06d53020e05d1bb96d1c55ecb1eb7d5010c4d54#1"; +const POSTGRES_STATE_FILE = + "/var/lib/agents-marketplace/openrouter-reseller-postgres.env"; +const DEFAULT_STATE_FILE = "/var/lib/agents-marketplace/openrouter-reseller.env"; +const RESELLER_HOSTNAME = + "mp-suppliers-openrouter-reseller.vector.apexfusion.org"; + +export function parseSetupArgs(argv: string[]): CliOptions { + let configPath = ""; + let startStack = true; + for (let index = 0; index < argv.length; index += 1) { + const arg = argv[index]; + if (arg === "--config") { + const value = argv[index + 1]; + if (!value) throw new Error("--config requires a path"); + configPath = value; + index += 1; + } else if (arg === "--no-start") { + startStack = false; + } else { + throw new Error(`unknown setup argument: ${arg}`); + } + } + if (configPath === "") throw new Error("--config is required"); + return { configPath, startStack }; +} + +export async function runResellerSetup( + options: CliOptions, + env: Record, + log: (line: string) => void = (line) => console.error(line), +): Promise { + const config = await loadResellerSetupConfig( + resolve(REPO_ROOT, options.configPath), + ); + if (new URL(config.endpointUrl).hostname !== RESELLER_HOSTNAME) { + throw new Error(`endpointUrl host must be ${RESELLER_HOSTNAME}`); + } + const statePath = resolveStatePath(env.RESELLER_STATE_FILE, options.startStack); + process.env.VECTOR_ZERO_TIME_MS = MAINNET_ZERO_TIME_MS; + process.env.ESCROW_REF_UTXO = MAINNET_ESCROW_REF; + process.env.ADVERT_REF_UTXO = MAINNET_ADVERT_REF; + const prior = await readState(statePath); + const privateKey = prior.SUPPLIER_PRIV_KEY_HEX ?? randomBytes(32).toString("hex"); + const derived = deriveKeypair(privateKey, config.networkId); + const postgresPassword = prior.POSTGRES_PASSWORD ?? randomBytes(24).toString("hex"); + const openRouterKey = env.OPENROUTER_API_KEY ?? prior.OPENAI_API_KEY; + if (!openRouterKey) { + throw new Error("OPENROUTER_API_KEY is required on the first setup run"); + } + + const runtimeState = buildRuntimeState( + config, + { + privateKeyHex: derived.privateKeyHex, + pubKeyHex: derived.publicKeyHex, + pubKeyHash: derived.pubKeyHash, + address: derived.address, + }, + postgresPassword, + openRouterKey, + prior.ADVERT_REF ?? "", + ); + await writeState(statePath, runtimeState); + await writeState(POSTGRES_STATE_FILE, { + POSTGRES_USER: runtimeState.POSTGRES_USER, + POSTGRES_DB: runtimeState.POSTGRES_DB, + POSTGRES_PASSWORD: runtimeState.POSTGRES_PASSWORD, + }); + + log("checking capped OpenRouter allowance and model pricing"); + const capacity = new OpenRouterCapacityProvider({ + baseUrl: OPENROUTER_BASE_URL, + apiKey: openRouterKey, + timeoutMs: config.providerTimeoutMs, + }); + await capacity.readCapacity(config.openRouterModel); + + log("running one-token OpenRouter inference probe"); + await callOpenAi({ + baseUrl: OPENROUTER_BASE_URL, + apiKey: openRouterKey, + model: config.openRouterModel, + messages: [{ role: "user", content: "Reply OK" }], + maxTokens: 1, + disableReasoning: true, + timeoutMs: config.inferenceTimeoutMs, + }); + + const chain = new LiveOgmiosProvider({ ogmiosUrl: config.ogmiosUrl }); + log(`seller wallet ${derived.address}`); + await waitForFunding( + chain, + derived.address, + config.minimumWalletLovelace, + config.fundingTimeoutMs, + log, + ); + + const walletKey: WalletKey = { + privateKeyHex: derived.privateKeyHex, + pubKeyHex: derived.publicKeyHex, + pubKeyHash: derived.pubKeyHash, + address: derived.address, + }; + const advert = buildAdvert(config, walletKey); + const advertRef = await findOrPostAdvert( + chain, + walletKey, + advert, + runtimeState.ADVERT_REF, + log, + async (pendingRef) => { + runtimeState.ADVERT_REF = `${pendingRef.txHash}#${pendingRef.index}`; + await writeState(statePath, runtimeState); + }, + ); + runtimeState.ADVERT_REF = `${advertRef.txHash}#${advertRef.index}`; + await writeState(statePath, runtimeState); + + if (options.startStack) { + log("starting reseller and Postgres stack"); + await runCompose(config.composeFile); + } + + return { + supplierAddress: walletKey.address, + supplierPkh: walletKey.pubKeyHash, + advertRef: runtimeState.ADVERT_REF, + endpointUrl: config.endpointUrl, + dashboardUrl: `${config.endpointUrl}/reseller`, + marketplaceModel: config.marketplaceModel, + providerModel: config.openRouterModel, + stackStarted: options.startStack, + }; +} + +function buildRuntimeState( + config: ResellerSetupConfig, + wallet: WalletKey, + postgresPassword: string, + openRouterKey: string, + advertRef: string, +): SetupState { + const hostname = new URL(config.endpointUrl).hostname; + return { + SUPPLIER_PRIV_KEY_HEX: wallet.privateKeyHex, + SUPPLIER_PUB_KEY_HEX: wallet.pubKeyHex, + SUPPLIER_PKH: wallet.pubKeyHash, + SUPPLIER_ADDRESS: wallet.address, + POSTGRES_USER: "reseller", + POSTGRES_DB: "reseller", + POSTGRES_PASSWORD: postgresPassword, + OPENAI_API_KEY: openRouterKey, + ADVERT_REF: advertRef, + OGMIOS_URL: config.ogmiosUrl, + NETWORK_ID: String(config.networkId), + LIVE_CHAIN: "1", + CAPABILITY_KIND: "chat", + LLM_BACKEND: "openai", + OPENAI_BASE_URL: OPENROUTER_BASE_URL, + OPENAI_MODEL_OVERRIDE: config.openRouterModel, + OPENAI_MAX_TOKENS: String(config.maxOutputTokens), + OPENAI_REASONING: "off", + OPENAI_TIMEOUT_MS: String(config.inferenceTimeoutMs), + RESELLER_PROVIDER: "openrouter", + RESELLER_DATABASE_URL: `postgresql://reseller:${postgresPassword}@postgres:5432/reseller`, + RESELLER_RESERVE_USD: config.reserveUsd, + RESELLER_MAX_INPUT_TOKENS: String(config.maxInputTokens), + RESELLER_POLL_INTERVAL_MS: String(config.pollIntervalMs), + RESELLER_PROVIDER_TIMEOUT_MS: String(config.providerTimeoutMs), + RESELLER_PREVIEW_MAX_CHARS: String(config.previewMaxChars), + RESELLER_HOSTNAME: hostname, + RESELLER_ENDPOINT_URL: config.endpointUrl, + RESELLER_MARKETPLACE_MODEL: config.marketplaceModel, + RESELLER_PROVIDER_MODEL: config.openRouterModel, + PORT: "8080", + WALLET_HEALTH_INTERVAL_MS: "600000", + VECTOR_ZERO_TIME_MS: MAINNET_ZERO_TIME_MS, + ESCROW_REF_UTXO: MAINNET_ESCROW_REF, + ADVERT_REF_UTXO: MAINNET_ADVERT_REF, + }; +} + +function buildAdvert(config: ResellerSetupConfig, wallet: WalletKey): AdvertDatum { + return { + supplier_pkh: wallet.pubKeyHash, + capability_id: "llm.text.generate.v1", + model: config.marketplaceModel, + max_output_tokens: config.maxOutputTokens, + max_processing_ms: config.maxProcessingMs, + price_lovelace: config.priceLovelace, + supplier_bond_lovelace: config.supplierBondLovelace, + buyer_bond_lovelace: config.buyerBondLovelace, + endpoint_url: config.endpointUrl, + detail_uri: BOUNDED_INPUT_DETAIL_MARKER, + detail_hash: SHA256_EMPTY, + advertised_at: 0, + status: "Active", + }; +} + +async function findOrPostAdvert( + chain: LiveOgmiosProvider, + walletKey: WalletKey, + expected: AdvertDatum, + savedRef: string, + log: (line: string) => void, + journal: (ref: OutputReference) => Promise, +): Promise { + const parsedSaved = parseRef(savedRef); + if (parsedSaved) { + const utxo = await chain.queryUtxo(parsedSaved); + if (utxo?.datumHex) { + assertAdvertMatches(decodeAdvertDatum(utxo.datumHex), expected); + log(`reusing saved advert ${savedRef}`); + return parsedSaved; + } + try { + await chain.awaitTx(parsedSaved.txHash, 120_000); + const confirmed = await chain.queryUtxo(parsedSaved); + if (confirmed?.datumHex) { + assertAdvertMatches(decodeAdvertDatum(confirmed.datumHex), expected); + log(`reusing confirmed pending advert ${savedRef}`); + return parsedSaved; + } + } catch { + // The pending transaction is no longer viable; authoritative scan follows. + } + } + + const advertAddress = loadBlueprint().advertScriptAddress(1); + const advertUtxos = await chain.queryUtxosByAddress(advertAddress); + const sameModel: Array<{ ref: OutputReference; datum: AdvertDatum }> = []; + for (const utxo of advertUtxos) { + if (!utxo.datumHex) continue; + try { + const datum = decodeAdvertDatum(utxo.datumHex); + if (datum.status === "Active" && datum.model === expected.model) { + sameModel.push({ ref: utxo.ref, datum }); + } + } catch { + // Ignore unrelated or malformed script outputs. + } + } + const otherSeller = sameModel.find( + (candidate) => candidate.datum.supplier_pkh !== expected.supplier_pkh, + ); + if (otherSeller) { + throw new Error( + `marketplace model ${expected.model} is already advertised by another supplier`, + ); + } + const own = sameModel.filter( + (candidate) => candidate.datum.supplier_pkh === expected.supplier_pkh, + ); + if (own.length > 1) { + throw new Error(`multiple active adverts already exist for ${expected.model}`); + } + if (own.length === 1) { + assertAdvertMatches(own[0].datum, expected); + log(`reusing discovered advert ${own[0].ref.txHash}#${own[0].ref.index}`); + return own[0].ref; + } + + const result = await runPostAdvert({ + chain, + walletKey, + advertDatum: expected, + log, + beforeSubmit: async (built) => journal(built.advertOutputRef), + }); + return result.advertRef; +} + +function assertAdvertMatches(actual: AdvertDatum, expected: AdvertDatum): void { + const fieldsMatch = + actual.status === "Active" && + actual.supplier_pkh === expected.supplier_pkh && + actual.capability_id === expected.capability_id && + actual.model === expected.model && + actual.max_output_tokens === expected.max_output_tokens && + actual.max_processing_ms === expected.max_processing_ms && + actual.price_lovelace === expected.price_lovelace && + actual.supplier_bond_lovelace === expected.supplier_bond_lovelace && + actual.buyer_bond_lovelace === expected.buyer_bond_lovelace && + actual.endpoint_url === expected.endpoint_url && + actual.detail_uri === expected.detail_uri && + actual.detail_hash === expected.detail_hash; + if (!fieldsMatch) { + throw new Error("saved advert does not match the reseller config"); + } +} + + +async function waitForFunding( + chain: LiveOgmiosProvider, + address: string, + minimum: bigint, + timeoutMs: number, + log: (line: string) => void, +): Promise { + const deadline = Date.now() + timeoutMs; + let lastReported = -1n; + while (Date.now() < deadline) { + const utxos = await chain.queryUtxosByAddress(address); + const total = utxos.reduce((sum, utxo) => sum + utxo.lovelace, 0n); + if (total >= minimum) return; + if (total !== lastReported) { + log(`wallet funding ${total}/${minimum} lovelace`); + lastReported = total; + } + await new Promise((resolveWait) => setTimeout(resolveWait, 5_000)); + } + throw new Error(`wallet funding timed out for ${address}`); +} + +function resolveStatePath( + configured: string | undefined, + startStack: boolean, +): string { + const statePath = resolve(configured ?? DEFAULT_STATE_FILE); + const repoRoot = REPO_ROOT; + const fromRepo = relative(repoRoot, statePath); + if (fromRepo === "" || (!fromRepo.startsWith("../") && fromRepo !== "..")) { + throw new Error("RESELLER_STATE_FILE must be outside the repository"); + } + if (startStack && statePath !== DEFAULT_STATE_FILE) { + throw new Error( + `stack startup requires the fixed state file ${DEFAULT_STATE_FILE}`, + ); + } + return statePath; +} + +async function readState(path: string): Promise> { + let raw: string; + try { + const handle = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW); + try { + const info = await handle.stat(); + if (!info.isFile()) { + throw new Error(`reseller state must be a regular file: ${path}`); + } + if (typeof process.getuid === "function" && info.uid !== process.getuid()) { + throw new Error(`reseller state is not owned by the current user: ${path}`); + } + if ((info.mode & 0o077) !== 0) { + throw new Error(`reseller state permissions must be 0600: ${path}`); + } + raw = await handle.readFile("utf8"); + } finally { + await handle.close(); + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return {}; + throw error; + } + const state: Record = {}; + for (const line of raw.split(/\r?\n/)) { + if (line === "" || line.startsWith("#")) continue; + const separator = line.indexOf("="); + if (separator <= 0) throw new Error(`invalid reseller state line: ${line}`); + state[line.slice(0, separator)] = line.slice(separator + 1); + } + return state; +} + +async function writeState(path: string, state: Record): Promise { + for (const [name, value] of Object.entries(state)) { + if (/[\r\n]/.test(name) || /[\r\n]/.test(value)) { + throw new Error(`reseller state ${name} must be one line`); + } + } + await mkdir(dirname(path), { recursive: true, mode: 0o700 }); + await chmod(dirname(path), 0o700); + const temp = `${path}.${randomBytes(12).toString("hex")}.tmp`; + const body = Object.entries(state) + .sort(([left], [right]) => left.localeCompare(right)) + .map(([name, value]) => `${name}=${value}`) + .join("\n") + "\n"; + const handle = await open(temp, "wx", 0o600); + try { + await handle.writeFile(body, "utf8"); + await handle.sync(); + } finally { + await handle.close(); + } + await rename(temp, path); + await chmod(path, 0o600); +} + +function parseRef(value: string): OutputReference | null { + const match = /^([0-9a-fA-F]{64})#(0|[1-9]\d*)$/.exec(value); + if (!match) return null; + return { txHash: match[1], index: Number(match[2]) }; +} + +async function runCompose(composeFile: string): Promise { + const args = [ + "compose", + "-f", + resolve(REPO_ROOT, composeFile), + "up", + "-d", + "--build", + "--wait", + ]; + await new Promise((resolveRun, rejectRun) => { + const child = spawn("docker", args, { stdio: "inherit" }); + child.once("error", rejectRun); + child.once("exit", (code) => { + if (code === 0) resolveRun(); + else rejectRun(new Error(`docker compose exited with code ${code}`)); + }); + }); +} + +export async function main( + argv: string[], + env: Record, +): Promise { + try { + const result = await runResellerSetup(parseSetupArgs(argv), env); + process.stdout.write(`${JSON.stringify(result, null, 2)}\n`); + return 0; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + process.stderr.write(`reseller setup failed: ${message}\n`); + return 1; + } +} + +const invoked = fileURLToPath(import.meta.url) === resolve(process.argv[1] ?? ""); +if (invoked) { + void main(process.argv.slice(2), process.env).then((code) => { + process.exitCode = code; + }); +} diff --git a/supplier/src/config.ts b/supplier/src/config.ts index 0ed01b1..d94d20c 100644 --- a/supplier/src/config.ts +++ b/supplier/src/config.ts @@ -62,6 +62,7 @@ const HEX64_RE = /^[0-9a-fA-F]{64}$/; const TX_HASH_RE = /^[0-9a-fA-F]{64}$/; const NON_NEG_INT_RE = /^(?:0|[1-9]\d*)$/; const POS_INT_RE = /^[1-9]\d*$/; +const USD_RE = /^(?:0|[1-9]\d*)(?:\.\d{1,9})?$/; /** Lowercased hostname of a URL, or "" if it doesn't parse. */ function hostnameOf(url: string): string { @@ -72,6 +73,24 @@ function hostnameOf(url: string): string { } } +function isOpenRouterApiBaseUrl(value: string): boolean { + try { + const url = new URL(value); + return ( + url.protocol === "https:" && + url.hostname === "openrouter.ai" && + url.port === "" && + url.username === "" && + url.password === "" && + (url.pathname === "/api" || url.pathname === "/api/") && + url.search === "" && + url.hash === "" + ); + } catch { + return false; + } +} + export interface AdvertRef { txHash: string; index: number; @@ -112,6 +131,22 @@ export type LlmBackend = "ollama" | "openai"; * the buyer's sweeper Reclaims the Open escrow after deliver_by. */ export type ChatSettleMode = "full" | "ticket"; +export type ResellerProvider = "openrouter"; + +export interface ResellerConfig { + provider: ResellerProvider; + databaseUrl: string; + /** Fixed protected amount in provider USD credits. Parsed exactly at runtime. */ + reserveUsd: string; + /** Conservative off-chain input bound advertised by /capability. */ + maxInputTokens: number; + pollIntervalMs: number; + staleAfterMs: number; + providerTimeoutMs: number; + previewMaxChars: number; +} + + export interface SupplierConfig { supplierPrivKeyHex: string; ogmiosUrl: string; @@ -226,6 +261,8 @@ export interface SupplierConfig { * Default 600_000 (10 min). Only fires while liveChain=true. */ walletHealthIntervalMs: number; + /** Optional balance-aware provider resale mode. Null preserves legacy behavior. */ + reseller: ResellerConfig | null; } function requireField(env: Record, name: string): string { @@ -532,6 +569,121 @@ export function loadConfig(env: Record): SupplierCon walletHealthIntervalMs = Number(whIntervalStr); } + const resellerFields = [ + "RESELLER_DATABASE_URL", + "RESELLER_RESERVE_USD", + "RESELLER_MAX_INPUT_TOKENS", + "RESELLER_POLL_INTERVAL_MS", + "RESELLER_PROVIDER_TIMEOUT_MS", + "RESELLER_PREVIEW_MAX_CHARS", + ] as const; + const resellerProviderRaw = (env.RESELLER_PROVIDER ?? "").trim(); + const hasResellerFields = resellerFields.some( + (name) => env[name] !== undefined && env[name] !== "", + ); + let reseller: ResellerConfig | null = null; + if (resellerProviderRaw === "") { + if (hasResellerFields) { + throw new Error( + "loadConfig: RESELLER_PROVIDER is required when reseller fields are set", + ); + } + } else { + if (resellerProviderRaw !== "openrouter") { + throw new Error('loadConfig: RESELLER_PROVIDER must be "openrouter"'); + } + if (capabilityKind !== "chat" || llmBackend !== "openai") { + throw new Error( + 'loadConfig: RESELLER_PROVIDER requires CAPABILITY_KIND="chat" and LLM_BACKEND="openai"', + ); + } + if (!isOpenRouterApiBaseUrl(openaiBaseUrl)) { + throw new Error( + "loadConfig: RESELLER_PROVIDER=openrouter requires OPENAI_BASE_URL=https://openrouter.ai/api", + ); + } + if (openaiApiKey === "") { + throw new Error( + "loadConfig: RESELLER_PROVIDER=openrouter requires OPENAI_API_KEY", + ); + } + if (openaiMaxTokens <= 0) { + throw new Error( + "loadConfig: RESELLER_PROVIDER requires a positive OPENAI_MAX_TOKENS", + ); + } + + const databaseUrl = requireField(env, "RESELLER_DATABASE_URL"); + let databaseProtocol = ""; + try { + databaseProtocol = new URL(databaseUrl).protocol; + } catch { + throw new Error( + "loadConfig: RESELLER_DATABASE_URL must be a valid PostgreSQL URL", + ); + } + if (databaseProtocol !== "postgres:" && databaseProtocol !== "postgresql:") { + throw new Error( + "loadConfig: RESELLER_DATABASE_URL must use postgres:// or postgresql://", + ); + } + + const reserveUsd = requireField(env, "RESELLER_RESERVE_USD").trim(); + if (!USD_RE.test(reserveUsd)) { + throw new Error( + "loadConfig: RESELLER_RESERVE_USD must be a non-negative decimal with at most 9 places", + ); + } + + const maxInputRaw = requireField(env, "RESELLER_MAX_INPUT_TOKENS"); + if (!POS_INT_RE.test(maxInputRaw) || !Number.isSafeInteger(Number(maxInputRaw))) { + throw new Error( + "loadConfig: RESELLER_MAX_INPUT_TOKENS must be a positive safe integer", + ); + } + + const pollRaw = env.RESELLER_POLL_INTERVAL_MS ?? "30000"; + if (!POS_INT_RE.test(pollRaw) || !Number.isSafeInteger(Number(pollRaw))) { + throw new Error( + "loadConfig: RESELLER_POLL_INTERVAL_MS must be a positive safe integer", + ); + } + const pollIntervalMs = Number(pollRaw); + + const providerTimeoutRaw = env.RESELLER_PROVIDER_TIMEOUT_MS ?? "10000"; + if ( + !POS_INT_RE.test(providerTimeoutRaw) || + !Number.isSafeInteger(Number(providerTimeoutRaw)) + ) { + throw new Error( + "loadConfig: RESELLER_PROVIDER_TIMEOUT_MS must be a positive safe integer", + ); + } + + const previewMaxRaw = env.RESELLER_PREVIEW_MAX_CHARS ?? "160"; + if ( + !POS_INT_RE.test(previewMaxRaw) || + !Number.isSafeInteger(Number(previewMaxRaw)) || + Number(previewMaxRaw) > 1000 + ) { + throw new Error( + "loadConfig: RESELLER_PREVIEW_MAX_CHARS must be an integer from 1 to 1000", + ); + } + + reseller = { + provider: "openrouter", + databaseUrl, + reserveUsd, + maxInputTokens: Number(maxInputRaw), + pollIntervalMs, + staleAfterMs: Math.max(pollIntervalMs * 2, pollIntervalMs + 5000), + providerTimeoutMs: Number(providerTimeoutRaw), + previewMaxChars: Number(previewMaxRaw), + }; + } + + return { supplierPrivKeyHex, ogmiosUrl, @@ -565,5 +717,6 @@ export function loadConfig(env: Record): SupplierCon revisionProbeMode, liveChain, walletHealthIntervalMs, + reseller, }; } diff --git a/supplier/src/index.ts b/supplier/src/index.ts index 4404f71..fc312af 100644 --- a/supplier/src/index.ts +++ b/supplier/src/index.ts @@ -22,6 +22,12 @@ import { JobStore } from "./jobs.js"; import { createApp, type SupplierDeps } from "./server.js"; import { buildChainProvider } from "./chain.js"; import { startWalletHealthTicker } from "./walletHealth.js"; +import { decodeAdvertDatum } from "@marketplace/shared/cbor"; +import { BOUNDED_INPUT_DETAIL_MARKER } from "@marketplace/shared/tx"; +import { OpenRouterCapacityProvider } from "./reseller/openRouterCapacity.js"; +import { CapacityGate } from "./reseller/capacityGate.js"; +import { ResellerEvidenceStore } from "./reseller/evidenceStore.js"; +import { ResellerRuntime } from "./reseller/runtime.js"; async function main(): Promise { const cfg = loadConfig(process.env); @@ -45,8 +51,74 @@ async function main(): Promise { // chat-session kind serves N concurrent sessions; one-shot kinds stay 1. const state = new SupplierState(cfg.capabilityKind === "chat-session" ? cfg.maxChatSessions : 1); const jobs = new JobStore(); + let reseller: ResellerRuntime | undefined; + if (cfg.reseller) { + const advertUtxo = await chain.queryUtxo(cfg.advertRef); + if (!advertUtxo?.datumHex) { + throw new Error("reseller boot: configured advert UTxO is unavailable"); + } + const advert = decodeAdvertDatum(advertUtxo.datumHex); + if (advert.status !== "Active") { + throw new Error(`reseller boot: advert status is ${advert.status}`); + } + if (advert.capability_id !== "llm.text.generate.v1") { + throw new Error( + `reseller boot: advert capability must be llm.text.generate.v1, got ${advert.capability_id}`, + ); + } + if (!advert.detail_uri.endsWith(BOUNDED_INPUT_DETAIL_MARKER)) { + throw new Error("reseller boot: advert is missing bounded-input metadata"); + } + if (advert.max_output_tokens !== cfg.openaiMaxTokens) { + throw new Error( + "reseller boot: OPENAI_MAX_TOKENS must equal advert max_output_tokens", + ); + } + if (advert.supplier_pkh !== supplierKey.pubKeyHash) { + throw new Error("reseller boot: advert supplier does not match SUPPLIER_PKH"); + } - const deps: SupplierDeps = { chain, state, config: cfg, supplierKey, jobs }; + const providerModel = cfg.openaiModelOverride || advert.model; + const provider = new OpenRouterCapacityProvider({ + baseUrl: cfg.openaiBaseUrl, + apiKey: cfg.openaiApiKey, + timeoutMs: cfg.reseller.providerTimeoutMs, + }); + const gate = new CapacityGate({ + provider, + model: providerModel, + reserveUsd: cfg.reseller.reserveUsd, + maxInputTokens: cfg.reseller.maxInputTokens, + maxOutputTokens: advert.max_output_tokens, + staleAfterMs: cfg.reseller.staleAfterMs, + }); + const store = new ResellerEvidenceStore( + cfg.reseller.databaseUrl, + cfg.reseller.previewMaxChars, + ); + reseller = new ResellerRuntime({ + chain, + state, + gate, + store, + advert, + advertRef: cfg.advertRef, + providerModel, + pollIntervalMs: cfg.reseller.pollIntervalMs, + maxInputTokens: cfg.reseller.maxInputTokens, + }); + await reseller.initialize(); + } + + + const deps: SupplierDeps = { + chain, + state, + config: cfg, + supplierKey, + jobs, + reseller, + }; const app = createApp(deps); const server = app.listen(cfg.port, () => { @@ -80,7 +152,9 @@ async function main(): Promise { clearInterval(evictInterval); walletHealthTicker?.stop(); state.markOffline(); - server.close(() => process.exit(0)); + server.close(() => { + void (reseller?.close() ?? Promise.resolve()).finally(() => process.exit(0)); + }); setTimeout(() => process.exit(1), 5_000).unref(); }; process.on("SIGINT", () => shutdown("SIGINT")); diff --git a/supplier/src/jobRunner.ts b/supplier/src/jobRunner.ts index 632c07b..a29dffc 100644 --- a/supplier/src/jobRunner.ts +++ b/supplier/src/jobRunner.ts @@ -48,6 +48,8 @@ import type { TtsJobResponsePayload, OcrJobResponsePayload, } from "./jobs.js"; +import type { ResellerRuntime } from "./reseller/runtime.js"; +import { parseUsdNumberCeil } from "./reseller/money.js"; export interface RunChatJobDeps { chain: ChainProvider; @@ -55,6 +57,7 @@ export interface RunChatJobDeps { config: SupplierConfig; supplierKey: WalletKey; jobs: JobStore; + reseller?: ResellerRuntime; } export interface RunChatJobParams { @@ -71,6 +74,15 @@ function sha256Hex(s: string): string { return createHash("sha256").update(s, "utf8").digest("hex"); } +async function recordResellerFailure( + reseller: ResellerRuntime | undefined, + escrowRef: string, + reason: string, +): Promise { + if (!reseller) return; + await reseller.recordFailed(escrowRef, reason).catch(() => undefined); +} + /** * Run Ollama → receipt → Submit in the background. * Always resolves (never rejects). Terminal state written to jobs store. @@ -83,11 +95,31 @@ export async function runChatJob(params: RunChatJobParams): Promise { deps.jobs.setRunning(jobId); try { + if (deps.reseller) { + try { + await deps.reseller.recordInferenceStarted(escrowRef); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + deps.jobs.fail(jobId, { + httpStatus: 503, + reason: "evidence_unavailable", + message, + }); + return; + } + } + // ── 2. Call upstream LLM ─────────────────────────────────────────── // Both backends return the same { content, prompt_tokens, // completion_tokens, wallclock_ms } shape so receipt construction // downstream is identical. - let inference: { content: string; prompt_tokens: number; completion_tokens: number; wallclock_ms: number }; + let inference: { + content: string; + prompt_tokens: number; + completion_tokens: number; + wallclock_ms: number; + cost_usd?: number; + }; try { if (deps.config.llmBackend === "openai") { inference = await openai.callOpenAi({ @@ -127,6 +159,7 @@ export async function runChatJob(params: RunChatJobParams): Promise { reason: collapsedReason, message, }); + await recordResellerFailure(deps.reseller, escrowRef, collapsedReason); return; } @@ -146,8 +179,39 @@ export async function runChatJob(params: RunChatJobParams): Promise { }); const signed = signReceipt(receipt, deps.supplierKey.privateKeyHex); const resultHash = receiptResultHash(signed); + if (deps.reseller) { + try { + await deps.reseller.recordOutput(escrowRef, { + response: assistantMessage, + receipt: { + receipt: signed.receipt, + signature: signed.signature, + }, + promptTokens: inference.prompt_tokens, + completionTokens: inference.completion_tokens, + outputText: inference.content, + actualCostUsdNanos: inference.cost_usd === undefined + ? null + : parseUsdNumberCeil(inference.cost_usd, "OpenRouter usage cost"), + }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + deps.jobs.fail(jobId, { + httpStatus: 503, + reason: "evidence_unavailable", + message, + }); + await recordResellerFailure( + deps.reseller, + escrowRef, + "output_persistence_failed", + ); + return; + } + } - // ── 4+5. Submit tx: build + confirm as ONE wallet critical section ── + + // ── 4+5. Broadcast, journal, then confirm Submit under one wallet lock ─ const submitOutcome = await deps.state.walletMutex.run(async () => { let built: { txCborHex: string; expectedTxHash: string }; try { @@ -156,19 +220,30 @@ export async function runChatJob(params: RunChatJobParams): Promise { supplierKey: deps.supplierKey, escrowRef: claimedRef, receiptHash: resultHash, + beforeSubmit: deps.reseller + ? async (candidate) => { + await deps.reseller!.recordSubmitting( + escrowRef, + `${candidate.expectedTxHash}#0`, + ); + } + : undefined, }); } catch (err) { return { kind: "build_failed" as const, err }; } + const submittedRef = `${built.expectedTxHash}#0`; try { await deps.chain.awaitTx(built.expectedTxHash, 60_000); } catch (err) { - return { kind: "await_failed" as const, err }; + return { kind: "await_failed" as const, err, built, submittedRef }; } - return { kind: "ok" as const }; + return { kind: "ok" as const, built, submittedRef }; }); if (submitOutcome.kind === "build_failed") { - const message = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err); + const message = submitOutcome.err instanceof Error + ? submitOutcome.err.message + : String(submitOutcome.err); console.warn( `[job_failed] jobId=${jobId} reason=submit_failed httpStatus=502 msg=${message}`, ); @@ -177,20 +252,42 @@ export async function runChatJob(params: RunChatJobParams): Promise { reason: "submit_failed", message: `Submit tx failed: ${message}`, }); + await recordResellerFailure(deps.reseller, escrowRef, "submit_failed"); return; } if (submitOutcome.kind === "await_failed") { - const msg = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err); + const message = submitOutcome.err instanceof Error + ? submitOutcome.err.message + : String(submitOutcome.err); console.warn( - `[job_failed] jobId=${jobId} reason=submit_timeout httpStatus=502 msg=${msg}`, + `[job_failed] jobId=${jobId} reason=submit_timeout httpStatus=502 msg=${message}`, ); deps.jobs.fail(jobId, { httpStatus: 502, reason: "submit_timeout", - message: `Submit awaitTx failed: ${msg}`, + message: `Submit awaitTx failed: ${message}`, }); + deps.reseller?.watchSubmitting(escrowRef, submitOutcome.submittedRef); return; } + if (deps.reseller) { + try { + await deps.reseller.recordSubmitConfirmed( + escrowRef, + submitOutcome.submittedRef, + ); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + deps.jobs.fail(jobId, { + httpStatus: 503, + reason: "evidence_unavailable", + message, + }); + deps.reseller.watchSubmitting(escrowRef, submitOutcome.submittedRef); + return; + } + } + // ── 6. Mark complete ──────────────────────────────────────────────── const payload: ChatJobResponsePayload = { @@ -212,6 +309,9 @@ export async function runChatJob(params: RunChatJobParams): Promise { }; deps.jobs.complete(jobId, payload); } finally { + if (deps.reseller) { + await deps.reseller.refreshCapacity().catch(() => undefined); + } // ── 7. Always release this job's slot ─────────────────────────────── try { deps.state.release(escrowRef); diff --git a/supplier/src/openai.ts b/supplier/src/openai.ts index b74ea71..4a76768 100644 --- a/supplier/src/openai.ts +++ b/supplier/src/openai.ts @@ -74,6 +74,8 @@ export interface OpenAiResult { prompt_tokens: number; completion_tokens: number; wallclock_ms: number; + /** Provider-reported request cost in USD when present (OpenRouter extension). */ + cost_usd?: number; /** Present when the model requested tool calls (streaming path only). */ tool_calls?: ToolCall[]; /** Upstream finish_reason when reported (e.g. "stop", "tool_calls"). */ @@ -188,10 +190,18 @@ export async function callOpenAi(params: CallOpenAiParams): Promise; if (typeof usage.prompt_tokens === "number") promptTokens = usage.prompt_tokens; if (typeof usage.completion_tokens === "number") completionTokens = usage.completion_tokens; + if ( + typeof usage.cost === "number" && + Number.isFinite(usage.cost) && + usage.cost >= 0 + ) { + costUsd = usage.cost; + } } const wallclockMs = Date.now() - startedAt; @@ -201,6 +211,7 @@ export async function callOpenAi(params: CallOpenAiParams): Promise number; +} + +export class CapacityGate { + private readonly provider: CapacityProvider; + private readonly model: string; + private readonly reserveUsdNanos: bigint; + private readonly maxInputTokens: number; + private readonly maxOutputTokens: number; + private readonly staleAfterMs: number; + private readonly now: () => number; + private current: CapacitySnapshot; + private inFlight: Promise | null = null; + + constructor(options: CapacityGateOptions) { + this.provider = options.provider; + this.model = options.model; + this.reserveUsdNanos = parseUsdDecimal(options.reserveUsd, "ceil"); + this.maxInputTokens = options.maxInputTokens; + this.maxOutputTokens = options.maxOutputTokens; + this.staleAfterMs = options.staleAfterMs; + this.now = options.now ?? Date.now; + this.current = { + provider: options.provider.id, + model: options.model, + canServe: false, + reason: "starting", + checkedAtMs: null, + lastSuccessAtMs: null, + keyLimitUsdNanos: 0n, + remainingAllowanceUsdNanos: 0n, + reserveUsdNanos: this.reserveUsdNanos, + sellableUsdNanos: 0n, + worstCaseJobUsdNanos: 0n, + promptPriceUsdNanosPerToken: 0n, + completionPriceUsdNanosPerToken: 0n, + requestPriceUsdNanos: 0n, + availableJobs: 0n, + limitReset: null, + error: null, + }; + } + + snapshot(): CapacitySnapshot { + const snapshot = this.current; + if ( + snapshot.lastSuccessAtMs !== null && + this.now() - snapshot.lastSuccessAtMs > this.staleAfterMs + ) { + return { + ...snapshot, + canServe: false, + reason: "stale", + error: "provider capacity data is stale", + }; + } + return { ...snapshot }; + } + + async refresh(): Promise { + if (this.inFlight) return this.inFlight; + this.inFlight = this.performRefresh(); + try { + return await this.inFlight; + } finally { + this.inFlight = null; + } + } + + forceUnavailable( + reason: Extract, + error: string, + ): CapacitySnapshot { + this.current = { + ...this.current, + canServe: false, + reason, + checkedAtMs: this.now(), + error, + }; + return this.snapshot(); + } + + private async performRefresh(): Promise { + try { + const reading = await this.provider.readCapacity(this.model); + const pricing = reading.pricing; + const worstCaseJobUsdNanos = + pricing.requestUsdNanos + + pricing.promptUsdNanosPerToken * BigInt(this.maxInputTokens) + + pricing.completionUsdNanosPerToken * BigInt(this.maxOutputTokens); + const remaining = reading.keyRemainingUsdNanos; + const sellable = remaining > this.reserveUsdNanos + ? remaining - this.reserveUsdNanos + : 0n; + const enoughForOne = + worstCaseJobUsdNanos === 0n || sellable >= worstCaseJobUsdNanos; + const aboveReserve = remaining >= this.reserveUsdNanos; + const canServe = aboveReserve && enoughForOne; + const reason: CapacityReason = canServe + ? "ready" + : !aboveReserve + ? "below_reserve" + : "insufficient_for_job"; + const availableJobs = worstCaseJobUsdNanos === 0n + ? null + : sellable / worstCaseJobUsdNanos; + + this.current = { + provider: reading.provider, + model: this.model, + canServe, + reason, + checkedAtMs: reading.checkedAtMs, + lastSuccessAtMs: reading.checkedAtMs, + keyLimitUsdNanos: reading.keyLimitUsdNanos, + remainingAllowanceUsdNanos: remaining, + reserveUsdNanos: this.reserveUsdNanos, + sellableUsdNanos: sellable, + worstCaseJobUsdNanos, + promptPriceUsdNanosPerToken: pricing.promptUsdNanosPerToken, + completionPriceUsdNanosPerToken: pricing.completionUsdNanosPerToken, + requestPriceUsdNanos: pricing.requestUsdNanos, + availableJobs, + limitReset: reading.limitReset, + error: null, + }; + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + this.current = { + ...this.current, + canServe: false, + reason: "provider_unavailable", + checkedAtMs: this.now(), + error: detail, + }; + } + return this.snapshot(); + } +} diff --git a/supplier/src/reseller/dashboardPage.ts b/supplier/src/reseller/dashboardPage.ts new file mode 100644 index 0000000..630999e --- /dev/null +++ b/supplier/src/reseller/dashboardPage.ts @@ -0,0 +1,83 @@ +export function renderResellerDashboardPage(): string { + return ` + + + + +Reseller capacity + + + +
+

Provider capacity resale

Bounded upstream capacity settled through Vector escrow.

Loading current state
+
+
RouteLoading
+
Marketplace modelLoading
+
Supplier statusLoading
+
+
+
Sellable allowance--Provider allowance --
+
Worst-case job---- jobs available
+
AP3X earned---- settled jobs
+
Upstream spend---- failed jobs
+
+

Conversion timeline

Public previews require buyer consent
+
TimeStatusModelPrompt previewOutput previewUpstreamPayoutEscrow
Loading jobs
+
+ + +`; +} diff --git a/supplier/src/reseller/evidenceStore.ts b/supplier/src/reseller/evidenceStore.ts new file mode 100644 index 0000000..4681723 --- /dev/null +++ b/supplier/src/reseller/evidenceStore.ts @@ -0,0 +1,446 @@ +import { Pool } from "pg"; +import type { ChatMessage } from "@marketplace/shared/tx"; +import type { CapacitySnapshot } from "./capacityGate.js"; +import { formatUsdNanos } from "./money.js"; + +const SCHEMA_SQL = ` +CREATE TABLE IF NOT EXISTS reseller_jobs ( + escrow_ref text PRIMARY KEY, + local_job_id text, + status text NOT NULL CHECK (status IN ('received','claimed','running','submitting','submitted','settled','failed')), + provider text NOT NULL, + provider_model text NOT NULL, + marketplace_model text NOT NULL, + request_json jsonb NOT NULL, + response_json jsonb, + receipt_json jsonb, + public_preview boolean NOT NULL DEFAULT false, + prompt_preview text, + output_preview text, + balance_before_nanos bigint NOT NULL, + balance_after_nanos bigint, + worst_case_cost_nanos bigint NOT NULL, + actual_cost_nanos bigint, + price_lovelace bigint NOT NULL, + prompt_tokens integer, + completion_tokens integer, + claim_tx_hash text, + submitted_ref text, + failure_reason text, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + settled_at timestamptz +); +CREATE INDEX IF NOT EXISTS reseller_jobs_status_idx + ON reseller_jobs (status, updated_at DESC); +CREATE INDEX IF NOT EXISTS reseller_jobs_created_idx + ON reseller_jobs (created_at DESC); + +CREATE TABLE IF NOT EXISTS reseller_capacity ( + singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton), + provider text NOT NULL, + model text NOT NULL, + can_serve boolean NOT NULL, + reason text NOT NULL, + checked_at timestamptz, + remaining_allowance_nanos bigint NOT NULL, + reserve_nanos bigint NOT NULL, + sellable_nanos bigint NOT NULL, + worst_case_job_nanos bigint NOT NULL, + available_jobs bigint, + limit_reset text, + updated_at timestamptz NOT NULL DEFAULT now() +); +`; + +export interface ReceivedJob { + escrowRef: string; + provider: string; + providerModel: string; + marketplaceModel: string; + messages: ChatMessage[]; + publicPreview: boolean; + balanceBeforeUsdNanos: bigint; + worstCaseCostUsdNanos: bigint; + priceLovelace: bigint; +} + +export interface RecordedOutput { + response: Record; + receipt: Record; + promptTokens: number; + completionTokens: number; + actualCostUsdNanos: bigint | null; + outputText: string; +} + +export interface SubmittedJobRow { + escrowRef: string; + submittedRef: string; +} + +export interface PublicResellerJob { + escrow_ref: string; + status: "settled" | "failed"; + provider: string; + provider_model: string; + marketplace_model: string; + prompt_preview: string | null; + output_preview: string | null; + upstream_cost_usd: string | null; + ap3x_payout: string; + claim_tx_hash: string | null; + submitted_ref: string | null; + failure_reason: string | null; + created_at: string; + settled_at: string | null; +} + +export interface PublicResellerTotals { + settled_jobs: number; + failed_jobs: number; + upstream_spend_usd: string; + ap3x_earned: string; +} + +export class ResellerEvidenceStore { + private readonly pool: Pool; + private readonly previewMaxChars: number; + + constructor(databaseUrl: string, previewMaxChars: number) { + this.pool = new Pool({ connectionString: databaseUrl, max: 4 }); + this.previewMaxChars = previewMaxChars; + } + + async initialize(): Promise { + const client = await this.pool.connect(); + try { + await client.query("SELECT pg_advisory_lock(9042713861245)"); + await client.query(SCHEMA_SQL); + } finally { + await client.query("SELECT pg_advisory_unlock(9042713861245)").catch(() => undefined); + client.release(); + } + } + + async ping(): Promise { + await this.pool.query("SELECT 1"); + } + + async close(): Promise { + await this.pool.end(); + } + + async markInterruptedJobs(): Promise { + await this.pool.query( + `UPDATE reseller_jobs + SET status = 'failed', failure_reason = 'interrupted_on_restart', updated_at = now() + WHERE status IN ('received','claimed','running')`, + ); + } + + async saveCapacity(snapshot: CapacitySnapshot): Promise { + await this.pool.query( + `INSERT INTO reseller_capacity ( + singleton, provider, model, can_serve, reason, checked_at, + remaining_allowance_nanos, reserve_nanos, sellable_nanos, + worst_case_job_nanos, available_jobs, limit_reset, updated_at + ) VALUES (true,$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,now()) + ON CONFLICT (singleton) DO UPDATE SET + provider=EXCLUDED.provider, model=EXCLUDED.model, + can_serve=EXCLUDED.can_serve, reason=EXCLUDED.reason, + checked_at=EXCLUDED.checked_at, + remaining_allowance_nanos=EXCLUDED.remaining_allowance_nanos, + reserve_nanos=EXCLUDED.reserve_nanos, + sellable_nanos=EXCLUDED.sellable_nanos, + worst_case_job_nanos=EXCLUDED.worst_case_job_nanos, + available_jobs=EXCLUDED.available_jobs, + limit_reset=EXCLUDED.limit_reset, updated_at=now()`, + [ + snapshot.provider, + snapshot.model, + snapshot.canServe, + snapshot.reason, + snapshot.checkedAtMs === null ? null : new Date(snapshot.checkedAtMs), + snapshot.remainingAllowanceUsdNanos.toString(), + snapshot.reserveUsdNanos.toString(), + snapshot.sellableUsdNanos.toString(), + snapshot.worstCaseJobUsdNanos.toString(), + snapshot.availableJobs?.toString() ?? null, + snapshot.limitReset, + ], + ); + } + + async recordReceived(job: ReceivedJob): Promise { + const promptPreview = job.publicPreview + ? redactPublicPreview( + job.messages.map((message) => `${message.role}: ${String(message.content)}`).join(" "), + this.previewMaxChars, + ) + : null; + await this.pool.query( + `INSERT INTO reseller_jobs ( + escrow_ref, status, provider, provider_model, marketplace_model, + request_json, public_preview, prompt_preview, balance_before_nanos, + worst_case_cost_nanos, price_lovelace + ) VALUES ($1,'received',$2,$3,$4,$5::jsonb,$6,$7,$8,$9,$10)`, + [ + job.escrowRef, + job.provider, + job.providerModel, + job.marketplaceModel, + JSON.stringify(job.messages), + job.publicPreview, + promptPreview, + job.balanceBeforeUsdNanos.toString(), + job.worstCaseCostUsdNanos.toString(), + job.priceLovelace.toString(), + ], + ); + } + + async recordClaimed( + escrowRef: string, + claimTxHash: string, + localJobId: string, + ): Promise { + await this.transition( + escrowRef, + ["received"], + `status='claimed', claim_tx_hash=$2, local_job_id=$3`, + [claimTxHash, localJobId], + ); + } + + async recordInferenceStarted(escrowRef: string): Promise { + await this.transition(escrowRef, ["claimed"], `status='running'`, []); + } + + async recordOutput(escrowRef: string, output: RecordedOutput): Promise { + const outputPreview = await this.publicPreviewEnabled(escrowRef) + ? redactPublicPreview(output.outputText, this.previewMaxChars) + : null; + await this.transition( + escrowRef, + ["running"], + `response_json=$2::jsonb, receipt_json=$3::jsonb, + prompt_tokens=$4, completion_tokens=$5, actual_cost_nanos=$6, + output_preview=$7`, + [ + JSON.stringify(output.response), + JSON.stringify(output.receipt), + output.promptTokens, + output.completionTokens, + output.actualCostUsdNanos?.toString() ?? null, + outputPreview, + ], + false, + ); + } + + async recordSubmitting(escrowRef: string, submittedRef: string): Promise { + await this.transition( + escrowRef, + ["running"], + `status='submitting', submitted_ref=$2`, + [submittedRef], + ); + } + + async recordSubmitConfirmed(escrowRef: string): Promise { + await this.transition( + escrowRef, + ["submitting"], + `status='submitted'`, + [], + ); + } + + async recordSettled( + escrowRef: string, + balanceAfterUsdNanos: bigint | null, + ): Promise { + await this.transition( + escrowRef, + ["submitted"], + `status='settled', balance_after_nanos=$2, settled_at=now()`, + [balanceAfterUsdNanos?.toString() ?? null], + ); + } + + async recordFailed(escrowRef: string, reason: string): Promise { + await this.pool.query( + `UPDATE reseller_jobs + SET status='failed', failure_reason=$2, updated_at=now() + WHERE escrow_ref=$1 AND status <> 'settled'`, + [escrowRef, reason.slice(0, 1000)], + ); + } + + async listSubmitting(): Promise { + const result = await this.pool.query<{ + escrow_ref: string; + submitted_ref: string; + }>( + `SELECT escrow_ref, submitted_ref FROM reseller_jobs + WHERE status='submitting' AND submitted_ref IS NOT NULL`, + ); + return result.rows.map((row) => ({ + escrowRef: row.escrow_ref, + submittedRef: row.submitted_ref, + })); + } + + async listSubmitted(): Promise { + const result = await this.pool.query<{ + escrow_ref: string; + submitted_ref: string; + }>( + `SELECT escrow_ref, submitted_ref FROM reseller_jobs + WHERE status='submitted' AND submitted_ref IS NOT NULL`, + ); + return result.rows.map((row) => ({ + escrowRef: row.escrow_ref, + submittedRef: row.submitted_ref, + })); + } + + async listPublicJobs(limit: number): Promise { + const safeLimit = Math.max(1, Math.min(100, Math.trunc(limit))); + const result = await this.pool.query<{ + escrow_ref: string; + status: "settled" | "failed"; + provider: string; + provider_model: string; + marketplace_model: string; + prompt_preview: string | null; + output_preview: string | null; + actual_cost_nanos: string | null; + price_lovelace: string; + claim_tx_hash: string | null; + submitted_ref: string | null; + failure_reason: string | null; + created_at: Date; + settled_at: Date | null; + }>( + `SELECT escrow_ref,status,provider,provider_model,marketplace_model, + prompt_preview,output_preview,actual_cost_nanos,price_lovelace, + claim_tx_hash,submitted_ref,failure_reason,created_at,settled_at + FROM reseller_jobs WHERE status IN ('settled','failed') + ORDER BY created_at DESC LIMIT $1`, + [safeLimit], + ); + return result.rows.map((row) => ({ + escrow_ref: row.escrow_ref, + status: row.status, + provider: row.provider, + provider_model: row.provider_model, + marketplace_model: row.marketplace_model, + prompt_preview: row.prompt_preview, + output_preview: row.output_preview, + upstream_cost_usd: row.actual_cost_nanos === null + ? null + : formatUsdNanos(BigInt(row.actual_cost_nanos)), + ap3x_payout: row.status === "settled" + ? formatAp3x(BigInt(row.price_lovelace)) + : "0.000000", + claim_tx_hash: row.claim_tx_hash, + submitted_ref: row.submitted_ref, + failure_reason: row.failure_reason, + created_at: row.created_at.toISOString(), + settled_at: row.settled_at?.toISOString() ?? null, + })); + } + + async publicTotals(): Promise { + const result = await this.pool.query<{ + settled_jobs: string; + failed_jobs: string; + spend_nanos: string; + earned_lovelace: string; + }>( + `SELECT + count(*) FILTER (WHERE status='settled')::text AS settled_jobs, + count(*) FILTER (WHERE status='failed')::text AS failed_jobs, + coalesce(sum(actual_cost_nanos),0)::text AS spend_nanos, + coalesce(sum(price_lovelace) FILTER (WHERE status='settled'),0)::text AS earned_lovelace + FROM reseller_jobs`, + ); + const row = result.rows[0]; + return { + settled_jobs: Number(row?.settled_jobs ?? 0), + failed_jobs: Number(row?.failed_jobs ?? 0), + upstream_spend_usd: formatUsdNanos(BigInt(row?.spend_nanos ?? 0)), + ap3x_earned: formatAp3x(BigInt(row?.earned_lovelace ?? 0)), + }; + } + + private async publicPreviewEnabled(escrowRef: string): Promise { + const result = await this.pool.query<{ public_preview: boolean }>( + "SELECT public_preview FROM reseller_jobs WHERE escrow_ref=$1", + [escrowRef], + ); + return result.rows[0]?.public_preview === true; + } + + private async transition( + escrowRef: string, + expected: string[], + assignment: string, + values: unknown[], + requireStatusChange = true, + ): Promise { + const parameters = [escrowRef, ...values]; + const expectedPlaceholders = expected + .map((_, index) => `$${parameters.length + index + 1}`) + .join(","); + parameters.push(...expected); + const result = await this.pool.query( + `UPDATE reseller_jobs SET ${assignment}, updated_at=now() + WHERE escrow_ref=$1 AND status IN (${expectedPlaceholders})`, + parameters, + ); + if (result.rowCount !== 1 && requireStatusChange) { + throw new Error( + `invalid reseller job transition for ${escrowRef}; expected ${expected.join("|")}`, + ); + } + if (result.rowCount !== 1 && !requireStatusChange) { + throw new Error(`reseller job not running: ${escrowRef}`); + } + } +} + +export function redactPublicPreview(value: string, maxChars: number): string { + const redacted = value + .replace( + /-----BEGIN [^-]+-----[\s\S]*?-----END [^-]+-----/g, + "[private material]", + ) + .replace( + /["']?\b[A-Z0-9_]*(?:PASSWORD|PASSWD|SECRET|TOKEN|API[_-]?KEY|PRIVATE[_-]?KEY|PRIV[_-]?KEY)\b["']?\s*[:=]\s*(?:"(?:\\.|[^"])*"|'(?:\\.|[^'])*'|[^\s,}]+)/gi, + "[secret assignment]", + ) + .replace(/\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b/g, "[jwt]") + .replace(/\b(?:gh[opsu]_|github_pat_)[A-Za-z0-9_]{16,}\b/g, "[secret]") + .replace(/\b[0-9a-fA-F]{64}\b/g, "[64-hex secret]") + .replace(/\b(?:sk|key)-[A-Za-z0-9_-]{8,}\b/g, "[secret]") + .replace(/\bBearer\s+\S+/gi, "Bearer [secret]") + .replace(/\bBasic\s+[A-Za-z0-9+/=]+/gi, "Basic [secret]") + .replace(/\b(?:xox[baprs]-|AKIA)[A-Za-z0-9_-]{12,}\b/g, "[secret]") + .replace(/\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b/gi, "[email]") + .replace(/https?:\/\/\S+/gi, "[url]") + .replace(/\s+/g, " ") + .trim(); + return redacted.length <= maxChars + ? redacted + : `${redacted.slice(0, Math.max(0, maxChars - 1))}…`; +} + +function formatAp3x(lovelace: bigint): string { + const whole = lovelace / 1_000_000n; + const fraction = (lovelace % 1_000_000n).toString().padStart(6, "0"); + return `${whole}.${fraction}`; +} + diff --git a/supplier/src/reseller/money.ts b/supplier/src/reseller/money.ts new file mode 100644 index 0000000..16f315c --- /dev/null +++ b/supplier/src/reseller/money.ts @@ -0,0 +1,54 @@ +export const USD_NANOS = 1_000_000_000n; + +export type DecimalRounding = "floor" | "ceil"; + +/** Parse a non-negative decimal USD string into integer nanodollars. */ +export function parseUsdDecimal( + raw: string, + rounding: DecimalRounding, +): bigint { + const match = /^(?:0|[1-9]\d*)(?:\.(\d+))?$/.exec(raw); + if (!match) throw new Error(`invalid USD decimal: ${raw}`); + + const dot = raw.indexOf("."); + const wholeRaw = dot === -1 ? raw : raw.slice(0, dot); + const fractionRaw = dot === -1 ? "" : raw.slice(dot + 1); + const kept = fractionRaw.slice(0, 9).padEnd(9, "0"); + let nanos = BigInt(wholeRaw) * USD_NANOS + BigInt(kept || "0"); + if (rounding === "ceil" && /[1-9]/.test(fractionRaw.slice(9))) { + nanos += 1n; + } + return nanos; +} + +/** + * Convert a provider JSON number to a conservative remaining-balance value. + * One nanodollar is removed to protect against binary floating-point rounding. + */ +export function parseUsdNumberFloor(value: unknown, field: string): bigint { + if (typeof value !== "number" || !Number.isFinite(value) || value < 0) { + throw new Error(`${field} must be a non-negative finite number`); + } + const scaled = BigInt(Math.floor(value * Number(USD_NANOS))); + return scaled > 0n ? scaled - 1n : 0n; +} + +/** Convert a provider JSON cost to nanodollars without rounding down. */ +export function parseUsdNumberCeil(value: unknown, field: string): bigint { + if (typeof value !== "number" || !Number.isFinite(value) || value < 0) { + throw new Error(`${field} must be a non-negative finite number`); + } + if (value === 0) return 0n; + return BigInt(Math.ceil(value * Number(USD_NANOS))) + 1n; +} + +export function formatUsdNanos(value: bigint, decimals = 6): string { + if (value < 0n) throw new Error("USD value must be non-negative"); + if (!Number.isInteger(decimals) || decimals < 0 || decimals > 9) { + throw new Error("USD display decimals must be from 0 to 9"); + } + const whole = value / USD_NANOS; + if (decimals === 0) return whole.toString(); + const fraction = (value % USD_NANOS).toString().padStart(9, "0"); + return `${whole}.${fraction.slice(0, decimals)}`; +} diff --git a/supplier/src/reseller/openRouterCapacity.ts b/supplier/src/reseller/openRouterCapacity.ts new file mode 100644 index 0000000..14ad251 --- /dev/null +++ b/supplier/src/reseller/openRouterCapacity.ts @@ -0,0 +1,189 @@ +import type { + CapacityProvider, + ProviderCapacityReading, + ProviderModelPricing, +} from "./provider.js"; +import { parseUsdDecimal, parseUsdNumberFloor } from "./money.js"; + +interface OpenRouterCapacityOptions { + baseUrl: string; + apiKey: string; + timeoutMs: number; + pricingCacheMs?: number; + fetchFn?: typeof globalThis.fetch; + now?: () => number; +} + +export class OpenRouterCapacityProvider implements CapacityProvider { + readonly id = "openrouter"; + + private readonly apiRoot: string; + private readonly apiKey: string; + private readonly timeoutMs: number; + private readonly pricingCacheMs: number; + private readonly fetchFn: typeof globalThis.fetch; + private readonly now: () => number; + private pricingCache: ProviderModelPricing | null = null; + + constructor(options: OpenRouterCapacityOptions) { + this.apiRoot = `${options.baseUrl.replace(/\/+$/, "")}/v1`; + this.apiKey = options.apiKey; + this.timeoutMs = options.timeoutMs; + this.pricingCacheMs = options.pricingCacheMs ?? 300_000; + this.fetchFn = options.fetchFn ?? globalThis.fetch; + this.now = options.now ?? Date.now; + } + + async readCapacity(model: string): Promise { + const [key, pricing] = await Promise.all([ + this.readKeyLimit(), + this.readModelPricing(model), + ]); + return { + provider: this.id, + keyLimitUsdNanos: key.limit, + keyRemainingUsdNanos: key.remaining, + limitReset: key.reset, + pricing, + checkedAtMs: this.now(), + }; + } + + private async readKeyLimit(): Promise<{ + limit: bigint; + remaining: bigint; + reset: string | null; + }> { + const body = await this.fetchJson(`${this.apiRoot}/key`); + const data = readRecord(body, "OpenRouter key response").data; + const key = readRecord(data, "OpenRouter key response.data"); + if (key.limit === null || key.limit_remaining === null) { + throw new Error("OpenRouter resale key must have a finite spending limit"); + } + const limit = parseUsdNumberFloor(key.limit, "OpenRouter key limit"); + const remaining = parseUsdNumberFloor( + key.limit_remaining, + "OpenRouter key limit_remaining", + ); + const reset = key.limit_reset; + if (reset !== null && typeof reset !== "string") { + throw new Error("OpenRouter key limit_reset must be a string or null"); + } + return { limit, remaining, reset }; + } + + private async readModelPricing(model: string): Promise { + const cached = this.pricingCache; + const now = this.now(); + if ( + cached && + cached.model === model && + now - cached.fetchedAtMs < this.pricingCacheMs + ) { + return cached; + } + + const body = await this.fetchJson(`${this.apiRoot}/models`); + const root = readRecord(body, "OpenRouter models response"); + if (!Array.isArray(root.data)) { + throw new Error("OpenRouter models response.data must be an array"); + } + const rawModel = root.data.find( + (entry) => + typeof entry === "object" && + entry !== null && + (entry as Record).id === model, + ); + if (!rawModel) throw new Error(`OpenRouter model not found: ${model}`); + const modelRecord = rawModel as Record; + const pricing = readRecord(modelRecord.pricing, `OpenRouter pricing for ${model}`); + const prompt = readPrice(pricing.prompt, `${model} prompt price`); + const completion = readPrice( + pricing.completion, + `${model} completion price`, + ); + const request = pricing.request === undefined + ? 0n + : readPrice(pricing.request, `${model} request price`); + + const result: ProviderModelPricing = { + model, + promptUsdNanosPerToken: prompt, + completionUsdNanosPerToken: completion, + requestUsdNanos: request, + fetchedAtMs: now, + }; + this.pricingCache = result; + return result; + } + + private async fetchJson(url: string): Promise { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), this.timeoutMs); + try { + const response = await this.fetchFn(url, { + headers: { authorization: `Bearer ${this.apiKey}` }, + signal: controller.signal, + }); + const text = await readBoundedBody(response, 8 * 1024 * 1024, controller); + if (!response.ok) { + throw new Error( + `OpenRouter capacity request returned HTTP ${response.status}${ + text ? `: ${text.slice(0, 200)}` : "" + }`, + ); + } + try { + return JSON.parse(text); + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + throw new Error(`OpenRouter capacity response was not JSON: ${detail}`); + } + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + throw new Error(`OpenRouter capacity request failed: ${detail}`); + } finally { + clearTimeout(timer); + } + } +} + +async function readBoundedBody( + response: Response, + maxBytes: number, + controller: AbortController, +): Promise { + if (!response.body) return ""; + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + const parts: string[] = []; + let bytes = 0; + while (true) { + const chunk = await reader.read(); + if (chunk.done) break; + bytes += chunk.value.byteLength; + if (bytes > maxBytes) { + controller.abort(); + throw new Error(`OpenRouter capacity response exceeded ${maxBytes} bytes`); + } + parts.push(decoder.decode(chunk.value, { stream: true })); + } + parts.push(decoder.decode()); + return parts.join(""); +} + +function readRecord(value: unknown, label: string): Record { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + throw new Error(`${label} must be an object`); + } + return value as Record; +} + +function readPrice(value: unknown, label: string): bigint { + if (typeof value !== "string") throw new Error(`${label} must be a string`); + try { + return parseUsdDecimal(value, "ceil"); + } catch { + throw new Error(`${label} must be a non-negative decimal`); + } +} diff --git a/supplier/src/reseller/provider.ts b/supplier/src/reseller/provider.ts new file mode 100644 index 0000000..886aa3d --- /dev/null +++ b/supplier/src/reseller/provider.ts @@ -0,0 +1,22 @@ +export interface ProviderModelPricing { + model: string; + promptUsdNanosPerToken: bigint; + completionUsdNanosPerToken: bigint; + requestUsdNanos: bigint; + fetchedAtMs: number; +} + +export interface ProviderCapacityReading { + provider: string; + keyLimitUsdNanos: bigint; + keyRemainingUsdNanos: bigint; + limitReset: string | null; + pricing: ProviderModelPricing; + checkedAtMs: number; +} + +/** Provider-specific balance and price discovery. Inference stays OpenAI-compatible. */ +export interface CapacityProvider { + readonly id: string; + readCapacity(model: string): Promise; +} diff --git a/supplier/src/reseller/runtime.ts b/supplier/src/reseller/runtime.ts new file mode 100644 index 0000000..9651eea --- /dev/null +++ b/supplier/src/reseller/runtime.ts @@ -0,0 +1,331 @@ +import type { ChainProvider, OutputReference } from "@marketplace/shared/chain"; +import type { ChatMessage } from "@marketplace/shared/tx"; +import type { AdvertDatum } from "@marketplace/shared/cbor"; +import type { SupplierState } from "../state.js"; +import { CapacityGate, type CapacitySnapshot } from "./capacityGate.js"; +import { + ResellerEvidenceStore, + type PublicResellerJob, + type RecordedOutput, +} from "./evidenceStore.js"; +import { formatUsdNanos } from "./money.js"; + +interface ResellerRuntimeOptions { + chain: ChainProvider; + state: SupplierState; + gate: CapacityGate; + store: ResellerEvidenceStore; + advert: AdvertDatum; + advertRef: OutputReference; + providerModel: string; + pollIntervalMs: number; + settlementPollMs?: number; +} + +export interface PublicResellerState { + provider: string; + provider_model: string; + marketplace_model: string; + advert_ref: string; + supplier_pkh: string; + status: "free" | "working" | "offline"; + reason: string; + capacity: { + remaining_allowance_usd: string; + protected_reserve_usd: string; + sellable_usd: string; + committed_usd: string; + worst_case_job_usd: string; + available_jobs: string | null; + checked_at: string | null; + limit_reset: string | null; + }; + totals: { + settled_jobs: number; + failed_jobs: number; + upstream_spend_usd: string; + ap3x_earned: string; + }; +} + +export class ResellerRuntime { + readonly maxInputTokens: number; + + private readonly chain: ChainProvider; + private readonly state: SupplierState; + private readonly gate: CapacityGate; + private readonly store: ResellerEvidenceStore; + private readonly advert: AdvertDatum; + private readonly advertRef: OutputReference; + private readonly providerModel: string; + private readonly pollIntervalMs: number; + private readonly settlementPollMs: number; + private capacityTimer: NodeJS.Timeout | null = null; + private readonly settlementTimers = new Map(); + private readonly submissionTimers = new Map(); + + constructor(options: ResellerRuntimeOptions & { maxInputTokens: number }) { + this.chain = options.chain; + this.state = options.state; + this.gate = options.gate; + this.store = options.store; + this.advert = options.advert; + this.advertRef = options.advertRef; + this.providerModel = options.providerModel; + this.pollIntervalMs = options.pollIntervalMs; + this.settlementPollMs = options.settlementPollMs ?? 5_000; + this.maxInputTokens = options.maxInputTokens; + } + + async initialize(): Promise { + await this.store.initialize(); + await this.store.markInterruptedJobs(); + await this.refreshCapacity(); + this.capacityTimer = setInterval(() => { + void this.refreshCapacity().catch(() => undefined); + }, this.pollIntervalMs); + this.capacityTimer.unref(); + + for (const row of await this.store.listSubmitting()) { + this.trackSubmissionConfirmation(row.escrowRef, row.submittedRef); + } + for (const row of await this.store.listSubmitted()) { + this.trackSettlement(row.escrowRef, row.submittedRef); + } + } + + async close(): Promise { + clearInterval(this.capacityTimer ?? undefined); + this.capacityTimer = null; + for (const timer of this.settlementTimers.values()) clearInterval(timer); + this.settlementTimers.clear(); + for (const timer of this.submissionTimers.values()) clearInterval(timer); + this.submissionTimers.clear(); + await this.store.close(); + } + + capacitySnapshot(): CapacitySnapshot { + return this.gate.snapshot(); + } + + effectiveStatus(): "free" | "working" | "offline" { + const base = this.state.snapshot().status; + if (base === "working") return "working"; + if (base === "offline") return "offline"; + return this.gate.snapshot().canServe ? "free" : "offline"; + } + + async refreshCapacity(): Promise { + try { + await this.store.ping(); + const snapshot = await this.gate.refresh(); + await this.store.saveCapacity(snapshot); + return snapshot; + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + this.gate.forceUnavailable("database_unavailable", detail); + throw error; + } + } + + async recordReceived( + escrowRef: string, + messages: ChatMessage[], + publicPreview: boolean, + ): Promise { + const snapshot = this.gate.snapshot(); + await this.databaseOperation(() => this.store.recordReceived({ + escrowRef, + provider: snapshot.provider, + providerModel: this.providerModel, + marketplaceModel: this.advert.model, + messages, + publicPreview, + balanceBeforeUsdNanos: snapshot.remainingAllowanceUsdNanos, + worstCaseCostUsdNanos: snapshot.worstCaseJobUsdNanos, + priceLovelace: this.advert.price_lovelace, + })); + } + + async recordClaimed( + escrowRef: string, + claimTxHash: string, + localJobId: string, + ): Promise { + await this.databaseOperation( + () => this.store.recordClaimed(escrowRef, claimTxHash, localJobId), + ); + } + + async recordInferenceStarted(escrowRef: string): Promise { + await this.databaseOperation(() => this.store.recordInferenceStarted(escrowRef)); + } + + async recordOutput( + escrowRef: string, + output: Omit & { + actualCostUsdNanos?: bigint | null; + }, + ): Promise { + const snapshot = this.gate.snapshot(); + const calculatedCost = + snapshot.requestPriceUsdNanos + + snapshot.promptPriceUsdNanosPerToken * BigInt(output.promptTokens) + + snapshot.completionPriceUsdNanosPerToken * BigInt(output.completionTokens); + await this.databaseOperation(() => this.store.recordOutput(escrowRef, { + ...output, + actualCostUsdNanos: output.actualCostUsdNanos ?? calculatedCost, + })); + } + + async recordSubmitting(escrowRef: string, submittedRef: string): Promise { + await this.databaseOperation( + () => this.store.recordSubmitting(escrowRef, submittedRef), + ); + } + + async recordSubmitConfirmed( + escrowRef: string, + submittedRef: string, + ): Promise { + await this.databaseOperation( + () => this.store.recordSubmitConfirmed(escrowRef), + ); + this.trackSettlement(escrowRef, submittedRef); + } + + watchSubmitting(escrowRef: string, submittedRef: string): void { + this.trackSubmissionConfirmation(escrowRef, submittedRef); + } + + async recordFailed(escrowRef: string, reason: string): Promise { + await this.databaseOperation(() => this.store.recordFailed(escrowRef, reason)); + } + + async publicState(): Promise { + try { + await this.store.ping(); + const totals = await this.store.publicTotals(); + const capacity = this.gate.snapshot(); + const base = this.state.snapshot(); + const status = this.effectiveStatus(); + const committed = base.status === "working" + ? capacity.worstCaseJobUsdNanos + : 0n; + return { + provider: capacity.provider, + provider_model: this.providerModel, + marketplace_model: this.advert.model, + advert_ref: `${this.advertRef.txHash}#${this.advertRef.index}`, + supplier_pkh: this.advert.supplier_pkh, + status, + reason: status === "working" ? "working" : capacity.reason, + capacity: { + remaining_allowance_usd: formatUsdNanos(capacity.remainingAllowanceUsdNanos), + protected_reserve_usd: formatUsdNanos(capacity.reserveUsdNanos), + sellable_usd: formatUsdNanos(capacity.sellableUsdNanos), + committed_usd: formatUsdNanos(committed), + worst_case_job_usd: formatUsdNanos(capacity.worstCaseJobUsdNanos), + available_jobs: capacity.availableJobs?.toString() ?? null, + checked_at: capacity.checkedAtMs === null + ? null + : new Date(capacity.checkedAtMs).toISOString(), + limit_reset: capacity.limitReset, + }, + totals, + }; + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + this.gate.forceUnavailable("database_unavailable", detail); + throw error; + } + } + + async publicJobs(limit: number): Promise { + try { + return await this.store.listPublicJobs(limit); + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + this.gate.forceUnavailable("database_unavailable", detail); + throw error; + } + } + + + private async databaseOperation(operation: () => Promise): Promise { + try { + return await operation(); + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + this.gate.forceUnavailable("database_unavailable", detail); + throw error; + } + } + + private trackSubmissionConfirmation( + escrowRef: string, + submittedRef: string, + ): void { + if (this.submissionTimers.has(escrowRef)) return; + const ref = parseOutputReference(submittedRef); + if (!ref) return; + let running = false; + const check = async (): Promise => { + if (running) return; + running = true; + try { + const utxo = await this.chain.queryUtxo(ref); + if (utxo === null) return; + await this.recordSubmitConfirmed(escrowRef, submittedRef); + clearInterval(this.submissionTimers.get(escrowRef)); + this.submissionTimers.delete(escrowRef); + } catch { + // A later poll retries. Provider inference is never replayed. + } finally { + running = false; + } + }; + const timer = setInterval(() => void check(), this.settlementPollMs); + timer.unref(); + this.submissionTimers.set(escrowRef, timer); + void check(); + } + + private trackSettlement(escrowRef: string, submittedRef: string): void { + if (this.settlementTimers.has(escrowRef)) return; + const ref = parseOutputReference(submittedRef); + if (!ref) return; + let running = false; + const check = async (): Promise => { + if (running) return; + running = true; + try { + const utxo = await this.chain.queryUtxo(ref); + if (utxo !== null) return; + await this.refreshCapacity().catch(() => this.gate.snapshot()); + const balance = this.gate.snapshot().error === null + ? this.gate.snapshot().remainingAllowanceUsdNanos + : null; + await this.databaseOperation( + () => this.store.recordSettled(escrowRef, balance), + ); + clearInterval(this.settlementTimers.get(escrowRef)); + this.settlementTimers.delete(escrowRef); + } catch { + // A later poll retries chain or database recovery. Inference is never replayed. + } finally { + running = false; + } + }; + const timer = setInterval(() => void check(), this.settlementPollMs); + timer.unref(); + this.settlementTimers.set(escrowRef, timer); + void check(); + } +} + +function parseOutputReference(value: string): OutputReference | null { + const match = /^([0-9a-fA-F]{64})#(0|[1-9]\d*)$/.exec(value); + if (!match) return null; + return { txHash: match[1], index: Number(match[2]) }; +} diff --git a/supplier/src/reseller/setupConfig.ts b/supplier/src/reseller/setupConfig.ts new file mode 100644 index 0000000..d9efb9c --- /dev/null +++ b/supplier/src/reseller/setupConfig.ts @@ -0,0 +1,176 @@ +import { readFile } from "node:fs/promises"; + +export interface ResellerSetupConfig { + networkId: 1; + ogmiosUrl: string; + endpointUrl: string; + marketplaceModel: string; + openRouterModel: string; + maxInputTokens: number; + maxOutputTokens: number; + maxProcessingMs: number; + inferenceTimeoutMs: number; + priceLovelace: bigint; + supplierBondLovelace: bigint; + buyerBondLovelace: bigint; + reserveUsd: string; + pollIntervalMs: number; + providerTimeoutMs: number; + previewMaxChars: number; + minimumWalletLovelace: bigint; + fundingTimeoutMs: number; + composeFile: string; +} + +export async function loadResellerSetupConfig( + path: string, +): Promise { + let parsed: unknown; + try { + parsed = JSON.parse(await readFile(path, "utf8")); + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + throw new Error(`unable to read reseller config ${path}: ${detail}`); + } + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) { + throw new Error("reseller config must be a JSON object"); + } + const value = parsed as Record; + const networkId = integerField(value, "networkId"); + if (networkId !== 1) throw new Error("reseller config networkId must be 1"); + + const endpoint = new URL(urlField(value, "endpointUrl", ["https:"])); + if ( + endpoint.username !== "" || + endpoint.password !== "" || + endpoint.search !== "" || + endpoint.hash !== "" || + endpoint.pathname.replace(/\/+$/, "") !== "" + ) { + throw new Error( + "reseller config endpointUrl must be a plain HTTPS origin without credentials, path, query, or fragment", + ); + } + const endpointUrl = endpoint.origin; + + const previewMaxChars = optionalInteger(value, "previewMaxChars", 160); + if (previewMaxChars > 1000) { + throw new Error("reseller config previewMaxChars must not exceed 1000"); + } + const providerTimeoutMs = optionalInteger(value, "providerTimeoutMs", 10_000); + const maxProcessingMs = integerField(value, "maxProcessingMs"); + const settlementMarginMs = 150_000 + providerTimeoutMs; + const defaultInferenceTimeoutMs = maxProcessingMs - settlementMarginMs; + if (defaultInferenceTimeoutMs <= 0) { + throw new Error( + "reseller config maxProcessingMs does not leave time for provider checks and chain settlement", + ); + } + const inferenceTimeoutMs = value.inferenceTimeoutMs === undefined + ? defaultInferenceTimeoutMs + : integerField(value, "inferenceTimeoutMs"); + if (inferenceTimeoutMs + settlementMarginMs > maxProcessingMs) { + throw new Error( + "reseller config inferenceTimeoutMs does not leave time for provider checks and chain settlement", + ); + } + + + return { + networkId: 1, + ogmiosUrl: urlField(value, "ogmiosUrl", ["http:", "https:"]), + endpointUrl, + marketplaceModel: stringField(value, "marketplaceModel"), + openRouterModel: stringField(value, "openRouterModel"), + maxInputTokens: integerField(value, "maxInputTokens"), + maxOutputTokens: integerField(value, "maxOutputTokens"), + maxProcessingMs, + inferenceTimeoutMs, + priceLovelace: bigintField(value, "priceLovelace"), + supplierBondLovelace: bigintField(value, "supplierBondLovelace"), + buyerBondLovelace: bigintField(value, "buyerBondLovelace"), + reserveUsd: usdField(value, "reserveUsd"), + pollIntervalMs: optionalInteger(value, "pollIntervalMs", 30_000), + providerTimeoutMs, + previewMaxChars, + minimumWalletLovelace: optionalBigint(value, "minimumWalletLovelace", 20_000_000n), + fundingTimeoutMs: optionalInteger(value, "fundingTimeoutMs", 3_600_000), + composeFile: stringField(value, "composeFile"), + }; +} + +function stringField(value: Record, name: string): string { + const field = value[name]; + if (typeof field !== "string" || field.trim() === "") { + throw new Error(`reseller config ${name} must be a non-empty string`); + } + if (/[\r\n]/.test(field)) { + throw new Error(`reseller config ${name} must be one line`); + } + return field.trim(); +} + +function urlField( + value: Record, + name: string, + protocols: string[], +): string { + const raw = stringField(value, name); + let url: URL; + try { + url = new URL(raw); + } catch { + throw new Error(`reseller config ${name} must be a valid URL`); + } + if (!protocols.includes(url.protocol)) { + throw new Error( + `reseller config ${name} must use ${protocols.join(" or ")}`, + ); + } + return raw; +} + +function integerField(value: Record, name: string): number { + const field = value[name]; + if (!Number.isSafeInteger(field) || (field as number) <= 0) { + throw new Error(`reseller config ${name} must be a positive safe integer`); + } + return field as number; +} + +function optionalInteger( + value: Record, + name: string, + fallback: number, +): number { + return value[name] === undefined ? fallback : integerField(value, name); +} + +function bigintField(value: Record, name: string): bigint { + const field = value[name]; + if (typeof field !== "string" || !/^[1-9]\d*$/.test(field)) { + throw new Error(`reseller config ${name} must be a positive integer string`); + } + return BigInt(field); +} + +function optionalBigint( + value: Record, + name: string, + fallback: bigint, +): bigint { + return value[name] === undefined ? fallback : bigintField(value, name); +} + +function usdField(value: Record, name: string): string { + const field = value[name]; + if ( + typeof field !== "string" || + !/^(?:0|[1-9]\d*)(?:\.\d{1,9})?$/.test(field) + ) { + throw new Error( + `reseller config ${name} must be a non-negative decimal string`, + ); + } + return field; +} diff --git a/supplier/src/server.ts b/supplier/src/server.ts index bf42e55..fd80b60 100644 --- a/supplier/src/server.ts +++ b/supplier/src/server.ts @@ -38,6 +38,7 @@ import type { ChainProvider, OutputReference } from "@marketplace/shared/chain"; import { decodeAdvertDatum, decodeEscrowDatum, canonicalize } from "@marketplace/shared/cbor"; import type { AdvertDatum, EscrowDatum } from "@marketplace/shared/cbor"; import type { ChatMessage, WalletKey } from "@marketplace/shared/tx"; +import { chatInputTokenUpperBound } from "@marketplace/shared/tx"; import { buildClaimTx, chatSessionPromptHash, @@ -62,6 +63,9 @@ import { endChatSession, type EndChatSessionDeps } from "./chatSessionRunner.js" import { callOpenAiStream } from "./openai.js"; import { healthzRouter } from "./routes/healthz.js"; import { triggerOnFailureConsolidate } from "./walletHealth.js"; +import type { ResellerRuntime } from "./reseller/runtime.js"; +import { formatUsdNanos } from "./reseller/money.js"; +import { renderResellerDashboardPage } from "./reseller/dashboardPage.js"; export interface SupplierDeps { chain: ChainProvider; @@ -72,6 +76,8 @@ export interface SupplierDeps { jobs?: JobStore; /** Optional: defaults to a fresh ChatSessionStore (chat-session capability). */ chatSessions?: ChatSessionStore; + /** Optional balance-aware resale runtime. Absent for every legacy supplier. */ + reseller?: ResellerRuntime; } const ESCROW_REF_RE = /^[0-9a-fA-F]{64}#(?:0|[1-9]\d*)$/; @@ -102,6 +108,7 @@ interface ResolvedDeps { supplierKey: WalletKey; jobs: JobStore; chatSessions: ChatSessionStore; + reseller?: ResellerRuntime; } function makeCapabilityHandler(deps: ResolvedDeps) { @@ -138,6 +145,9 @@ function makeCapabilityHandler(deps: ResolvedDeps) { supplier_pkh: datum.supplier_pkh, // SPEC FIX 2026-04-25: pub_key_hex required for buyer-side receipt verification pub_key_hex: deps.supplierKey.pubKeyHex, + ...(deps.reseller + ? { max_input_tokens: deps.reseller.maxInputTokens } + : {}), }); } catch (err) { next(err); @@ -151,8 +161,9 @@ function makeCapabilityHandler(deps: ResolvedDeps) { function makeStatusHandler(deps: ResolvedDeps) { return (_req: Request, res: Response) => { const snap = deps.state.snapshot(); + const effectiveStatus = deps.reseller?.effectiveStatus() ?? snap.status; const payload: Record = { - status: snap.status, + status: effectiveStatus, last_seen: snap.lastSeenIso, active_sessions: snap.activeSessions, max_sessions: snap.maxSessions, @@ -160,6 +171,17 @@ function makeStatusHandler(deps: ResolvedDeps) { if (snap.status === "working" && snap.currentEscrowRef) { payload.current_escrow_ref = snap.currentEscrowRef; } + if (deps.reseller) { + const capacity = deps.reseller.capacitySnapshot(); + payload.capacity = { + provider: capacity.provider, + reason: effectiveStatus === "working" ? "working" : capacity.reason, + sellable_usd: formatUsdNanos(capacity.sellableUsdNanos), + checked_at: capacity.checkedAtMs === null + ? null + : new Date(capacity.checkedAtMs).toISOString(), + }; + } return res.status(200).json(payload); }; } @@ -202,6 +224,8 @@ function makeChatHandler(deps: ResolvedDeps) { // Slot held by THIS request (null until acquired / after hand-off to the // job runner) — the catch-all must only release its own acquisition. let acquiredRef: string | null = null; + let evidenceRecorded = false; + let evidenceEscrowRef: string | null = null; try { // ── 1. Header validation ────────────────────────────────────────── const headerVal = req.header("X-Escrow-Ref"); @@ -214,6 +238,7 @@ function makeChatHandler(deps: ResolvedDeps) { 'X-Escrow-Ref must match "<64-hex>#"'); } const escrowRefStr = `${escrowRef.txHash}#${escrowRef.index}`; + evidenceEscrowRef = escrowRefStr; // ── 2. Body shape validation ───────────────────────────────────── const body = (req.body ?? {}) as ChatBody; @@ -240,6 +265,18 @@ function makeChatHandler(deps: ResolvedDeps) { } } const validatedMessages = messages as Array<{ role: "system" | "user" | "assistant"; content: string }>; + if (deps.reseller) { + const inputUnits = chatInputTokenUpperBound(validatedMessages); + if (inputUnits > deps.reseller.maxInputTokens) { + return jsonError( + res, + 413, + "input_cap_exceeded", + `input bound ${inputUnits} exceeds supplier cap ${deps.reseller.maxInputTokens}`, + ); + } + } + // ── 3. Resolve advert ──────────────────────────────────────────── const advertResult = await fetchActiveAdvert(deps); @@ -314,6 +351,28 @@ function makeChatHandler(deps: ResolvedDeps) { return jsonError(res, 408, "past_deliver_by", `now ${nowMs} >= deliver_by ${escrowDatum.deliver_by}`); } + if (deps.reseller) { + let capacity; + try { + capacity = await deps.reseller.refreshCapacity(); + } catch { + return jsonError( + res, + 503, + "capacity_unavailable", + "provider capacity or evidence database is unavailable", + ); + } + if (!capacity.canServe) { + return jsonError( + res, + 503, + "capacity_unavailable", + `provider capacity is unavailable (${capacity.reason})`, + ); + } + } + // ── 8. Acquire session slot ────────────────────────────────────── if (!deps.state.tryAcquire(escrowRefStr)) { @@ -322,6 +381,26 @@ function makeChatHandler(deps: ResolvedDeps) { acquiredRef = escrowRefStr; // From here on, every error path MUST release the slot until the // runner takes ownership of it. + if (deps.reseller) { + try { + await deps.reseller.recordReceived( + escrowRefStr, + validatedMessages, + req.header("X-Vector-Public-Preview") === "1", + ); + evidenceRecorded = true; + } catch { + deps.state.release(escrowRefStr); + acquiredRef = null; + return jsonError( + res, + 503, + "evidence_unavailable", + "reseller evidence database is unavailable", + ); + } + } + // ── 9+10. Claim tx: build + confirm as ONE wallet critical section ─ // (splitting them would let another wallet spend coin-select mid-flight) @@ -343,7 +422,12 @@ function makeChatHandler(deps: ResolvedDeps) { } return { kind: "ok" as const, built }; }); + const failEvidence = async (reason: string): Promise => { + if (!evidenceRecorded || !deps.reseller) return; + await deps.reseller.recordFailed(escrowRefStr, reason).catch(() => undefined); + }; if (claimOutcome.kind === "build_failed") { + await failEvidence("claim_submit_failed"); deps.state.release(escrowRefStr); acquiredRef = null; // Backstop the periodic wallet-health ticker: most Claim build failures @@ -359,6 +443,7 @@ function makeChatHandler(deps: ResolvedDeps) { `Claim tx submit failed: ${(claimOutcome.err as Error).message}`); } if (claimOutcome.kind === "await_failed") { + await failEvidence("claim_confirmation_timeout"); deps.state.release(escrowRefStr); acquiredRef = null; return jsonError(res, 504, "claim_timeout", @@ -372,6 +457,32 @@ function makeChatHandler(deps: ResolvedDeps) { index: 0, }; const jobId = deps.jobs.create(escrowRefStr); + if (deps.reseller) { + try { + await deps.reseller.recordClaimed( + escrowRefStr, + claimResult.expectedTxHash, + jobId, + ); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + deps.jobs.fail(jobId, { + httpStatus: 503, + reason: "evidence_unavailable", + message, + }); + deps.state.release(escrowRefStr); + acquiredRef = null; + await failEvidence("claim_record_failed"); + return jsonError( + res, + 503, + "evidence_unavailable", + "Claim confirmed but evidence persistence failed", + ); + } + } + // Slot release now happens inside runChatJob's finally. acquiredRef = null; @@ -382,6 +493,7 @@ function makeChatHandler(deps: ResolvedDeps) { config: deps.config, supplierKey: deps.supplierKey, jobs: deps.jobs, + reseller: deps.reseller, }, jobId, escrowRef: escrowRefStr, @@ -401,6 +513,11 @@ function makeChatHandler(deps: ResolvedDeps) { if (acquiredRef) { try { deps.state.release(acquiredRef); } catch { /* ignore */ } } + if (evidenceRecorded && deps.reseller && evidenceEscrowRef) { + await deps.reseller + .recordFailed(evidenceEscrowRef, err instanceof Error ? err.message : String(err)) + .catch(() => undefined); + } next(err); return; } @@ -1383,6 +1500,7 @@ export function createApp(deps: SupplierDeps): Application { supplierKey: deps.supplierKey, jobs: deps.jobs ?? new JobStore(), chatSessions: deps.chatSessions ?? new ChatSessionStore(), + reseller: deps.reseller, }; const app = express(); @@ -1396,6 +1514,41 @@ export function createApp(deps: SupplierDeps): Application { // /healthz is mounted FIRST and uses no deps — it must succeed regardless // of chain/state/config (independent of /status free/working/offline). app.use(healthzRouter()); + if (resolved.reseller) { + app.get("/reseller", (_req: Request, res: Response) => { + res.setHeader("Cache-Control", "no-store"); + res.type("html").send(renderResellerDashboardPage()); + }); + app.get("/reseller/api/status", async (_req: Request, res: Response) => { + res.setHeader("Cache-Control", "no-store"); + try { + res.status(200).json(await resolved.reseller!.publicState()); + } catch { + jsonError( + res, + 503, + "evidence_unavailable", + "reseller evidence database is unavailable", + ); + } + }); + app.get("/reseller/api/jobs", async (req: Request, res: Response) => { + res.setHeader("Cache-Control", "no-store"); + const parsed = Number(req.query.limit ?? 50); + const limit = Number.isSafeInteger(parsed) ? parsed : 50; + try { + res.status(200).json({ jobs: await resolved.reseller!.publicJobs(limit) }); + } catch { + jsonError( + res, + 503, + "evidence_unavailable", + "reseller evidence database is unavailable", + ); + } + }); + } + app.get("/capability", makeCapabilityHandler(resolved)); app.get("/status", makeStatusHandler(resolved)); diff --git a/tests/fixtures/supplier-side/sample-config.ts b/tests/fixtures/supplier-side/sample-config.ts index 9e8e980..b6ea0e7 100644 --- a/tests/fixtures/supplier-side/sample-config.ts +++ b/tests/fixtures/supplier-side/sample-config.ts @@ -34,6 +34,7 @@ export function buildSampleConfig(): SupplierConfig { maxChatSessions: 1, chatSettleMode: "full", liveChain: false, + reseller: null, }; } diff --git a/tests/unit/buyer-sdk-submit-prompt-reject.test.ts b/tests/unit/buyer-sdk-submit-prompt-reject.test.ts index 56ae10c..7f7ea64 100644 --- a/tests/unit/buyer-sdk-submit-prompt-reject.test.ts +++ b/tests/unit/buyer-sdk-submit-prompt-reject.test.ts @@ -30,6 +30,7 @@ import type { AdvertDatum, EscrowDatum } from "../../packages/shared/src/cbor/ty import type { Utxo, OutputReference } from "../../packages/shared/src/chain/ChainProvider.js"; import type { ChatMessage } from "../../packages/shared/src/tx/types.js"; import type { ProgressEvent } from "../../buyer/src/sdk/types.js"; +import { BOUNDED_INPUT_DETAIL_MARKER } from "../../packages/shared/src/tx/inputBound.js"; // ─── Fixtures ───────────────────────────────────────────────────────────────── @@ -473,6 +474,43 @@ describe("Marketplace.submitPrompt() — rejection paths", () => { expect(e.name).toBe("ReceiptVerificationError"); expect(e.reason).toBe("test reason"); }); + it("rejects a bounded-input request before posting escrow", async () => { + const advert = makeActiveAdvert({ + detail_uri: `https://supplier.example.com/reseller${BOUNDED_INPUT_DETAIL_MARKER}`, + }); + seedAdvertUtxo(chain, advert); + const submitSpy = vi.spyOn(chain, "submitTx"); + fetchSpy.mockImplementation((url: unknown) => { + if (String(url).endsWith("/capability")) { + return jsonFetch({ + capability_id: advert.capability_id, + model: advert.model, + max_output_tokens: advert.max_output_tokens, + max_processing_ms: advert.max_processing_ms, + max_input_tokens: 1, + price_lovelace: advert.price_lovelace.toString(), + advert_ref: `${ADVERT_REF.txHash}#${ADVERT_REF.index}`, + supplier_pkh: advert.supplier_pkh, + pub_key_hex: supplier.pubKeyHex, + }); + } + return jsonFetch({}); + }); + const mp = makeMarketplace(chain, fetchSpy); + await expect( + mp.submitPrompt({ + advertRef: ADVERT_REF, + messages: SAMPLE_MESSAGES, + payment_lovelace: PAYMENT, + }), + ).rejects.toSatisfy( + (error: unknown) => + error instanceof TxConstructionError && + error.reason === "input_cap_exceeded", + ); + expect(submitSpy).not.toHaveBeenCalled(); + }); + }); // ─── M1-F-1: isSyncThrow dead-code removal (RED) ───────────────────────────── diff --git a/tests/unit/reseller-capacity.test.ts b/tests/unit/reseller-capacity.test.ts new file mode 100644 index 0000000..e3224a1 --- /dev/null +++ b/tests/unit/reseller-capacity.test.ts @@ -0,0 +1,151 @@ +import { describe, expect, it, vi } from "vitest"; +import { chatInputTokenUpperBound } from "../../packages/shared/src/tx/inputBound.js"; +import { CapacityGate } from "../../supplier/src/reseller/capacityGate.js"; +import type { + CapacityProvider, + ProviderCapacityReading, +} from "../../supplier/src/reseller/provider.js"; +import { OpenRouterCapacityProvider } from "../../supplier/src/reseller/openRouterCapacity.js"; + +function reading(remaining: bigint): ProviderCapacityReading { + return { + provider: "openrouter", + keyLimitUsdNanos: 10_000_000_000n, + keyRemainingUsdNanos: remaining, + limitReset: "monthly", + checkedAtMs: 1_000, + pricing: { + model: "provider/model", + promptUsdNanosPerToken: 1_000_000n, + completionUsdNanosPerToken: 2_000_000n, + requestUsdNanos: 0n, + fetchedAtMs: 1_000, + }, + }; +} + +describe("chatInputTokenUpperBound", () => { + it("uses canonical UTF-8 bytes and fixed chat overhead", () => { + expect(chatInputTokenUpperBound([])).toBe(5); + expect(chatInputTokenUpperBound([{ role: "user", content: "a" }])).toBe(42); + expect(chatInputTokenUpperBound([{ role: "user", content: "é" }])).toBe(43); + expect(chatInputTokenUpperBound([{ role: "user", content: "🙂" }])).toBe(45); + }); + + it("adds each message independently", () => { + expect(chatInputTokenUpperBound([ + { role: "system", content: "x" }, + { role: "user", content: "yz" }, + ])).toBe(83); + }); + + it("counts tool-call arguments present on direct SDK messages", () => { + expect(chatInputTokenUpperBound([{ + role: "assistant", + content: "", + tool_calls: [{ + id: "x", + type: "function", + function: { name: "f", arguments: "a".repeat(1000) }, + }], + }])).toBe(1129); + }); +}); + +describe("CapacityGate", () => { + it("opens only when allowance above reserve covers a worst-case job", async () => { + let current = reading(2_000_000_000n); + const provider: CapacityProvider = { + id: "openrouter", + readCapacity: vi.fn(async () => current), + }; + const gate = new CapacityGate({ + provider, + model: "provider/model", + reserveUsd: "1", + maxInputTokens: 100, + maxOutputTokens: 10, + staleAfterMs: 5_000, + now: () => 1_000, + }); + + const open = await gate.refresh(); + expect(open.canServe).toBe(true); + expect(open.worstCaseJobUsdNanos).toBe(120_000_000n); + expect(open.sellableUsdNanos).toBe(1_000_000_000n); + expect(open.availableJobs).toBe(8n); + + current = reading(1_100_000_000n); + const closed = await gate.refresh(); + expect(closed.canServe).toBe(false); + expect(closed.reason).toBe("insufficient_for_job"); + }); + + it("fails closed when the provider check fails", async () => { + const provider: CapacityProvider = { + id: "openrouter", + readCapacity: vi.fn(async () => { + throw new Error("provider down"); + }), + }; + const gate = new CapacityGate({ + provider, + model: "provider/model", + reserveUsd: "0", + maxInputTokens: 10, + maxOutputTokens: 10, + staleAfterMs: 5_000, + }); + const snapshot = await gate.refresh(); + expect(snapshot.canServe).toBe(false); + expect(snapshot.reason).toBe("provider_unavailable"); + }); +}); + +describe("OpenRouterCapacityProvider", () => { + it("reads a capped key and rounds model prices up", async () => { + const fetchFn = vi.fn(async (input: string | URL | Request) => { + const url = String(input); + if (url.endsWith("/v1/key")) { + return new Response(JSON.stringify({ + data: { limit: 5, limit_remaining: 4.5, limit_reset: "monthly" }, + }), { status: 200 }); + } + return new Response(JSON.stringify({ + data: [{ + id: "provider/model", + pricing: { prompt: "0.00000007952", completion: "0.00000015904" }, + }], + }), { status: 200 }); + }); + const provider = new OpenRouterCapacityProvider({ + baseUrl: "https://openrouter.ai/api", + apiKey: "secret", + timeoutMs: 1_000, + fetchFn: fetchFn as typeof fetch, + now: () => 123, + }); + const result = await provider.readCapacity("provider/model"); + expect(result.keyRemainingUsdNanos).toBe(4_499_999_999n); + expect(result.pricing.promptUsdNanosPerToken).toBe(80n); + expect(result.pricing.completionUsdNanosPerToken).toBe(160n); + }); + + it("rejects an uncapped inference key", async () => { + const fetchFn = vi.fn(async (input: string | URL | Request) => { + if (String(input).endsWith("/v1/key")) { + return new Response(JSON.stringify({ + data: { limit: null, limit_remaining: null, limit_reset: null }, + }), { status: 200 }); + } + return new Response(JSON.stringify({ data: [] }), { status: 200 }); + }); + const provider = new OpenRouterCapacityProvider({ + baseUrl: "https://openrouter.ai/api", + apiKey: "secret", + timeoutMs: 1_000, + fetchFn: fetchFn as typeof fetch, + }); + await expect(provider.readCapacity("provider/model")).rejects.toThrow(/finite/); + }); +}); diff --git a/tests/unit/reseller-redaction.test.ts b/tests/unit/reseller-redaction.test.ts new file mode 100644 index 0000000..0cfa96e --- /dev/null +++ b/tests/unit/reseller-redaction.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from "vitest"; +import { redactPublicPreview } from "../../supplier/src/reseller/evidenceStore.js"; + +describe("redactPublicPreview", () => { + it("removes common credentials and personal data before publication", () => { + const preview = redactPublicPreview( + [ + "SUPPLIER_PRIV_KEY_HEX=" + "a".repeat(64), + "password=hunter2", + "Authorization: Bearer secret-token", + '{"password":"hunter two words"}', + "GATEWAY_API_KEY=plain-secret-value", + "Authorization: Basic dXNlcjpwYXNz", + "email test@example.com", + "https://example.com/private", + "eyJheader.payload.signature", + "ghp_1234567890abcdefghijklmnop", + ].join(" "), + 1000, + ); + expect(preview).not.toContain("hunter2"); + expect(preview).not.toContain("secret-token"); + expect(preview).not.toContain("hunter two words"); + expect(preview).not.toContain("plain-secret-value"); + expect(preview).not.toContain("dXNlcjpwYXNz"); + expect(preview).not.toContain("test@example.com"); + expect(preview).not.toContain("example.com/private"); + expect(preview).not.toContain("eyJheader"); + expect(preview).not.toContain("ghp_"); + expect(preview).not.toContain("a".repeat(64)); + }); + + it("applies the configured public preview limit", () => { + const preview = redactPublicPreview("x".repeat(100), 20); + expect(preview).toHaveLength(20); + expect(preview.endsWith("…")).toBe(true); + }); +}); diff --git a/tests/unit/reseller-setup-config.test.ts b/tests/unit/reseller-setup-config.test.ts new file mode 100644 index 0000000..3e3f994 --- /dev/null +++ b/tests/unit/reseller-setup-config.test.ts @@ -0,0 +1,68 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { loadResellerSetupConfig } from "../../supplier/src/reseller/setupConfig.js"; + +const tempDirs: string[] = []; + +async function configPath(overrides: Record = {}): Promise { + const dir = await mkdtemp(join(tmpdir(), "reseller-config-")); + tempDirs.push(dir); + const path = join(dir, "config.json"); + await writeFile(path, JSON.stringify({ + networkId: 1, + ogmiosUrl: "https://ogmios.vector.mainnet.apexfusion.org", + endpointUrl: "https://mp-suppliers-openrouter-reseller.vector.apexfusion.org", + marketplaceModel: "vector/resold-model", + openRouterModel: "provider/model", + maxInputTokens: 4096, + maxOutputTokens: 256, + maxProcessingMs: 600000, + inferenceTimeoutMs: 180000, + priceLovelace: "200000", + supplierBondLovelace: "1000000", + buyerBondLovelace: "1000000", + reserveUsd: "1.00", + composeFile: "deploy/mainnet/docker-compose.supplier-openrouter-reseller.yml", + ...overrides, + })); + return path; +} + +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { recursive: true }))); +}); + +describe("loadResellerSetupConfig", () => { + it("accepts a bounded mainnet setup", async () => { + const config = await loadResellerSetupConfig(await configPath()); + expect(config.inferenceTimeoutMs).toBe(180000); + expect(config.endpointUrl).toBe( + "https://mp-suppliers-openrouter-reseller.vector.apexfusion.org", + ); + }); + + it("rejects unsupported Ogmios transports", async () => { + await expect(loadResellerSetupConfig(await configPath({ + ogmiosUrl: "wss://ogmios.example", + }))).rejects.toThrow(/ogmiosUrl/); + }); + + it("rejects endpoint credentials, query data, and fragments", async () => { + for (const endpointUrl of [ + "https://user:secret@example.com", + "https://example.com?token=secret", + "https://example.com#secret", + ]) { + await expect(loadResellerSetupConfig(await configPath({ endpointUrl }))) + .rejects.toThrow(/plain HTTPS origin/); + } + }); + + it("requires settlement time outside the inference timeout", async () => { + await expect(loadResellerSetupConfig(await configPath({ + inferenceTimeoutMs: 500000, + }))).rejects.toThrow(/does not leave time/); + }); +}); diff --git a/tests/unit/supplier-config.test.ts b/tests/unit/supplier-config.test.ts index 4cc5081..6963817 100644 --- a/tests/unit/supplier-config.test.ts +++ b/tests/unit/supplier-config.test.ts @@ -331,3 +331,62 @@ describe("loadConfig() — OPENAI_SESSION_PASSTHROUGH", () => { expect(cfg.openaiModelOverride).toBe("openclaw"); }); }); + +describe("loadConfig() — OpenRouter reseller", () => { + const resellerEnv = () => ({ + ...buildSampleEnv(), + CAPABILITY_KIND: "chat", + LLM_BACKEND: "openai", + OPENAI_BASE_URL: "https://openrouter.ai/api", + OPENAI_API_KEY: "seller-only-key", + OPENAI_MAX_TOKENS: "256", + RESELLER_PROVIDER: "openrouter", + RESELLER_DATABASE_URL: "postgresql://reseller:secret@postgres:5432/reseller", + RESELLER_RESERVE_USD: "1.25", + RESELLER_MAX_INPUT_TOKENS: "4096", + }); + + it("stays disabled for every legacy config", () => { + expect(loadConfig(buildSampleEnv()).reseller).toBeNull(); + }); + + it("parses a complete balance-aware config", () => { + const cfg = loadConfig(resellerEnv()); + expect(cfg.reseller).toMatchObject({ + provider: "openrouter", + reserveUsd: "1.25", + maxInputTokens: 4096, + pollIntervalMs: 30_000, + providerTimeoutMs: 10_000, + }); + }); + + it("rejects partial and unsafe configurations", () => { + expect(() => loadConfig({ + ...buildSampleEnv(), + RESELLER_RESERVE_USD: "1", + })).toThrow(/RESELLER_PROVIDER/); + expect(() => loadConfig({ + ...resellerEnv(), + OPENAI_BASE_URL: "https://example.com", + })).toThrow(/OPENAI_BASE_URL/); + for (const OPENAI_BASE_URL of [ + "http://openrouter.ai/api", + "https://openrouter.ai/v1", + "https://user:secret@openrouter.ai/api", + ]) { + expect(() => loadConfig({ + ...resellerEnv(), + OPENAI_BASE_URL, + })).toThrow(/OPENAI_BASE_URL/); + } + expect(() => loadConfig({ + ...resellerEnv(), + OPENAI_MAX_TOKENS: "", + })).toThrow(/OPENAI_MAX_TOKENS/); + expect(() => loadConfig({ + ...resellerEnv(), + RESELLER_RESERVE_USD: "-1", + })).toThrow(/RESELLER_RESERVE_USD/); + }); +}); diff --git a/tests/unit/supplier-reseller-routes.test.ts b/tests/unit/supplier-reseller-routes.test.ts new file mode 100644 index 0000000..7202a3b --- /dev/null +++ b/tests/unit/supplier-reseller-routes.test.ts @@ -0,0 +1,78 @@ +import { describe, expect, it, vi } from "vitest"; +import request from "supertest"; +import { MockChainProvider } from "../../packages/shared/src/chain/MockChainProvider.js"; +import { createApp } from "../../supplier/src/server.js"; +import { SupplierState } from "../../supplier/src/state.js"; +import type { ResellerRuntime } from "../../supplier/src/reseller/runtime.js"; +import { buildSampleConfig } from "../fixtures/supplier-side/sample-config.js"; +import { buildSupplierWalletKey } from "../fixtures/supplier-side/wallet-keys.js"; + +const ESCROW_REF = `${"f".repeat(64)}#0`; + +function fakeReseller(canServe: boolean, maxInputTokens = 100): ResellerRuntime { + const snapshot = { + provider: "openrouter", + model: "provider/model", + canServe, + reason: canServe ? "ready" as const : "insufficient_for_job" as const, + checkedAtMs: 1_000, + lastSuccessAtMs: 1_000, + keyLimitUsdNanos: 5_000_000_000n, + remainingAllowanceUsdNanos: 2_000_000_000n, + reserveUsdNanos: 1_000_000_000n, + sellableUsdNanos: 1_000_000_000n, + worstCaseJobUsdNanos: 100_000_000n, + promptPriceUsdNanosPerToken: 1n, + completionPriceUsdNanosPerToken: 1n, + requestPriceUsdNanos: 0n, + availableJobs: 10n, + limitReset: "monthly", + error: null, + }; + return { + maxInputTokens, + effectiveStatus: vi.fn(() => canServe ? "free" : "offline"), + capacitySnapshot: vi.fn(() => snapshot), + refreshCapacity: vi.fn(async () => snapshot), + } as unknown as ResellerRuntime; +} + +describe("reseller supplier routes", () => { + it("reports offline without retiring an exhausted supplier", async () => { + const app = createApp({ + chain: new MockChainProvider(), + state: new SupplierState(), + config: buildSampleConfig(), + supplierKey: buildSupplierWalletKey(), + reseller: fakeReseller(false), + }); + const response = await request(app).get("/status"); + expect(response.status).toBe(200); + expect(response.body).toMatchObject({ + status: "offline", + capacity: { + provider: "openrouter", + reason: "insufficient_for_job", + }, + }); + }); + + it("rejects oversized input before any chain submission", async () => { + const chain = new MockChainProvider(); + const submitSpy = vi.spyOn(chain, "submitTx"); + const app = createApp({ + chain, + state: new SupplierState(), + config: buildSampleConfig(), + supplierKey: buildSupplierWalletKey(), + reseller: fakeReseller(true, 1), + }); + const response = await request(app) + .post("/v1/chat/completions") + .set("X-Escrow-Ref", ESCROW_REF) + .send({ messages: [{ role: "user", content: "too large" }] }); + expect(response.status).toBe(413); + expect(response.body.reason).toBe("input_cap_exceeded"); + expect(submitSpy).not.toHaveBeenCalled(); + }); +}); diff --git a/tests/unit/tx-submit.test.ts b/tests/unit/tx-submit.test.ts index 0878bac..b8522ed 100644 --- a/tests/unit/tx-submit.test.ts +++ b/tests/unit/tx-submit.test.ts @@ -104,6 +104,28 @@ describe("buildSubmitTx() — happy path", () => { const newUtxo = await chain.queryUtxo({ txHash: result.expectedTxHash, index: 0 }); expect(newUtxo!.lovelace).toBe(TOTAL_LOCKED); }); + it("journals signed Submit CBOR before broadcast", async () => { + const supplier = buildSupplierWalletKey(); + const utxo = buildClaimedEscrowUtxo(); + let hookCalled = false; + const result = await buildSubmitTx({ + chain, + supplierKey: supplier, + escrowRef: utxo.ref, + receiptHash: VALID_RECEIPT_HASH, + beforeSubmit: async (built) => { + hookCalled = true; + expect( + await chain.queryUtxo({ txHash: built.expectedTxHash, index: 0 }), + ).toBeNull(); + }, + }); + expect(hookCalled).toBe(true); + expect( + await chain.queryUtxo({ txHash: result.expectedTxHash, index: 0 }), + ).not.toBeNull(); + }); + }); // ─── Rejection: double-submit ────────────────────────────────────────────────