feat: add balance-aware provider resale agent

This commit is contained in:
david
2026-08-27 09:54:53 +02:00
parent 8baf127c77
commit e54eb55f2b
41 changed files with 3579 additions and 38 deletions

View File

@@ -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<string, unknown>;
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<SubmitPromptResult> {
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;

View File

@@ -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;
}
/**

View File

@@ -128,7 +128,7 @@ done
drain_supplier() { # drain_supplier <compose-file>
[ "$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

View File

@@ -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

View File

@@ -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"
}

View File

@@ -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:*",

View File

@@ -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<string, string> = {};
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<Record<string, unknown>> {
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<string, unknown> | 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<PublicJob> {
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<void>((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<string, string | undefined>,
): Promise<number> {
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;
});
}

View File

@@ -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,

View File

@@ -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);

View File

@@ -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;
}
const stream = b.stream === true;
return { model, messages, maxTokens, stream, tools, toolChoice };
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, publicPreview };
}

View File

@@ -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") {

View File

@@ -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<void>;
}
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;
}

View File

@@ -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<void>;
}
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 };

View File

@@ -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";

View File

@@ -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
);
}

View File

@@ -101,6 +101,7 @@ export interface LiveSubmitParams {
deliverBy: number;
tipMs: number;
receiptHash: string;
beforeSubmit?: (built: BuildResult) => Promise<void>;
}
export interface LiveAcceptParams {
@@ -136,6 +137,7 @@ export interface LivePostAdvertParams {
walletKey: WalletKey;
advertDatum: AdvertDatum;
deposit_lovelace: bigint;
beforeSubmit?: (built: PostAdvertBuildResult) => Promise<void>;
}
export interface LiveRetireAdvertParams {
@@ -499,6 +501,7 @@ async function buildSpendOpenOrClaimedTx(opts: {
validFromMs: number;
validToMs: number;
signerPkh: string;
beforeSubmit?: (built: BuildResult) => Promise<void>;
}): Promise<BuildResult> {
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<Bu
validFromMs: tipMs,
validToMs: stampMs,
signerPkh: params.supplierKey.pubKeyHash,
beforeSubmit: params.beforeSubmit,
});
}
@@ -910,14 +915,16 @@ export async function buildLiveTxForAdvert(
const txCborHex = signed.toCBOR();
const expectedTxHash = signed.toHash();
await chain.submitTx(txCborHex);
return {
const built: PostAdvertBuildResult = {
txCborHex,
expectedTxHash,
advertOutputRef: { txHash: expectedTxHash, index: 0 },
};
await params.beforeSubmit?.(built);
await chain.submitTx(txCborHex);
return built;
}
// ─── RetireAdvert ────────────────────────────────────────────────────

116
pnpm-lock.yaml generated
View File

@@ -321,10 +321,16 @@ importers:
express:
specifier: ^4.21.2
version: 4.22.1
pg:
specifier: ^8.16.3
version: 8.23.0
devDependencies:
'@types/express':
specifier: ^5.0.3
version: 5.0.6
'@types/pg':
specifier: ^8.15.5
version: 8.23.1
'@types/supertest':
specifier: ^6.0.2
version: 6.0.3
@@ -1244,6 +1250,9 @@ packages:
'@types/pdf-parse@1.1.5':
resolution: {integrity: sha512-kBfrSXsloMnUJOKi25s3+hRmkycHfLK6A09eRGqF/N8BkQoPUmaCr+q8Cli5FnfohEz/rsv82zAiPz/LXtOGhA==}
'@types/pg@8.23.1':
resolution: {integrity: sha512-fKVHpikPdg4GKks3JuLEhvwSyvwzF23hnabPy6DD8ljVbC7+6J5dQzdv4arV6jqq57djnMgs1HKBxX4P8aBI3A==}
'@types/prop-types@15.7.15':
resolution: {integrity: sha512-F6bEyamV9jKGAFBEmlQnesRPGOQqS2+Uwi0Em15xenOxHaf2hv6L8YCVn3rPdPJOiJfPiCnLIRyvwVaqMY3MIw==}
@@ -2360,6 +2369,40 @@ packages:
resolution: {integrity: sha512-XRIRcLgk6ZnUbsHsYXExMw+krrPE81hJ6FQPLdBNhhBefqIQKXu/WeTgNBGSwPrfU0v+UCEwn7AoAUOsVKHFvQ==}
engines: {node: '>=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: {}

View File

@@ -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"

View File

@@ -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<void>;
}
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}`);

View File

@@ -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<string, string> {
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<string, string | undefined>,
log: (line: string) => void = (line) => console.error(line),
): Promise<SetupResult> {
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<void>,
): Promise<OutputReference> {
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<void> {
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<void>((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<Record<string, string>> {
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<string, string> = {};
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<string, string>): Promise<void> {
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<void> {
const args = [
"compose",
"-f",
resolve(REPO_ROOT, composeFile),
"up",
"-d",
"--build",
"--wait",
];
await new Promise<void>((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<string, string | undefined>,
): Promise<number> {
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;
});
}

View File

@@ -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<string, string | undefined>, name: string): string {
@@ -532,6 +569,121 @@ export function loadConfig(env: Record<string, string | undefined>): 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<string, string | undefined>): SupplierCon
revisionProbeMode,
liveChain,
walletHealthIntervalMs,
reseller,
};
}

View File

@@ -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<void> {
const cfg = loadConfig(process.env);
@@ -45,8 +51,74 @@ async function main(): Promise<void> {
// 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<void> {
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"));

View File

@@ -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<void> {
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<void> {
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<void> {
reason: collapsedReason,
message,
});
await recordResellerFailure(deps.reseller, escrowRef, collapsedReason);
return;
}
@@ -146,8 +179,39 @@ export async function runChatJob(params: RunChatJobParams): Promise<void> {
});
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<void> {
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<void> {
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<void> {
};
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);

View File

@@ -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<OpenAiResult
const usageRaw = obj.usage;
let promptTokens = 0;
let completionTokens = 0;
let costUsd: number | undefined;
if (usageRaw && typeof usageRaw === "object") {
const usage = usageRaw as Record<string, unknown>;
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<OpenAiResult
prompt_tokens: promptTokens,
completion_tokens: completionTokens,
wallclock_ms: wallclockMs,
...(costUsd === undefined ? {} : { cost_usd: costUsd }),
};
}

View File

@@ -0,0 +1,179 @@
import type { CapacityProvider } from "./provider.js";
import { parseUsdDecimal } from "./money.js";
export type CapacityReason =
| "starting"
| "ready"
| "below_reserve"
| "insufficient_for_job"
| "provider_unavailable"
| "database_unavailable"
| "stale";
export interface CapacitySnapshot {
provider: string;
model: string;
canServe: boolean;
reason: CapacityReason;
checkedAtMs: number | null;
lastSuccessAtMs: number | null;
keyLimitUsdNanos: bigint;
remainingAllowanceUsdNanos: bigint;
reserveUsdNanos: bigint;
sellableUsdNanos: bigint;
worstCaseJobUsdNanos: bigint;
promptPriceUsdNanosPerToken: bigint;
completionPriceUsdNanosPerToken: bigint;
requestPriceUsdNanos: bigint;
availableJobs: bigint | null;
limitReset: string | null;
error: string | null;
}
interface CapacityGateOptions {
provider: CapacityProvider;
model: string;
reserveUsd: string;
maxInputTokens: number;
maxOutputTokens: number;
staleAfterMs: number;
now?: () => 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<CapacitySnapshot> | 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<CapacitySnapshot> {
if (this.inFlight) return this.inFlight;
this.inFlight = this.performRefresh();
try {
return await this.inFlight;
} finally {
this.inFlight = null;
}
}
forceUnavailable(
reason: Extract<CapacityReason, "database_unavailable" | "provider_unavailable">,
error: string,
): CapacitySnapshot {
this.current = {
...this.current,
canServe: false,
reason,
checkedAtMs: this.now(),
error,
};
return this.snapshot();
}
private async performRefresh(): Promise<CapacitySnapshot> {
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();
}
}

View File

@@ -0,0 +1,83 @@
export function renderResellerDashboardPage(): string {
return `<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Reseller capacity</title>
<style>
:root{color-scheme:dark;--bg:#071014;--panel:#0d1a20;--line:#24363d;--text:#eef5f5;--muted:#98aeb5;--good:#50d890;--warn:#ffcb66;--bad:#ff7b72;--accent:#75bfff}
*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 15% 0,#12303a 0,transparent 34rem),var(--bg);color:var(--text);font:14px/1.5 ui-sans-serif,system-ui,-apple-system,BlinkMacSystemFont,"Segoe UI",sans-serif}
main{width:min(1180px,calc(100% - 32px));margin:0 auto;padding:42px 0 64px}header{display:flex;align-items:flex-end;justify-content:space-between;gap:24px;margin-bottom:24px}h1{font-size:clamp(28px,5vw,48px);line-height:1.05;margin:0;letter-spacing:-.04em}header p{margin:8px 0 0;color:var(--muted)}.stamp{color:var(--muted);font-variant-numeric:tabular-nums}.panel{background:color-mix(in srgb,var(--panel) 94%,transparent);border:1px solid var(--line);border-radius:16px;box-shadow:0 16px 48px #0005}.identity{display:grid;grid-template-columns:1.4fr 1fr 1fr;gap:1px;overflow:hidden;margin-bottom:16px}.identity>div{padding:18px 20px;background:var(--panel)}.label{display:block;color:var(--muted);font-size:11px;text-transform:uppercase;letter-spacing:.12em;margin-bottom:6px}.value{font-size:17px;font-weight:650;overflow-wrap:anywhere}.status{display:inline-flex;align-items:center;gap:8px}.dot{width:9px;height:9px;border-radius:50%;background:var(--muted);box-shadow:0 0 0 4px #fff1}.status.free .dot{background:var(--good)}.status.working .dot{background:var(--warn)}.status.offline .dot{background:var(--bad)}.metrics{display:grid;grid-template-columns:repeat(4,1fr);gap:12px;margin:0 0 24px}.metric{padding:18px}.metric strong{display:block;font-size:24px;letter-spacing:-.03em;font-variant-numeric:tabular-nums}.metric small{color:var(--muted)}.section-head{display:flex;justify-content:space-between;align-items:center;margin:28px 0 12px}h2{font-size:18px;margin:0}.table-wrap{overflow:auto}table{width:100%;border-collapse:collapse;min-width:940px}th,td{text-align:left;padding:13px 14px;border-bottom:1px solid var(--line);vertical-align:top}th{color:var(--muted);font-size:11px;text-transform:uppercase;letter-spacing:.09em;font-weight:600}tbody tr:last-child td{border-bottom:0}.job-status{font-weight:700}.job-status.settled{color:var(--good)}.job-status.failed{color:var(--bad)}.mono{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px}.preview{max-width:250px;color:#c9d8dc;white-space:normal}.empty,.error{padding:30px;color:var(--muted)}.error{color:var(--bad)}
@media(max-width:820px){header{align-items:flex-start;flex-direction:column}.identity{grid-template-columns:1fr}.metrics{grid-template-columns:repeat(2,1fr)}}@media(max-width:480px){main{width:min(100% - 20px,1180px);padding-top:24px}.metrics{grid-template-columns:1fr}.metric strong{font-size:21px}}
</style>
</head>
<body>
<main>
<header><div><h1>Provider capacity resale</h1><p>Bounded upstream capacity settled through Vector escrow.</p></div><div class="stamp" id="updated">Loading current state</div></header>
<section class="panel identity" aria-label="Agent identity">
<div><span class="label">Route</span><span class="value" id="route">Loading</span></div>
<div><span class="label">Marketplace model</span><span class="value" id="market-model">Loading</span></div>
<div><span class="label">Supplier status</span><span class="status" id="status"><span class="dot"></span><span id="status-text">Loading</span></span><div class="stamp" id="reason"></div></div>
</section>
<section class="metrics" aria-label="Capacity and settlement metrics">
<div class="panel metric"><span class="label">Sellable allowance</span><strong id="sellable">--</strong><small id="allowance">Provider allowance --</small></div>
<div class="panel metric"><span class="label">Worst-case job</span><strong id="worst-job">--</strong><small id="available-jobs">-- jobs available</small></div>
<div class="panel metric"><span class="label">AP3X earned</span><strong id="earned">--</strong><small id="settled-count">-- settled jobs</small></div>
<div class="panel metric"><span class="label">Upstream spend</span><strong id="spend">--</strong><small id="failed-count">-- failed jobs</small></div>
</section>
<div class="section-head"><h2>Conversion timeline</h2><span class="stamp">Public previews require buyer consent</span></div>
<section class="panel table-wrap"><table><thead><tr><th>Time</th><th>Status</th><th>Model</th><th>Prompt preview</th><th>Output preview</th><th>Upstream</th><th>Payout</th><th>Escrow</th></tr></thead><tbody id="jobs"><tr><td colspan="8" class="empty">Loading jobs</td></tr></tbody></table></section>
</main>
<script>
(function(){
'use strict';
var byId=function(id){return document.getElementById(id)};
var text=function(id,value){byId(id).textContent=value==null?'--':String(value)};
var money=function(value){return '$'+String(value==null?'0':value)};
function renderStatus(state){
text('route',state.provider+' / '+state.provider_model);
text('market-model',state.marketplace_model);
var status=byId('status');status.className='status '+state.status;
text('status-text',state.status);text('reason',state.reason);
text('sellable',money(state.capacity.sellable_usd));
text('allowance','Allowance '+money(state.capacity.remaining_allowance_usd)+' · reserve '+money(state.capacity.protected_reserve_usd)+' · committed '+money(state.capacity.committed_usd));
text('worst-job',money(state.capacity.worst_case_job_usd));
text('available-jobs',state.capacity.available_jobs==null?'No balance limit':state.capacity.available_jobs+' jobs available');
text('earned',state.totals.ap3x_earned+' AP3X');
text('settled-count',state.totals.settled_jobs+' settled jobs');
text('spend',money(state.totals.upstream_spend_usd));
text('failed-count',state.totals.failed_jobs+' failed jobs');
text('updated','Updated '+new Date().toLocaleTimeString());
}
function cell(row,value,className){var td=document.createElement('td');if(className)td.className=className;td.textContent=value==null?'--':String(value);row.appendChild(td)}
function renderJobs(rows){
var body=byId('jobs');body.replaceChildren();
if(!rows.length){var row=document.createElement('tr');cell(row,'No terminal jobs yet','empty');row.firstChild.colSpan=8;body.appendChild(row);return}
rows.forEach(function(job){
var row=document.createElement('tr');
cell(row,new Date(job.created_at).toLocaleString());
cell(row,job.status,'job-status '+job.status);
cell(row,job.marketplace_model,'mono');
cell(row,job.prompt_preview,'preview');
cell(row,job.output_preview,'preview');
cell(row,job.upstream_cost_usd==null?'--':money(job.upstream_cost_usd));
cell(row,job.ap3x_payout+' AP3X');
cell(row,job.escrow_ref,'mono');
body.appendChild(row);
});
}
function showError(message){var body=byId('jobs');body.replaceChildren();var row=document.createElement('tr');cell(row,message,'error');row.firstChild.colSpan=8;body.appendChild(row);text('updated','Data unavailable')}
async function refresh(){
try{
var results=await Promise.all([fetch('/reseller/api/status',{cache:'no-store'}),fetch('/reseller/api/jobs?limit=50',{cache:'no-store'})]);
if(!results[0].ok||!results[1].ok)throw new Error('Seller evidence is unavailable');
var data=await Promise.all([results[0].json(),results[1].json()]);renderStatus(data[0]);renderJobs(data[1].jobs||[]);
}catch(error){showError(error instanceof Error?error.message:'Seller evidence is unavailable')}
}
refresh();setInterval(refresh,5000);
})();
</script>
</body>
</html>`;
}

View File

@@ -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<string, unknown>;
receipt: Record<string, unknown>;
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<void> {
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<void> {
await this.pool.query("SELECT 1");
}
async close(): Promise<void> {
await this.pool.end();
}
async markInterruptedJobs(): Promise<void> {
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<void> {
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<void> {
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<void> {
await this.transition(
escrowRef,
["received"],
`status='claimed', claim_tx_hash=$2, local_job_id=$3`,
[claimTxHash, localJobId],
);
}
async recordInferenceStarted(escrowRef: string): Promise<void> {
await this.transition(escrowRef, ["claimed"], `status='running'`, []);
}
async recordOutput(escrowRef: string, output: RecordedOutput): Promise<void> {
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<void> {
await this.transition(
escrowRef,
["running"],
`status='submitting', submitted_ref=$2`,
[submittedRef],
);
}
async recordSubmitConfirmed(escrowRef: string): Promise<void> {
await this.transition(
escrowRef,
["submitting"],
`status='submitted'`,
[],
);
}
async recordSettled(
escrowRef: string,
balanceAfterUsdNanos: bigint | null,
): Promise<void> {
await this.transition(
escrowRef,
["submitted"],
`status='settled', balance_after_nanos=$2, settled_at=now()`,
[balanceAfterUsdNanos?.toString() ?? null],
);
}
async recordFailed(escrowRef: string, reason: string): Promise<void> {
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<SubmittedJobRow[]> {
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<SubmittedJobRow[]> {
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<PublicResellerJob[]> {
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<PublicResellerTotals> {
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<boolean> {
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<void> {
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}`;
}

View File

@@ -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)}`;
}

View File

@@ -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<ProviderCapacityReading> {
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<ProviderModelPricing> {
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<string, unknown>).id === model,
);
if (!rawModel) throw new Error(`OpenRouter model not found: ${model}`);
const modelRecord = rawModel as Record<string, unknown>;
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<unknown> {
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<string> {
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<string, unknown> {
if (typeof value !== "object" || value === null || Array.isArray(value)) {
throw new Error(`${label} must be an object`);
}
return value as Record<string, unknown>;
}
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`);
}
}

View File

@@ -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<ProviderCapacityReading>;
}

View File

@@ -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<string, NodeJS.Timeout>();
private readonly submissionTimers = new Map<string, NodeJS.Timeout>();
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<void> {
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<void> {
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<CapacitySnapshot> {
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<void> {
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<void> {
await this.databaseOperation(
() => this.store.recordClaimed(escrowRef, claimTxHash, localJobId),
);
}
async recordInferenceStarted(escrowRef: string): Promise<void> {
await this.databaseOperation(() => this.store.recordInferenceStarted(escrowRef));
}
async recordOutput(
escrowRef: string,
output: Omit<RecordedOutput, "actualCostUsdNanos"> & {
actualCostUsdNanos?: bigint | null;
},
): Promise<void> {
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<void> {
await this.databaseOperation(
() => this.store.recordSubmitting(escrowRef, submittedRef),
);
}
async recordSubmitConfirmed(
escrowRef: string,
submittedRef: string,
): Promise<void> {
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<void> {
await this.databaseOperation(() => this.store.recordFailed(escrowRef, reason));
}
async publicState(): Promise<PublicResellerState> {
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<PublicResellerJob[]> {
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<T>(operation: () => Promise<T>): Promise<T> {
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<void> => {
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<void> => {
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]) };
}

View File

@@ -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<ResellerSetupConfig> {
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<string, unknown>;
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<string, unknown>, 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<string, unknown>,
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<string, unknown>, 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<string, unknown>,
name: string,
fallback: number,
): number {
return value[name] === undefined ? fallback : integerField(value, name);
}
function bigintField(value: Record<string, unknown>, 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<string, unknown>,
name: string,
fallback: bigint,
): bigint {
return value[name] === undefined ? fallback : bigintField(value, name);
}
function usdField(value: Record<string, unknown>, 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;
}

View File

@@ -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<string, unknown> = {
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>#<int>"');
}
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<void> => {
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));

View File

@@ -34,6 +34,7 @@ export function buildSampleConfig(): SupplierConfig {
maxChatSessions: 1,
chatSettleMode: "full",
liveChain: false,
reseller: null,
};
}

View File

@@ -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) ─────────────────────────────

View File

@@ -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/);
});
});

View File

@@ -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);
});
});

View File

@@ -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<string, unknown> = {}): Promise<string> {
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/);
});
});

View File

@@ -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/);
});
});

View File

@@ -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();
});
});

View File

@@ -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 ────────────────────────────────────────────────