11 Commits

Author SHA1 Message Date
HalukG
5377d379b7 feat(supplier): HuggingFace Inference Providers router backend preset
Some checks failed
CI / typecheck + vitest (Node 20) (push) Has been cancelled
OpenAI-compatible compute-backend preset (config only, reuses LLM_BACKEND=openai) for pointing a supplier node at the HuggingFace Inference Providers router. Adds a boot guard that rejects the OpenRouter-only reasoning param against the HF router, unit tests, and an operator runbook.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-10 16:51:45 +03:00
david
2f7b0194a5 feat(supplier,gateway,buyer): N-slot chat suppliers + entry-ticket settle mode
Two structural fixes for demo concurrency (3 single-slot suppliers thrashed
under Open WebUI's ~3 conversations per message):

Multi-slot suppliers (MAX_CHAT_SESSIONS, default 1):
- SupplierState is now capacity-aware: a Set of active escrow refs with
  free-until-full status semantics, ref-scoped release(ref) (also fixes the
  latent bug where handler catch-alls released another request's slot), and
  a walletMutex serializing every wallet-spending chain op — Claim, Submit,
  and wallet-health consolidation queue instead of racing the single
  working/collateral UTxO. /status gains active_sessions/max_sessions.

Entry-ticket settle mode (CHAT_SETTLE_MODE=ticket, default full, both sides):
- The buyer's escrow post at session open is the ONLY chain op. The supplier
  verifies the Open escrow as an entry ticket but never Claims/Submits
  (/v1/chat/start → {status:"ticket"}; /v1/chat/end → receipt-less
  {status:"ended"}); no hard-cap timer, so sessions may outlive deliver_by.
  Used tickets can't reopen sessions (records retained; 409 session_exists).
- Buyer SDK: startChat/endChat are settle-mode aware, driven by the
  supplier's response — mixed-mode rollout windows settle correctly.
- Gateway: ticket closes bill cost 0; the sweeper Reclaims Open tickets
  after deliver_by WITHOUT closing live sessions (the conversation and its
  affinity mirror survive their reclaimed ticket), gates reclaims on
  deliver_by, and keeps an in-process settled-refs memory so terminal
  indexer rows aren't retried every sweep forever.

Deployment (kimi-chat trio + gateway): ticket mode, 8 slots per supplier,
idle policy gateway 10 min / supplier 15 min (fixes the inversion where the
supplier idle-closed before the gateway janitor).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-09 20:48:52 +02:00
david
9a4f9f0699 fix(gateway,supplier): demo sessions stick to their supplier; free pinned suppliers
The demo endpoint lost its session on nearly every second message: affinity
required clients to re-echo the assistant reply byte-identically, so clients
that strip <think> blocks or reshape whitespace (Open WebUI, litellm) missed
every turn, opened a fresh escrow session on the next free single-slot
supplier, and left the old one pinned. Open WebUI's background title/tag-gen
requests each pinned another supplier — one message could exhaust all three
kimi replicas ("no available chat supplier").

- affinity: lenient assistant-content matching (think-strip + whitespace
  collapse); user/system/tool messages and tool_calls stay byte-strict;
  strict prefixes outrank lenient ones. The mirror is never rewritten with
  client text, so receipt hashes stay valid.
- demoChat: trim trailing assistant/system messages (regenerate semantics)
  instead of 400 "the last message must have role user or tool".
- supplier: roll the transcript back when the LLM call fails so retries
  don't duplicate the delta (ChatSessionStore.truncateTranscript).
- demoChat: LRU-evict a demo session when no supplier is available (supplier
  /v1/chat/end frees the slot, sweeper Accepts the escrow) and retry the
  open with selectCandidates ignoreStatusFor to bypass the ~20s-stale
  indexer status poll. Max 2 evictions per request.
- sweeper: idle janitor closes demo sessions unused for DEMO_SESSION_IDLE_MS
  (default 3 min, below the supplier's 5 min CHAT_IDLE_TIMEOUT_MS) so
  abandoned one-off sessions release their suppliers quickly.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-09 02:29:10 +02:00
david
3d2b4bb203 chore(buyer): hide Book Summarizer tab from the nav
Route and all PDF summarize code preserved; only the nav link is
commented out.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 10:30:18 +02:00
david
4ebe7c9683 feat(gateway,supplier,shared): shared demo key with session-backed OpenAI endpoint + tool-calling
MVP demo: one operator-funded API key anyone can plug into an IDE/harness,
running real escrow-per-session marketplace conversations (llm.chat.v1)
behind the standard OpenAI /chat/completions wire shape.

- shared: ChatMessage gains tool role + tool_calls; normalizeChatMessage/
  chatMessagesEquivalent keep supplier transcript and gateway mirror
  hash-identical for receipts (field presence is part of the identity)
- supplier: /v1/chat/message accepts {messages,tools,tool_choice} alongside
  legacy {content}; callOpenAiStream forwards tools, accumulates fragmented
  delta.tool_calls, allows content-less tool turns; done frame carries the
  assistant message verbatim + finish_reason + usage
- gateway: demo flag on api_keys (additive migration), create-demo-key CLI,
  withdraw 403 + hidden deposit address for demo keys, per-IP demo rate
  limit (demo keys skip the shared per-key limit), models filtered to chat
  capability
- gateway executor (demoChat.ts): transcript-prefix affinity maps stateless
  calls onto open sessions, forwards only unseen deltas, transparently
  reopens on supplier idle-close; never falls back to one-shot (its settle
  path assumes <=1 in-flight escrow per wallet)
- mutex scope: key mutex held only for chain work (open/close); off-chain
  turns serialize on per-session locks so replicas serve users concurrently
- sweeper: chat-aware reclaim guard (35min vs 11min) and abandoned-session
  cleanup; terminal state confirmed via on-chain queryUtxo (indexer keeps
  lineage rows at their old state forever)
- deploy: kimi-chat replicas 2+3 compose files; chat idle timeout 2.5min

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 10:30:17 +02:00
david
b92bff56a7 feat(supplier): env-gated OpenRouter reasoning-off + max_tokens
The supplier sent a bare {model,messages,stream} to the OpenAI/OpenRouter
endpoint — no max_tokens and no reasoning control — so reasoning ("thinking")
models like kimi-k2.6 emitted reasoning tokens that could consume the
completion budget and finish with reason=length (truncated/empty answer =
failed job). deepseek-v4-flash has no thinking; kimi-k2.6 does by default.

Add two optional, env-gated request params (defaults preserve current behavior,
so ChatMock/DeepSeek-direct suppliers are unaffected):
  - OPENAI_REASONING=off  -> send reasoning:{enabled:false} (disable thinking:
    faster, cheaper, straight answers; no reasoning-starved length truncation)
  - OPENAI_MAX_TOKENS=N   -> forward max_tokens:N. Must stay below the model
    context (kimi context = 262144), so the advert's 262144 cap cannot be sent
    verbatim (provider 400s when prompt+max_tokens > context).

openai.ts builds the body conditionally; config.ts parses the env; wired
through jobRunner.ts (text) and server.ts (chat-session stream).

Deployed on the 4 kimi + 3 deepseek-flash suppliers with OPENAI_REASONING=off
and OPENAI_MAX_TOKENS=131072 (env files are host-only/gitignored).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-02 16:47:30 +02:00
david
ab649c55da feat(supplier): add 3 OpenRouter deepseek-v4-flash suppliers (deepseek-flash-1/2/3)
Add 3 single-slot suppliers serving deepseek/deepseek-v4-flash via OpenRouter,
increasing concurrent capacity for the independent buyer machines. Same shape
as the kimi replicas: capability llm.text.generate.v1, price 200000 lovelace,
max_output_tokens=262144, max_processing_ms=3600000, OPENAI_TIMEOUT_MS=3600000,
sharing the one OpenRouter key (accepted concurrency ceiling; no 429/backoff).

- deploy/mainnet/docker-compose.supplier-deepseek-flash-{1,2,3}.yml (new)
- buyer/scripts/monitor-wallets.ts: register the 3 replica wallets

Per-instance wallets, .env files, and on-chain adverts are provisioned on the
mainnet host (gitignored); the 3 containers are live and Active on-chain.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-02 14:35:55 +02:00
david
99c44442e7 feat(supplier): add 3 kimi replicas + raise all kimi caps to OpenRouter max
Add 3 parallel-capacity replicas of supplier-kimi (kimi-2/3/4) so the
marketplace can serve more concurrent jobs from the independent buyer
machines. Each is a single-slot worker on the same capability
(llm.text.generate.v1) and model (moonshotai/kimi-k2.6) at the same 0.2 AP3X
flat price, sharing the one OpenRouter key (accepted concurrency ceiling).

Also bring the original supplier-kimi up to the same config: raise its advert
caps (re-posted on-chain) and OPENAI_TIMEOUT_MS to match. All four kimi
suppliers now advertise max_output_tokens=262144, max_processing_ms=3600000
(OpenRouter's ceiling for kimi-k2.6), with OPENAI_TIMEOUT_MS=3600000 so the
upstream call isn't cut short before the advertised deadline.

- deploy/mainnet/docker-compose.supplier-kimi-{2,3,4}.yml (new)
- deploy/mainnet/docker-compose.supplier-kimi.yml: OPENAI_TIMEOUT_MS 300000 -> 3600000
- buyer/scripts/monitor-wallets.ts: register the 3 replica wallets

Per-instance wallets, .env files, and on-chain adverts are provisioned on the
mainnet host (gitignored); all four containers are live and Active on-chain.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-02 14:11:55 +02:00
david
9b5e5b5710 feat(gateway,buyer): self-serve "Generate API key" page in the frontend
Add an "API Keys" page to the buyer SPA (marketplace.vector.apexfusion.org)
that mints a custodial gateway API key by POSTing directly to the gateway's
public /signup, then shows the key once + deposit address + a base_url/curl
usage snippet. Page lives behind the SPA's existing operator login; per-user
registration is a later feature.

Gateway:
- GATEWAY_CORS_ORIGINS env -> corsOrigins allowlist; corsMiddleware echoes
  Access-Control-* only for allowlisted Origins, answers OPTIONS preflight
  204, no credentials. Empty allowlist = CORS off (default safe).

Buyer:
- GATEWAY_PUBLIC_URL env -> injected into __BUYER_BOOT__.gatewayUrl (only in
  the authenticated boot block). SPA resolveGatewayUrl() falls back to
  deriving "api." + location.host so the post-login-no-reload case (boot
  block absent) still works.
- New ApiKeys page + /api-keys route + nav link.

Deploy (mainnet compose): gateway GATEWAY_CORS_ORIGINS and buyer
GATEWAY_PUBLIC_URL set to the marketplace/api hosts. NOTE: both images must be
REBUILT (gateway runs tsx from copied source; buyer SPA is vite-built at image
build) — an env-only recreate would silently ship the old code.

Typecheck + vite build clean; 205 tests pass. Adversarial multi-agent review
(4 dimensions, per-finding verification) surfaced only the rebuild-required
deploy note above; no code/security defects.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-22 11:43:02 +02:00
david
f4f35866f5 fix(escrow): reserve pure-ADA collateral UTxO in PostEscrow coin selection
buildLiveTxForEscrow called .complete() with no presetWalletInputs, so lucid was
free to spend the dedicated 5-ADA collateral UTxO as an ordinary input. On
back-to-back requests this left the wallet with no collateral candidate and the
subsequent Accept failed with collateral_required, stranding the escrow.

Reserve the smallest pure-ADA UTxO >= 5 ADA and fund PostEscrow from the rest
(falls back to default selection for single/large-UTxO cases, which keep an
ample pure-ADA change UTxO anyway). Benefits the buyer too (same shared path).
Adds reserveCollateralInputs unit tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-22 11:05:08 +02:00
david
4ba074020c fix(gateway): set mainnet VECTOR_ZERO_TIME_MS + ref-script UTxOs in compose
Without VECTOR_ZERO_TIME_MS the live tx builder defaults to the testnet genesis,
placing PostEscrow validity bounds ~51 days in the future (Ogmios 3118). Escrow
spends (Accept/Reclaim) also need ESCROW_REF_UTXO. Mirror buyer/supplier mainnet env.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-22 10:48:20 +02:00
77 changed files with 5181 additions and 343 deletions

View File

@@ -83,6 +83,48 @@ const OPERATOR_SOURCES: OperatorSource[] = [
keyVar: "SUPPLIER_PRIV_KEY_HEX",
role: "supplier",
},
{
label: "supplier-kimi-2@vector-marketplace",
host: "root@91.98.147.172",
envPath: "/root/agents-marketplace/supplier/.env.kimi-2",
keyVar: "SUPPLIER_PRIV_KEY_HEX",
role: "supplier",
},
{
label: "supplier-kimi-3@vector-marketplace",
host: "root@91.98.147.172",
envPath: "/root/agents-marketplace/supplier/.env.kimi-3",
keyVar: "SUPPLIER_PRIV_KEY_HEX",
role: "supplier",
},
{
label: "supplier-kimi-4@vector-marketplace",
host: "root@91.98.147.172",
envPath: "/root/agents-marketplace/supplier/.env.kimi-4",
keyVar: "SUPPLIER_PRIV_KEY_HEX",
role: "supplier",
},
{
label: "supplier-deepseek-flash-1@vector-marketplace",
host: "root@91.98.147.172",
envPath: "/root/agents-marketplace/supplier/.env.deepseek-flash-1",
keyVar: "SUPPLIER_PRIV_KEY_HEX",
role: "supplier",
},
{
label: "supplier-deepseek-flash-2@vector-marketplace",
host: "root@91.98.147.172",
envPath: "/root/agents-marketplace/supplier/.env.deepseek-flash-2",
keyVar: "SUPPLIER_PRIV_KEY_HEX",
role: "supplier",
},
{
label: "supplier-deepseek-flash-3@vector-marketplace",
host: "root@91.98.147.172",
envPath: "/root/agents-marketplace/supplier/.env.deepseek-flash-3",
keyVar: "SUPPLIER_PRIV_KEY_HEX",
role: "supplier",
},
{
label: "buyer@open-webui",
host: "open-webui",

View File

@@ -36,6 +36,12 @@
* Defaults to "./data/archive" relative to cwd.
* When unset, archive is disabled (responses are
* not persisted; /v1/responses* return 503).
* GATEWAY_PUBLIC_URL — Public base URL of the OpenAI-compatible gateway
* (e.g. https://api.marketplace.vector.apexfusion.org).
* Injected into the SPA boot block so the "Generate
* API key" page knows where to POST /signup. When
* unset, the SPA falls back to deriving "api." +
* current host.
*/
const HEX64_RE = /^[0-9a-fA-F]{64}$/;
@@ -56,6 +62,7 @@ export interface BuyerConfig {
password: string;
sessionSecret: string;
cookieSecure: boolean;
gatewayPublicUrl: string;
}
function requireField(env: Record<string, string | undefined>, name: string): string {
@@ -141,6 +148,11 @@ export function loadConfig(env: Record<string, string | undefined>): BuyerConfig
cookieSecure = cookieSecureStr === "1";
}
// GATEWAY_PUBLIC_URL — optional. Public base URL of the gateway, surfaced to
// the SPA via the boot block (no trailing slash). Empty string means the SPA
// derives the URL from its own host ("api." + location.host).
const gatewayPublicUrl = (env.GATEWAY_PUBLIC_URL ?? "").trim().replace(/\/+$/, "");
return {
privKeyHex,
indexerUrl,
@@ -155,5 +167,6 @@ export function loadConfig(env: Record<string, string | undefined>): BuyerConfig
password,
sessionSecret,
cookieSecure,
gatewayPublicUrl,
};
}

View File

@@ -133,6 +133,7 @@ export async function runMain(env: Record<string, string | undefined>): Promise<
password: config.password,
sessionSecret: config.sessionSecret,
cookieSecure: config.cookieSecure,
gatewayPublicUrl: config.gatewayPublicUrl,
jobStore,
pdfCaps,
});

View File

@@ -90,6 +90,7 @@ import type {
SubmitTtsResult,
StartChatOptions,
StartChatResult,
ChatSettleMode,
EndChatOptions,
EndChatResult,
AcceptResultOptions,
@@ -899,7 +900,13 @@ export class Marketplace extends EventEmitterBase {
}
this.emitProgress({ type: "chat_started", escrow_ref: escrowRefStr });
return { escrowRef: escrowOutputRef, sessionNonce, supplierBaseUrl: advertDatum.endpoint_url };
// Ticket-mode suppliers answer {status:"ticket"} (no Claim); anything else
// — "claimed" or an older supplier without the field — is full settlement.
const startStatus = startRes.body && typeof startRes.body === "object"
? (startRes.body as { status?: string }).status
: undefined;
const settleMode: ChatSettleMode = startStatus === "ticket" ? "ticket" : "full";
return { escrowRef: escrowOutputRef, sessionNonce, supplierBaseUrl: advertDatum.endpoint_url, settleMode };
}
/**
@@ -937,10 +944,20 @@ export class Marketplace extends EventEmitterBase {
});
}
const endBody = endRes.body as {
status?: string;
settle_mode?: string;
receipt?: import("@marketplace/shared/receipt").Receipt;
receipt_signature?: string;
submitted_ref?: string;
};
// Ticket sessions produce no receipt and nothing to Accept — the Open
// escrow returns via reclaim after deliver_by. Trusting the supplier's
// response is incentive-safe: lying in either direction only forfeits the
// supplier's own payout (the buyer's sweeper reclaims unsettled escrows).
if (endBody.settle_mode === "ticket") {
this.emitProgress({ type: "chat_ended", escrow_ref: escrowRefStr });
return { settleMode: "ticket", escrowRef };
}
if (!endBody.receipt || !endBody.receipt_signature) {
throw new SupplierError("malformed_response", { message: "supplier end response missing receipt" });
}
@@ -1022,7 +1039,7 @@ export class Marketplace extends EventEmitterBase {
this.emitProgress({ type: "accept_submitted", escrow_ref: refToString(acceptedRef) });
this.emitProgress({ type: "chat_ended", escrow_ref: escrowRefStr });
return { receipt, receiptSignature, escrowRef, acceptedRef };
return { settleMode: "full", receipt, receiptSignature, escrowRef, acceptedRef };
}
// ─── acceptResult / reclaim ───────────────────────────────────────────

View File

@@ -16,6 +16,9 @@ export type {
AcceptResultOptions,
ReclaimOptions,
GetTaskHistoryOptions,
ChatSettleMode,
StartChatResult,
EndChatResult,
} from "./types.js";
export { ReceiptVerificationError, IndexerError, SupplierError } from "./types.js";
export type { TaskHistoryStore } from "./history.js";

View File

@@ -197,6 +197,11 @@ export interface StartChatOptions {
advertRef: OutputReference;
payment_lovelace: bigint;
}
/** Chain settlement mode a chat session runs under (reported by the supplier
* at /v1/chat/start): "full" = Claim/Submit/Accept lifecycle; "ticket" = the
* buyer's escrow post is the only chain op (reclaimed after deliver_by). */
export type ChatSettleMode = "full" | "ticket";
export interface StartChatResult {
escrowRef: OutputReference;
/** Random nonce committed into the escrow's session-init prompt_hash. The
@@ -204,6 +209,8 @@ export interface StartChatResult {
sessionNonce: string;
/** Supplier endpoint_url, cached by the buyer-app for message routing. */
supplierBaseUrl: string;
/** How the supplier settles this session ("full" when it Claimed). */
settleMode: ChatSettleMode;
}
/**
@@ -220,13 +227,21 @@ export interface EndChatOptions {
/** The browser's local transcript mirror, used to verify response_hash. */
transcript?: import("@marketplace/shared/tx").ChatMessage[];
}
export interface EndChatResult {
receipt: Receipt;
receiptSignature: string;
escrowRef: OutputReference;
/** The Submitted escrow UTxO that was Accepted. */
acceptedRef: OutputReference;
}
export type EndChatResult =
| {
settleMode: "full";
receipt: Receipt;
receiptSignature: string;
escrowRef: OutputReference;
/** The Submitted escrow UTxO that was Accepted. */
acceptedRef: OutputReference;
}
| {
/** Ticket session: nothing was Submitted/Accepted — no receipt exists.
* The Open escrow returns to the buyer via reclaim after deliver_by. */
settleMode: "ticket";
escrowRef: OutputReference;
};
/**
* AcceptResultOptions — arguments to Marketplace.acceptResult().

View File

@@ -109,6 +109,11 @@ export interface AppDeps {
* brute-force rate-limit window. Defaults to Date.now. Tests use a fake
* clock to drive expiry and rolling-renewal assertions deterministically. */
nowMs?: () => number;
/** Public base URL of the OpenAI-compatible gateway, injected into the SPA
* boot block so the "Generate API key" page can POST cross-origin to
* `${gatewayUrl}/signup`. Public, non-secret value. When empty/omitted the
* SPA derives "api." + its own host. */
gatewayPublicUrl?: string;
/** PDF book summarizer job registry. When provided, the /v1/pdf-* routes
* are live; when omitted they respond 503 (feature disabled). Built in
* runMain from the same marketplace/chain/wallet/archive deps. */
@@ -680,9 +685,11 @@ export function createApp(deps: AppDeps): Express {
const DEMO_CHAT_MODEL = "moonshotai/kimi-k2.6";
function isChatMessageArray(v: unknown): v is ChatMessage[] {
// The free chat demo stays text-only: the `tool` role (added to
// ChatMessage for the gateway demo endpoint) is not accepted here.
return Array.isArray(v) && v.length > 0 && v.every(
(m) => m && typeof m === "object"
&& (["system", "user", "assistant"] as const).includes((m as ChatMessage).role)
&& ["system", "user", "assistant"].includes((m as ChatMessage).role)
&& typeof (m as ChatMessage).content === "string",
);
}
@@ -927,6 +934,14 @@ export function createApp(deps: AppDeps): Express {
supplierBaseUrl: state.supplierBaseUrl,
});
chatRouteState.delete(rawRef);
if (result.settleMode === "ticket") {
// Ticket session: no receipt/Accept — funds return via reclaim.
return res.status(200).json({
status: "ended",
escrow_ref: rawRef,
settle_mode: "ticket",
});
}
return res.status(200).json({
status: "accepted",
escrow_ref: rawRef,
@@ -1310,6 +1325,9 @@ export function createApp(deps: AppDeps): Express {
pubKeyHash: resolved.walletKey.pubKeyHash,
address: displayAddress,
},
// Public gateway base URL for the "Generate API key" page. Empty
// string when unconfigured — the SPA then derives "api." + host.
gatewayUrl: deps.gatewayPublicUrl ?? "",
})};</script>`
: "";
const indexWithBoot = cachedTemplate.replace(

View File

@@ -6,6 +6,7 @@
* /tasks → <TaskHistory />
* /pending → <PendingReceipts /> UX-1: buyer-side Accept button
* /wallet → <Wallet />
* /api-keys → <ApiKeys /> self-serve gateway API key generation
*/
import { Routes, Route, Link } from "react-router-dom";
@@ -14,6 +15,7 @@ import TaskHistory from "./pages/TaskHistory.js";
import Wallet from "./pages/Wallet.js";
import PendingReceipts from "./pages/PendingReceipts.js";
import BookSummarizer from "./pages/BookSummarizer.js";
import ApiKeys from "./pages/ApiKeys.js";
import { RequireAuth, useAuth } from "./state/AuthContext.js";
function SignOutButton() {
@@ -35,10 +37,13 @@ export default function App() {
<div className="min-h-screen bg-gray-50 text-gray-900">
<nav className="bg-white border-b border-gray-200 px-6 py-3 flex items-center gap-6">
<Link to="/" className="font-medium hover:underline">Dashboard</Link>
<Link to="/book" className="font-medium hover:underline">Book Summarizer</Link>
{/* Book Summarizer tab hidden for now — the /book route and all PDF
summarize code remain intact; restore by re-adding this link:
<Link to="/book" className="font-medium hover:underline">Book Summarizer</Link> */}
<Link to="/tasks" className="font-medium hover:underline">Tasks</Link>
<Link to="/pending" className="font-medium hover:underline">Pending</Link>
<Link to="/wallet" className="font-medium hover:underline">Wallet</Link>
<Link to="/api-keys" className="font-medium hover:underline">API Keys</Link>
<SignOutButton />
</nav>
<main className="p-6">
@@ -48,6 +53,7 @@ export default function App() {
<Route path="/tasks" element={<TaskHistory />} />
<Route path="/pending" element={<PendingReceipts />} />
<Route path="/wallet" element={<Wallet />} />
<Route path="/api-keys" element={<ApiKeys />} />
</Routes>
</main>
</div>

View File

@@ -267,7 +267,18 @@ export default function ChatForm({ advertRef, payment_lovelace }: ChatFormProps
} catch { /* keep fallback */ }
throw new Error(msg);
}
setCharged(payment_lovelace !== undefined ? ap3x(payment_lovelace) : "0");
let endBody: { settle_mode?: string } = {};
try {
endBody = (await resp.json()) as { settle_mode?: string };
} catch { /* older buyer-app responses stay full-settle */ }
// Ticket sessions never charge — the escrow returns via reclaim.
setCharged(
endBody.settle_mode === "ticket"
? "0 (ticket session — escrow reclaimed)"
: payment_lovelace !== undefined
? ap3x(payment_lovelace)
: "0",
);
} catch (err) {
setError(err instanceof Error ? err.message : String(err));
} finally {

32
buyer/src/ui/gateway.ts Normal file
View File

@@ -0,0 +1,32 @@
/**
* buyer/src/ui/gateway.ts — resolve the OpenAI-compatible gateway base URL the
* SPA should call for the "Generate API key" page.
*
* Resolution order:
* 1. window.__BUYER_BOOT__.gatewayUrl — injected by the buyer server from the
* GATEWAY_PUBLIC_URL env (the configured value). Present on any page load
* made while the buyer_session cookie is valid.
* 2. Derived "api." + current host — the deploy convention is exactly
* api.<buyer-host> (marketplace.… → api.marketplace.…). This covers the
* gap right after a fresh login, when the SPA flips auth state WITHOUT a
* page reload so the boot block (which is only injected into
* server-rendered HTML) hasn't been re-fetched yet.
*
* The result never has a trailing slash.
*/
interface BuyerBootMaybeGateway {
gatewayUrl?: string;
}
function stripTrailingSlashes(url: string): string {
return url.replace(/\/+$/, "");
}
export function resolveGatewayUrl(): string {
const boot = (window as unknown as { __BUYER_BOOT__?: BuyerBootMaybeGateway })
.__BUYER_BOOT__;
const configured = boot?.gatewayUrl?.trim();
if (configured) return stripTrailingSlashes(configured);
return `${window.location.protocol}//api.${window.location.host}`;
}

View File

@@ -0,0 +1,205 @@
/**
* buyer/src/ui/pages/ApiKeys.tsx — self-serve "Generate API key" page.
*
* Mints a custodial gateway API key by POSTing directly to the gateway's
* public /signup endpoint (cross-origin; the gateway CORS-allows this origin).
* The raw key is shown ONCE — it is never stored server-side in retrievable
* form, so it cannot be listed or recovered later. The page also surfaces the
* funding deposit address and a ready-to-paste usage snippet.
*
* Access is gated by the SPA's existing operator login (this page only renders
* inside <RequireAuth>). The gateway /signup itself stays public + IP
* rate-limited; per-user accounts are a later feature.
*/
import { useState } from "react";
import { resolveGatewayUrl } from "../gateway.js";
interface SignupResult {
api_key: string;
key_prefix: string;
deposit_address: string;
}
/** Pull a human-readable message out of whatever error shape the gateway
* returns (OpenAI `{error:{message}}`, plain `{error|message}`, or neither). */
function errorMessage(body: unknown, status: number, statusText: string): string {
if (body && typeof body === "object") {
const b = body as Record<string, unknown>;
const err = b.error;
if (err && typeof err === "object" && typeof (err as Record<string, unknown>).message === "string") {
return (err as Record<string, string>).message;
}
if (typeof err === "string") return err;
if (typeof b.message === "string") return b.message;
}
return `${status} ${statusText}`.trim() || "request failed";
}
function CopyButton({ value, label }: { value: string; label?: string }) {
const [copied, setCopied] = useState(false);
return (
<button
type="button"
onClick={() => {
void navigator.clipboard
?.writeText(value)
.then(() => {
setCopied(true);
setTimeout(() => setCopied(false), 1500);
})
.catch(() => {
/* clipboard blocked (insecure context / permissions) — no-op */
});
}}
className="rounded border border-gray-300 px-2 py-0.5 text-xs text-gray-600 hover:bg-gray-50"
>
{copied ? "Copied!" : (label ?? "Copy")}
</button>
);
}
export default function ApiKeys() {
const [label, setLabel] = useState("");
const [loading, setLoading] = useState(false);
const [error, setError] = useState<string | null>(null);
const [result, setResult] = useState<SignupResult | null>(null);
const gatewayUrl = resolveGatewayUrl();
const baseUrl = `${gatewayUrl}/openai/v1`;
async function generate() {
setLoading(true);
setError(null);
setResult(null);
try {
const res = await fetch(`${gatewayUrl}/signup`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(label.trim() ? { label: label.trim() } : {}),
});
const body = (await res.json().catch(() => null)) as unknown;
if (!res.ok) {
throw new Error(errorMessage(body, res.status, res.statusText));
}
const r = body as Partial<SignupResult> | null;
if (!r || typeof r.api_key !== "string" || typeof r.deposit_address !== "string") {
throw new Error("unexpected response from gateway");
}
setResult({
api_key: r.api_key,
key_prefix: r.key_prefix ?? r.api_key.slice(0, 12),
deposit_address: r.deposit_address,
});
} catch (e) {
setError(e instanceof Error ? e.message : "could not reach the gateway");
} finally {
setLoading(false);
}
}
const curlSnippet =
result === null
? ""
: [
`curl ${baseUrl}/chat/completions \\`,
` -H "Authorization: Bearer ${result.api_key}" \\`,
` -H "Content-Type: application/json" \\`,
` -d '{"model": "<model>", "messages": [{"role": "user", "content": "Hello"}]}'`,
].join("\n");
return (
<div className="max-w-3xl space-y-6">
<div className="space-y-1">
<h1 className="text-2xl font-semibold">API Keys</h1>
<p className="text-sm text-gray-600">
Generate an OpenAI-compatible API key for the marketplace gateway. Each key has its own
custodial wallet — fund its deposit address with AP3X, then point any OpenAI SDK at{" "}
<code className="font-mono text-xs">{baseUrl}</code>.
</p>
</div>
<div className="rounded-lg border border-gray-200 bg-white p-4 space-y-3">
<label className="block text-sm font-medium text-gray-700">
Label <span className="font-normal text-gray-400">(optional)</span>
<input
type="text"
value={label}
onChange={(e) => setLabel(e.target.value)}
placeholder="e.g. my-laptop"
maxLength={120}
disabled={loading}
className="mt-1 block w-full rounded border border-gray-300 px-3 py-2 text-sm focus:border-gray-400 focus:outline-none disabled:opacity-50"
/>
</label>
<button
type="button"
onClick={() => void generate()}
disabled={loading}
className="rounded bg-gray-900 px-4 py-2 text-sm font-medium text-white hover:bg-gray-700 disabled:opacity-50"
data-testid="generate-api-key"
>
{loading ? "Generating…" : "Generate API key"}
</button>
{error !== null && (
<p className="text-sm text-red-600" data-testid="api-key-error">
{error}
</p>
)}
</div>
{result !== null && (
<div className="space-y-4 rounded-lg border border-amber-300 bg-amber-50 p-4" data-testid="api-key-result">
<div className="rounded border border-amber-400 bg-amber-100 px-3 py-2 text-sm text-amber-900">
Copy your API key now — it is shown <strong>only once</strong> and cannot be recovered. If you
lose it, generate a new one.
</div>
<div>
<div className="flex items-center justify-between">
<span className="text-sm font-medium text-gray-700">API key</span>
<CopyButton value={result.api_key} />
</div>
<code className="mt-1 block break-all rounded bg-white px-3 py-2 font-mono text-sm">
{result.api_key}
</code>
</div>
<div>
<div className="flex items-center justify-between">
<span className="text-sm font-medium text-gray-700">Deposit address (fund with AP3X)</span>
<CopyButton value={result.deposit_address} />
</div>
<code className="mt-1 block break-all rounded bg-white px-3 py-2 font-mono text-sm">
{result.deposit_address}
</code>
</div>
<div>
<div className="flex items-center justify-between">
<span className="text-sm font-medium text-gray-700">Base URL</span>
<CopyButton value={baseUrl} />
</div>
<code className="mt-1 block break-all rounded bg-white px-3 py-2 font-mono text-sm">
{baseUrl}
</code>
</div>
<div>
<div className="flex items-center justify-between">
<span className="text-sm font-medium text-gray-700">Quick start</span>
<CopyButton value={curlSnippet} />
</div>
<pre className="mt-1 overflow-x-auto rounded bg-gray-900 px-3 py-2 font-mono text-xs text-gray-100">
{curlSnippet}
</pre>
<p className="mt-1 text-xs text-gray-500">
Replace <code className="font-mono">&lt;model&gt;</code> with a model from{" "}
<code className="font-mono">{baseUrl}/models</code>.
</p>
</div>
</div>
)}
</div>
);
}

View File

@@ -29,6 +29,11 @@ services:
# the response archive + PDF job store (past summaries) across container
# recreation/redeploys.
ARCHIVE_DIR: /repo/data/archive
# Public base URL of the OpenAI gateway, surfaced to the SPA boot block so
# the "API Keys" page can POST cross-origin to ${gatewayUrl}/signup. The
# gateway must CORS-allow https://marketplace.vector.apexfusion.org (see
# GATEWAY_CORS_ORIGINS in docker-compose.gateway.yml).
GATEWAY_PUBLIC_URL: https://api.marketplace.vector.apexfusion.org
# NETWORK_ID intentionally not pinned — see supplier compose comment.
volumes:
- buyer-mainnet-archive:/repo/data/archive

View File

@@ -27,6 +27,24 @@ services:
NETWORK_ID: "1"
GATEWAY_PORT: "8080"
GATEWAY_DB_DIR: /repo/data/gateway
# Mainnet chain genesis (slot↔wallclock). MUST match buyer/supplier or
# PostEscrow validity bounds land ~51 days off and Ogmios rejects (code 3118).
VECTOR_ZERO_TIME_MS: "1756485600000"
# On-chain reference-script UTxOs (escrow spending validator) used by
# Accept/Reclaim/Submit. Same values the buyer/suppliers use on mainnet.
ESCROW_REF_UTXO: c8d84c6d67ec67a1efe5e9c6c06d53020e05d1bb96d1c55ecb1eb7d5010c4d54#0
ADVERT_REF_UTXO: c8d84c6d67ec67a1efe5e9c6c06d53020e05d1bb96d1c55ecb1eb7d5010c4d54#1
# Browser Origins allowed to call the gateway cross-origin. The buyer
# frontend's "API Keys" page (served from marketplace.vector…) POSTs to
# /signup here. Comma-separated; empty disables CORS entirely.
GATEWAY_CORS_ORIGINS: https://marketplace.vector.apexfusion.org
# Ticket mode: the escrow post at session open is the only chain op;
# the sweeper reclaims Open escrows after deliver_by. Must match the
# kimi-chat suppliers' CHAT_SETTLE_MODE.
CHAT_SETTLE_MODE: ticket
# Demo janitor closes idle sessions BEFORE the suppliers' 15 min
# CHAT_IDLE_TIMEOUT_MS backstop.
DEMO_SESSION_IDLE_MS: "600000"
volumes:
- gateway-mainnet-data:/repo/data/gateway
healthcheck:

View File

@@ -0,0 +1,65 @@
# deploy/mainnet/docker-compose.supplier-deepseek-flash-1.yml — Supplier — MAINNET (OpenRouter / DeepSeek V4 Flash #1)
#
# Parallel-capacity supplier for deepseek/deepseek-v4-flash via OpenRouter's
# OpenAI-compatible API. Same shape as the kimi replicas; distinct only in
# instance name, container, env file, Traefik host/router, and advert model.
# Shares the same OpenRouter API key as the kimi suppliers (accepted ceiling;
# supplier has no 429/backoff). Advert caps: max_output_tokens=262144,
# max_processing_ms=3600000; OPENAI_TIMEOUT_MS matched.
#
# Pre-flight:
# 1. DNS (proxied): mp-suppliers-deepseek-flash-1.vector.apexfusion.org → host
# 2. supplier/.env.deepseek-flash-1 on host (chmod 600): SUPPLIER_PRIV_KEY_HEX,
# OPENAI_API_KEY, OGMIOS_URL, NETWORK_ID=1, VECTOR_ZERO_TIME_MS,
# ESCROW_REF_UTXO, ADVERT_REF_UTXO, ADVERT_REF, LIVE_CHAIN=1
# 3. Networks marketplace-mainnet-net + traefik-net already exist.
name: marketplace-mainnet-supplier-deepseek-flash-1
services:
supplier:
build:
context: ../..
dockerfile: supplier/Dockerfile
container_name: marketplace-mainnet-supplier-deepseek-flash-1
restart: unless-stopped
env_file:
- path: ../../supplier/.env.deepseek-flash-1
required: false
environment:
LLM_BACKEND: openai
# No /v1 suffix — supplier/src/openai.ts appends /v1/chat/completions.
OPENAI_BASE_URL: https://openrouter.ai/api
# Matches the advert's max_processing_ms=3600000 (60 min).
OPENAI_TIMEOUT_MS: "3600000"
PORT: "8080"
LIVE_CHAIN: "1"
# NETWORK_ID and OGMIOS_URL come from supplier/.env.deepseek-flash-1.
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null || exit 1"]
interval: 20s
timeout: 5s
retries: 3
start_period: 15s
networks:
- marketplace-mainnet-net
- traefik-net
labels:
- "traefik.enable=true"
- "traefik.docker.network=traefik-net"
- "traefik.http.routers.mp-suppliers-deepseek-flash-1-mainnet.entrypoints=web"
- "traefik.http.routers.mp-suppliers-deepseek-flash-1-mainnet.rule=Host(`mp-suppliers-deepseek-flash-1.vector.apexfusion.org`)"
- "traefik.http.services.mp-suppliers-deepseek-flash-1-mainnet.loadbalancer.server.port=8080"
logging:
driver: json-file
options:
max-size: "10m"
max-file: "3"
networks:
marketplace-mainnet-net:
external: true
name: marketplace-mainnet-net
traefik-net:
external: true
name: traefik-net

View File

@@ -0,0 +1,65 @@
# deploy/mainnet/docker-compose.supplier-deepseek-flash-2.yml — Supplier — MAINNET (OpenRouter / DeepSeek V4 Flash #2)
#
# Parallel-capacity supplier for deepseek/deepseek-v4-flash via OpenRouter's
# OpenAI-compatible API. Same shape as the kimi replicas; distinct only in
# instance name, container, env file, Traefik host/router, and advert model.
# Shares the same OpenRouter API key as the kimi suppliers (accepted ceiling;
# supplier has no 429/backoff). Advert caps: max_output_tokens=262144,
# max_processing_ms=3600000; OPENAI_TIMEOUT_MS matched.
#
# Pre-flight:
# 1. DNS (proxied): mp-suppliers-deepseek-flash-2.vector.apexfusion.org → host
# 2. supplier/.env.deepseek-flash-2 on host (chmod 600): SUPPLIER_PRIV_KEY_HEX,
# OPENAI_API_KEY, OGMIOS_URL, NETWORK_ID=1, VECTOR_ZERO_TIME_MS,
# ESCROW_REF_UTXO, ADVERT_REF_UTXO, ADVERT_REF, LIVE_CHAIN=1
# 3. Networks marketplace-mainnet-net + traefik-net already exist.
name: marketplace-mainnet-supplier-deepseek-flash-2
services:
supplier:
build:
context: ../..
dockerfile: supplier/Dockerfile
container_name: marketplace-mainnet-supplier-deepseek-flash-2
restart: unless-stopped
env_file:
- path: ../../supplier/.env.deepseek-flash-2
required: false
environment:
LLM_BACKEND: openai
# No /v1 suffix — supplier/src/openai.ts appends /v1/chat/completions.
OPENAI_BASE_URL: https://openrouter.ai/api
# Matches the advert's max_processing_ms=3600000 (60 min).
OPENAI_TIMEOUT_MS: "3600000"
PORT: "8080"
LIVE_CHAIN: "1"
# NETWORK_ID and OGMIOS_URL come from supplier/.env.deepseek-flash-2.
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null || exit 1"]
interval: 20s
timeout: 5s
retries: 3
start_period: 15s
networks:
- marketplace-mainnet-net
- traefik-net
labels:
- "traefik.enable=true"
- "traefik.docker.network=traefik-net"
- "traefik.http.routers.mp-suppliers-deepseek-flash-2-mainnet.entrypoints=web"
- "traefik.http.routers.mp-suppliers-deepseek-flash-2-mainnet.rule=Host(`mp-suppliers-deepseek-flash-2.vector.apexfusion.org`)"
- "traefik.http.services.mp-suppliers-deepseek-flash-2-mainnet.loadbalancer.server.port=8080"
logging:
driver: json-file
options:
max-size: "10m"
max-file: "3"
networks:
marketplace-mainnet-net:
external: true
name: marketplace-mainnet-net
traefik-net:
external: true
name: traefik-net

View File

@@ -0,0 +1,65 @@
# deploy/mainnet/docker-compose.supplier-deepseek-flash-3.yml — Supplier — MAINNET (OpenRouter / DeepSeek V4 Flash #3)
#
# Parallel-capacity supplier for deepseek/deepseek-v4-flash via OpenRouter's
# OpenAI-compatible API. Same shape as the kimi replicas; distinct only in
# instance name, container, env file, Traefik host/router, and advert model.
# Shares the same OpenRouter API key as the kimi suppliers (accepted ceiling;
# supplier has no 429/backoff). Advert caps: max_output_tokens=262144,
# max_processing_ms=3600000; OPENAI_TIMEOUT_MS matched.
#
# Pre-flight:
# 1. DNS (proxied): mp-suppliers-deepseek-flash-3.vector.apexfusion.org → host
# 2. supplier/.env.deepseek-flash-3 on host (chmod 600): SUPPLIER_PRIV_KEY_HEX,
# OPENAI_API_KEY, OGMIOS_URL, NETWORK_ID=1, VECTOR_ZERO_TIME_MS,
# ESCROW_REF_UTXO, ADVERT_REF_UTXO, ADVERT_REF, LIVE_CHAIN=1
# 3. Networks marketplace-mainnet-net + traefik-net already exist.
name: marketplace-mainnet-supplier-deepseek-flash-3
services:
supplier:
build:
context: ../..
dockerfile: supplier/Dockerfile
container_name: marketplace-mainnet-supplier-deepseek-flash-3
restart: unless-stopped
env_file:
- path: ../../supplier/.env.deepseek-flash-3
required: false
environment:
LLM_BACKEND: openai
# No /v1 suffix — supplier/src/openai.ts appends /v1/chat/completions.
OPENAI_BASE_URL: https://openrouter.ai/api
# Matches the advert's max_processing_ms=3600000 (60 min).
OPENAI_TIMEOUT_MS: "3600000"
PORT: "8080"
LIVE_CHAIN: "1"
# NETWORK_ID and OGMIOS_URL come from supplier/.env.deepseek-flash-3.
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null || exit 1"]
interval: 20s
timeout: 5s
retries: 3
start_period: 15s
networks:
- marketplace-mainnet-net
- traefik-net
labels:
- "traefik.enable=true"
- "traefik.docker.network=traefik-net"
- "traefik.http.routers.mp-suppliers-deepseek-flash-3-mainnet.entrypoints=web"
- "traefik.http.routers.mp-suppliers-deepseek-flash-3-mainnet.rule=Host(`mp-suppliers-deepseek-flash-3.vector.apexfusion.org`)"
- "traefik.http.services.mp-suppliers-deepseek-flash-3-mainnet.loadbalancer.server.port=8080"
logging:
driver: json-file
options:
max-size: "10m"
max-file: "3"
networks:
marketplace-mainnet-net:
external: true
name: marketplace-mainnet-net
traefik-net:
external: true
name: traefik-net

View File

@@ -0,0 +1,67 @@
# deploy/mainnet/docker-compose.supplier-kimi-2.yml — Supplier — MAINNET (OpenRouter / Kimi K2.6 replica #2)
#
# Parallel-capacity replica of supplier-kimi (OpenRouter, moonshotai/kimi-k2.6).
# Identical to docker-compose.supplier-kimi.yml except for the instance name,
# container name, env file, Traefik host/router, and OPENAI_TIMEOUT_MS (raised
# to 3600000 to match this advert's max_processing_ms=3600000, max_output_tokens=262144).
# Shares the same OpenRouter API key as the other kimi suppliers (accepted
# concurrency ceiling; supplier has no 429/backoff).
#
# Pre-flight:
# 1. DNS (proxied): mp-suppliers-kimi-2.vector.apexfusion.org → host
# 2. supplier/.env.kimi-2 on host (chmod 600): SUPPLIER_PRIV_KEY_HEX, OPENAI_API_KEY,
# OGMIOS_URL, NETWORK_ID=1, VECTOR_ZERO_TIME_MS, ESCROW_REF_UTXO, ADVERT_REF_UTXO,
# ADVERT_REF, LIVE_CHAIN=1
# 3. Networks marketplace-mainnet-net + traefik-net already exist.
name: marketplace-mainnet-supplier-kimi-2
services:
supplier:
build:
context: ../..
dockerfile: supplier/Dockerfile
container_name: marketplace-mainnet-supplier-kimi-2
restart: unless-stopped
env_file:
- path: ../../supplier/.env.kimi-2
required: false
environment:
LLM_BACKEND: openai
# No /v1 suffix — supplier/src/openai.ts appends /v1/chat/completions.
OPENAI_BASE_URL: https://openrouter.ai/api
# Raised to match the advert's max_processing_ms=3600000 (60 min) so the
# upstream call isn't aborted before the advertised processing deadline.
OPENAI_TIMEOUT_MS: "3600000"
PORT: "8080"
LIVE_CHAIN: "1"
# NETWORK_ID and OGMIOS_URL intentionally not pinned here — they MUST
# come from supplier/.env.kimi-2 on the mainnet host (footgun avoidance).
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null || exit 1"]
interval: 20s
timeout: 5s
retries: 3
start_period: 15s
networks:
- marketplace-mainnet-net
- traefik-net
labels:
- "traefik.enable=true"
- "traefik.docker.network=traefik-net"
- "traefik.http.routers.mp-suppliers-kimi-2-mainnet.entrypoints=web"
- "traefik.http.routers.mp-suppliers-kimi-2-mainnet.rule=Host(`mp-suppliers-kimi-2.vector.apexfusion.org`)"
- "traefik.http.services.mp-suppliers-kimi-2-mainnet.loadbalancer.server.port=8080"
logging:
driver: json-file
options:
max-size: "10m"
max-file: "3"
networks:
marketplace-mainnet-net:
external: true
name: marketplace-mainnet-net
traefik-net:
external: true
name: traefik-net

View File

@@ -0,0 +1,67 @@
# deploy/mainnet/docker-compose.supplier-kimi-3.yml — Supplier — MAINNET (OpenRouter / Kimi K2.6 replica #3)
#
# Parallel-capacity replica of supplier-kimi (OpenRouter, moonshotai/kimi-k2.6).
# Identical to docker-compose.supplier-kimi.yml except for the instance name,
# container name, env file, Traefik host/router, and OPENAI_TIMEOUT_MS (raised
# to 3600000 to match this advert's max_processing_ms=3600000, max_output_tokens=262144).
# Shares the same OpenRouter API key as the other kimi suppliers (accepted
# concurrency ceiling; supplier has no 429/backoff).
#
# Pre-flight:
# 1. DNS (proxied): mp-suppliers-kimi-3.vector.apexfusion.org → host
# 2. supplier/.env.kimi-3 on host (chmod 600): SUPPLIER_PRIV_KEY_HEX, OPENAI_API_KEY,
# OGMIOS_URL, NETWORK_ID=1, VECTOR_ZERO_TIME_MS, ESCROW_REF_UTXO, ADVERT_REF_UTXO,
# ADVERT_REF, LIVE_CHAIN=1
# 3. Networks marketplace-mainnet-net + traefik-net already exist.
name: marketplace-mainnet-supplier-kimi-3
services:
supplier:
build:
context: ../..
dockerfile: supplier/Dockerfile
container_name: marketplace-mainnet-supplier-kimi-3
restart: unless-stopped
env_file:
- path: ../../supplier/.env.kimi-3
required: false
environment:
LLM_BACKEND: openai
# No /v1 suffix — supplier/src/openai.ts appends /v1/chat/completions.
OPENAI_BASE_URL: https://openrouter.ai/api
# Raised to match the advert's max_processing_ms=3600000 (60 min) so the
# upstream call isn't aborted before the advertised processing deadline.
OPENAI_TIMEOUT_MS: "3600000"
PORT: "8080"
LIVE_CHAIN: "1"
# NETWORK_ID and OGMIOS_URL intentionally not pinned here — they MUST
# come from supplier/.env.kimi-3 on the mainnet host (footgun avoidance).
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null || exit 1"]
interval: 20s
timeout: 5s
retries: 3
start_period: 15s
networks:
- marketplace-mainnet-net
- traefik-net
labels:
- "traefik.enable=true"
- "traefik.docker.network=traefik-net"
- "traefik.http.routers.mp-suppliers-kimi-3-mainnet.entrypoints=web"
- "traefik.http.routers.mp-suppliers-kimi-3-mainnet.rule=Host(`mp-suppliers-kimi-3.vector.apexfusion.org`)"
- "traefik.http.services.mp-suppliers-kimi-3-mainnet.loadbalancer.server.port=8080"
logging:
driver: json-file
options:
max-size: "10m"
max-file: "3"
networks:
marketplace-mainnet-net:
external: true
name: marketplace-mainnet-net
traefik-net:
external: true
name: traefik-net

View File

@@ -0,0 +1,67 @@
# deploy/mainnet/docker-compose.supplier-kimi-4.yml — Supplier — MAINNET (OpenRouter / Kimi K2.6 replica #4)
#
# Parallel-capacity replica of supplier-kimi (OpenRouter, moonshotai/kimi-k2.6).
# Identical to docker-compose.supplier-kimi.yml except for the instance name,
# container name, env file, Traefik host/router, and OPENAI_TIMEOUT_MS (raised
# to 3600000 to match this advert's max_processing_ms=3600000, max_output_tokens=262144).
# Shares the same OpenRouter API key as the other kimi suppliers (accepted
# concurrency ceiling; supplier has no 429/backoff).
#
# Pre-flight:
# 1. DNS (proxied): mp-suppliers-kimi-4.vector.apexfusion.org → host
# 2. supplier/.env.kimi-4 on host (chmod 600): SUPPLIER_PRIV_KEY_HEX, OPENAI_API_KEY,
# OGMIOS_URL, NETWORK_ID=1, VECTOR_ZERO_TIME_MS, ESCROW_REF_UTXO, ADVERT_REF_UTXO,
# ADVERT_REF, LIVE_CHAIN=1
# 3. Networks marketplace-mainnet-net + traefik-net already exist.
name: marketplace-mainnet-supplier-kimi-4
services:
supplier:
build:
context: ../..
dockerfile: supplier/Dockerfile
container_name: marketplace-mainnet-supplier-kimi-4
restart: unless-stopped
env_file:
- path: ../../supplier/.env.kimi-4
required: false
environment:
LLM_BACKEND: openai
# No /v1 suffix — supplier/src/openai.ts appends /v1/chat/completions.
OPENAI_BASE_URL: https://openrouter.ai/api
# Raised to match the advert's max_processing_ms=3600000 (60 min) so the
# upstream call isn't aborted before the advertised processing deadline.
OPENAI_TIMEOUT_MS: "3600000"
PORT: "8080"
LIVE_CHAIN: "1"
# NETWORK_ID and OGMIOS_URL intentionally not pinned here — they MUST
# come from supplier/.env.kimi-4 on the mainnet host (footgun avoidance).
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null || exit 1"]
interval: 20s
timeout: 5s
retries: 3
start_period: 15s
networks:
- marketplace-mainnet-net
- traefik-net
labels:
- "traefik.enable=true"
- "traefik.docker.network=traefik-net"
- "traefik.http.routers.mp-suppliers-kimi-4-mainnet.entrypoints=web"
- "traefik.http.routers.mp-suppliers-kimi-4-mainnet.rule=Host(`mp-suppliers-kimi-4.vector.apexfusion.org`)"
- "traefik.http.services.mp-suppliers-kimi-4-mainnet.loadbalancer.server.port=8080"
logging:
driver: json-file
options:
max-size: "10m"
max-file: "3"
networks:
marketplace-mainnet-net:
external: true
name: marketplace-mainnet-net
traefik-net:
external: true
name: traefik-net

View File

@@ -0,0 +1,78 @@
# deploy/mainnet/docker-compose.supplier-kimi-chat-2.yml — Supplier — MAINNET
# (OpenRouter / Kimi K2.6 — MULTI-TURN CHAT capability `llm.chat.v1`, REPLICA 2)
#
# Second llm.chat.v1 replica: a chat session holds a supplier's single slot for
# the whole conversation, so concurrent chat users = number of chat suppliers.
# Deployed for the shared-demo-key endpoint (session-backed /chat/completions).
# Identical to docker-compose.supplier-kimi-chat.yml except for its own wallet,
# advert, env file and hostname.
#
# Pre-flight:
# 1. DNS A-record: mp-suppliers-kimi-chat-2.vector.apexfusion.org → host IP
# 2. New supplier wallet (gen-keypair), funded with advert bond + fee float.
# 3. Post a fresh llm.chat.v1 advert for this wallet:
# --capability-id llm.chat.v1 --model moonshotai/kimi-k2.6
# --max-processing-ms 1800000 (same price as the primary chat advert)
# 4. supplier/.env.kimi-chat-2 written on host (chmod 600) with at minimum:
# SUPPLIER_PRIV_KEY_HEX, OPENAI_API_KEY, OGMIOS_URL, NETWORK_ID=1,
# VECTOR_ZERO_TIME_MS, ESCROW_REF_UTXO, ADVERT_REF_UTXO, ADVERT_REF,
# LIVE_CHAIN=1
name: marketplace-mainnet-supplier-kimi-chat-2
services:
supplier:
build:
context: ../..
dockerfile: supplier/Dockerfile
container_name: marketplace-mainnet-supplier-kimi-chat-2
restart: unless-stopped
env_file:
- path: ../../supplier/.env.kimi-chat-2
required: false
environment:
# Multi-turn chat session capability (mounts /v1/chat/{start,message,end}).
CAPABILITY_KIND: chat-session
LLM_BACKEND: openai
# No /v1 suffix — supplier/src/openai.ts appends /v1/chat/completions.
OPENAI_BASE_URL: https://openrouter.ai/api
# Kimi K2.6 generations can be slow; mirror the one-off Kimi compose.
OPENAI_TIMEOUT_MS: "300000"
# 2.5 min idle end: single-slot suppliers must free up quickly for the
# shared demo endpoint (the gateway transparently reopens on next turn).
CHAT_IDLE_TIMEOUT_MS: "900000"
CHAT_SETTLE_MODE: ticket
MAX_CHAT_SESSIONS: "8"
PORT: "8080"
# Enable LiveOgmiosProvider so Claim/Submit actually broadcast.
LIVE_CHAIN: "1"
# NETWORK_ID and OGMIOS_URL intentionally not pinned here — they MUST
# come from the operator-provided supplier/.env.kimi-chat-2 on a mainnet host.
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null || exit 1"]
interval: 20s
timeout: 5s
retries: 3
start_period: 15s
networks:
- marketplace-mainnet-net
- traefik-net
labels:
- "traefik.enable=true"
- "traefik.docker.network=traefik-net"
- "traefik.http.routers.mp-suppliers-kimi-chat-2-mainnet.entrypoints=web"
- "traefik.http.routers.mp-suppliers-kimi-chat-2-mainnet.rule=Host(`mp-suppliers-kimi-chat-2.vector.apexfusion.org`)"
- "traefik.http.services.mp-suppliers-kimi-chat-2-mainnet.loadbalancer.server.port=8080"
logging:
driver: json-file
options:
max-size: "10m"
max-file: "3"
networks:
marketplace-mainnet-net:
external: true
name: marketplace-mainnet-net
traefik-net:
external: true
name: traefik-net

View File

@@ -0,0 +1,78 @@
# deploy/mainnet/docker-compose.supplier-kimi-chat-3.yml — Supplier — MAINNET
# (OpenRouter / Kimi K2.6 — MULTI-TURN CHAT capability `llm.chat.v1`, REPLICA 3)
#
# Third llm.chat.v1 replica: a chat session holds a supplier's single slot for
# the whole conversation, so concurrent chat users = number of chat suppliers.
# Deployed for the shared-demo-key endpoint (session-backed /chat/completions).
# Identical to docker-compose.supplier-kimi-chat.yml except for its own wallet,
# advert, env file and hostname.
#
# Pre-flight:
# 1. DNS A-record: mp-suppliers-kimi-chat-3.vector.apexfusion.org → host IP
# 2. New supplier wallet (gen-keypair), funded with advert bond + fee float.
# 3. Post a fresh llm.chat.v1 advert for this wallet:
# --capability-id llm.chat.v1 --model moonshotai/kimi-k2.6
# --max-processing-ms 1800000 (same price as the primary chat advert)
# 4. supplier/.env.kimi-chat-3 written on host (chmod 600) with at minimum:
# SUPPLIER_PRIV_KEY_HEX, OPENAI_API_KEY, OGMIOS_URL, NETWORK_ID=1,
# VECTOR_ZERO_TIME_MS, ESCROW_REF_UTXO, ADVERT_REF_UTXO, ADVERT_REF,
# LIVE_CHAIN=1
name: marketplace-mainnet-supplier-kimi-chat-3
services:
supplier:
build:
context: ../..
dockerfile: supplier/Dockerfile
container_name: marketplace-mainnet-supplier-kimi-chat-3
restart: unless-stopped
env_file:
- path: ../../supplier/.env.kimi-chat-3
required: false
environment:
# Multi-turn chat session capability (mounts /v1/chat/{start,message,end}).
CAPABILITY_KIND: chat-session
LLM_BACKEND: openai
# No /v1 suffix — supplier/src/openai.ts appends /v1/chat/completions.
OPENAI_BASE_URL: https://openrouter.ai/api
# Kimi K2.6 generations can be slow; mirror the one-off Kimi compose.
OPENAI_TIMEOUT_MS: "300000"
# 2.5 min idle end: single-slot suppliers must free up quickly for the
# shared demo endpoint (the gateway transparently reopens on next turn).
CHAT_IDLE_TIMEOUT_MS: "900000"
CHAT_SETTLE_MODE: ticket
MAX_CHAT_SESSIONS: "8"
PORT: "8080"
# Enable LiveOgmiosProvider so Claim/Submit actually broadcast.
LIVE_CHAIN: "1"
# NETWORK_ID and OGMIOS_URL intentionally not pinned here — they MUST
# come from the operator-provided supplier/.env.kimi-chat-3 on a mainnet host.
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null || exit 1"]
interval: 20s
timeout: 5s
retries: 3
start_period: 15s
networks:
- marketplace-mainnet-net
- traefik-net
labels:
- "traefik.enable=true"
- "traefik.docker.network=traefik-net"
- "traefik.http.routers.mp-suppliers-kimi-chat-3-mainnet.entrypoints=web"
- "traefik.http.routers.mp-suppliers-kimi-chat-3-mainnet.rule=Host(`mp-suppliers-kimi-chat-3.vector.apexfusion.org`)"
- "traefik.http.services.mp-suppliers-kimi-chat-3-mainnet.loadbalancer.server.port=8080"
logging:
driver: json-file
options:
max-size: "10m"
max-file: "3"
networks:
marketplace-mainnet-net:
external: true
name: marketplace-mainnet-net
traefik-net:
external: true
name: traefik-net

View File

@@ -49,8 +49,12 @@ services:
OPENAI_BASE_URL: https://openrouter.ai/api
# Kimi K2.6 generations can be slow; mirror the one-off Kimi compose.
OPENAI_TIMEOUT_MS: "300000"
# Auto-end an idle session after 5 min (supplier Submits the transcript).
CHAT_IDLE_TIMEOUT_MS: "300000"
# 2.5 min idle end (supplier Submits the transcript): single-slot
# suppliers must free up quickly for the shared demo endpoint (the
# gateway transparently reopens a session on the next turn).
CHAT_IDLE_TIMEOUT_MS: "900000"
CHAT_SETTLE_MODE: ticket
MAX_CHAT_SESSIONS: "8"
PORT: "8080"
# Enable LiveOgmiosProvider so Claim/Submit actually broadcast.
LIVE_CHAIN: "1"

View File

@@ -48,8 +48,10 @@ services:
LLM_BACKEND: openai
# No /v1 suffix — supplier/src/openai.ts appends /v1/chat/completions.
OPENAI_BASE_URL: https://openrouter.ai/api
# Kimi K2.6 generations can be slow; mirror the DeepSeek/GPT composes.
OPENAI_TIMEOUT_MS: "300000"
# Raised to match the advert's max_processing_ms=3600000 (60 min) so the
# upstream call isn't aborted before the advertised processing deadline
# (matches the kimi-2/3/4 replicas).
OPENAI_TIMEOUT_MS: "3600000"
PORT: "8080"
# Enable LiveOgmiosProvider so submitTx/awaitTx actually broadcast.
# Without this the supplier silently falls back to ReadOnlyOgmiosProvider

View File

@@ -0,0 +1,168 @@
# HuggingFace Router Supplier — Setup Runbook (V1 / W1)
> **Status:** shipped. Configures a Vector supplier node to use the HuggingFace
> **Inference Providers router** as its **compute backend** for bonded inference
> work commissioned on Vector. Config only — no code changes.
---
## 1. What this is
The HuggingFace **Inference Providers router** (`https://router.huggingface.co`)
is an OpenAI-compatible endpoint that fronts hundreds of hosted models across
partner providers (Together, Fireworks, DeepInfra, Groq, …). It is the direct
analogue of the OpenRouter backend the supplier already supports.
Because the supplier's upstream client (`supplier/src/openai.ts`) already speaks
plain `POST /v1/chat/completions` with a Bearer token, pointing it at HuggingFace
is **pure configuration**: set `LLM_BACKEND=openai` and point `OPENAI_BASE_URL` at
the router. It is the same as adding any OpenAI-compatible compute backend, with
the HuggingFace endpoint and token substituted in.
```
buyer commissions bonded work ──▶ supplier node ──▶ HuggingFace router
(AP3X escrow on Vector) │ (compute backend, hf_ token)
▼
supplier fulfils the work; the escrow settles on-chain
```
The supplier node runs the inference to **fulfil work a buyer has commissioned
under stake**; HuggingFace is simply the compute it uses to produce the result.
**What this does NOT do:** it does not add creator/model-author work-sharing,
change protocol economics, or change how an advert's committed amount is sized.
Those are separate, larger pieces of work tracked in the team's internal planning
notes. This preset is the fast, low-risk on-ramp that proves the path.
---
## 2. Prerequisites
1. A HuggingFace account with **Inference Providers access** (a free monthly
credit allowance, then usage-based per token — this is HuggingFace's billing
to you as the node operator).
2. An access token from <https://huggingface.co/settings/tokens> with the
**"Make calls to Inference Providers"** permission. It starts with `hf_`.
Treat it like a password.
3. A funded supplier wallet (~5 AP3X for tx fees + bonds). Generate a seed with
`openssl rand -hex 32`.
4. The usual supplier deploy prerequisites (Ogmios URL, a publicly reachable
HTTPS endpoint for buyers). See `deploy/README.md`.
---
## 3. Steps
### 3.1 Create the env file
```bash
cp supplier/.env.huggingface-chat.example supplier/.env
chmod 600 supplier/.env
```
Fill in `SUPPLIER_PRIV_KEY_HEX`, `OPENAI_API_KEY` (your `hf_…` token),
`OGMIOS_URL`, and the `ADVERT_*` values. Key settings for this preset:
| Var | Value | Why |
|-----|-------|-----|
| `LLM_BACKEND` | `openai` | routes chat through the OpenAI-compatible client |
| `OPENAI_BASE_URL` | `https://router.huggingface.co` | HF router; the `/v1/chat/completions` suffix is appended by the client — do **not** add it |
| `OPENAI_API_KEY` | `hf_…` | your HF token (Bearer) |
| `OPENAI_REASONING` | **unset** | HF 400s on the OpenRouter-only `reasoning` param (see §5) |
| `ADVERT_MODEL` | provider-qualified id, e.g. `deepseek-ai/DeepSeek-V3:fastest` | see §4 |
### 3.2 Choose a model
HF uses **provider-qualified** model names. The value goes on-chain in
`AdvertDatum.model` and is forwarded verbatim to the router:
- `deepseek-ai/DeepSeek-V3:fastest` — bare id + `:fastest` picks the fastest provider
- `deepseek-ai/DeepSeek-V3:together` — pin a specific provider
- `openai/gpt-oss-120b` — open-weight, HF-hosted
Browse models and their providers at
<https://huggingface.co/models?inference_provider=all>.
Because the model is on-chain data, you can advertise any router-supported chat
model without changing the supplier code — just post a new advert.
### 3.3 Smoke-test the token before going on-chain
Confirm your token + model + base URL work end to end, independent of the chain:
```bash
curl https://router.huggingface.co/v1/chat/completions \
-H "Authorization: Bearer $HF_TOKEN" \
-H "Content-Type: application/json" \
-d '{"model":"deepseek-ai/DeepSeek-V3:fastest",
"messages":[{"role":"user","content":"say hi in 3 words"}],
"stream":false}'
```
A `choices[0].message.content` in the response means the compute backend is
reachable. (Note: **no** `reasoning` field — see §5.)
### 3.4 Post the advert
```bash
pnpm --filter @marketplace/supplier tx:post-advert \
--capability-id llm.text.generate.v1 \
--model deepseek-ai/DeepSeek-V3:fastest \
--max-output-tokens 512 \
--max-processing-ms 60000 \
--price-lovelace 2000000 \
--endpoint-url https://your-supplier.example.org
```
Copy the printed `<txHash>#0` into `ADVERT_REF` in `supplier/.env`. Use
`--dry-run` first to inspect the tx without submitting.
### 3.5 Boot
Start read-only first (`LIVE_CHAIN` unset) to confirm the node boots and can
reach both Ogmios and the HF router, then set `LIVE_CHAIN=1` and restart to
enable real Claim/Submit. Deploy via the supplier compose file as in
`deploy/README.md`.
---
## 4. Sizing the advert (read before you post one)
`ADVERT_PRICE_LOVELACE` is the **flat AP3X amount committed in escrow per job**,
but the HuggingFace backend meters **per token**. The committed amount is fixed
per job while upstream token usage varies, so:
- Size it for your **worst-case** prompt length, not the average.
- Cap output tightly with `--max-output-tokens` — it is your main lever on
upstream token usage.
- Prefer **small** models here; large models have wide token-usage variance and
are harder to size a fixed per-job amount around.
- Buyer **input** tokens are not capped on-chain today. Until an input-token
ceiling lands, stick to small models or add your own request-size guard.
This is a pre-existing protocol design point (fixed per-job advert amount vs.
variable upstream token usage), not specific to HuggingFace. It is the main thing
to account for when sizing an advert.
---
## 5. Footgun: `OPENAI_REASONING` must stay unset
`reasoning:{enabled:false}` is an **OpenRouter-specific** parameter. The HF
router **rejects it with HTTP 400 on every request**. When copying an existing
OpenRouter env, it is easy to carry `OPENAI_REASONING=off` over by accident.
The supplier guards against this: `loadConfig` **refuses to boot** if
`OPENAI_REASONING` is disabled while `OPENAI_BASE_URL` resolves to
`router.huggingface.co`, with a clear error naming the fix
(`supplier/src/config.ts`). Just leave `OPENAI_REASONING` unset for HF.
---
## 6. Where this sits
This preset is the multi-model on-ramp that uses the HuggingFace router as a
compute backend. Further work items (additional task adapters, advert-sizing
refinements, and protocol-economics changes) are tracked separately in the team's
internal planning notes, along with the upstream-terms check every operator must
complete before pointing a node at a third-party compute backend.

View File

@@ -113,6 +113,12 @@ All on-chain steps run inside the per-key `mutex.run(...)`.
- **close** → `endChat` (Submit + Accept, awaits) → settle + bond refund; record usage; wallet-health.
- Sessions are reclaimable by the sweeper if abandoned before close.
### Settle modes (`CHAT_SETTLE_MODE`, gateway + supplier, default `full`)
- **full** — the lifecycle above: supplier Claims at start, Submits a transcript receipt at end, gateway Accepts (buyer charged).
- **ticket** — the buyer's escrow post at open is the ONLY chain op. The supplier verifies the Open escrow as an entry ticket (`/v1/chat/start` → `{status:"ticket"}`) but never Claims/Submits; `/v1/chat/end` → `{status:"ended", settle_mode:"ticket"}` with no receipt; the sweeper Reclaims the Open escrow after `deliver_by`, returning all locked funds (net cost ≈ tx fees). Usage rows record `cost_lovelace: 0`. Sessions may outlive `deliver_by` (no hard-cap) — a live session whose ticket got reclaimed keeps running; only the idle janitor/close ends it. Close behavior is driven per-session by the supplier's `/end` response, so mixed-mode rollout windows settle correctly.
- Ticket caveats (accepted, demo-grade): a supplier restart forgets used tickets, so an unexpired un-reclaimed escrow ref could reopen a session (bounded by `deliver_by` + the sweep); `/account.locked_in_escrow_lovelace` over-reports because the indexer never updates spent rows.
- Suppliers additionally take `MAX_CHAT_SESSIONS` (default 1) — concurrent session slots; `/status` reports `active_sessions`/`max_sessions` and stays `free` until all slots are taken.
## Sweeper & wallet-health
- **Sweeper** (interval ~60s): per key, classify escrows. Open/Claimed past `deliver_by` → `reclaim`. Submitted **with a verified receipt** still in accept-window → retry `acceptResult`. Submitted past accept-window → leave for supplier Release (log). Never blind-`reclaim` Submitted.
- **Wallet-health tick**: periodically + after each settle, `planConsolidate` → `consolidateWallet` so every wallet keeps a ≥5 ADA collateral UTxO. Skip if already healthy.

View File

@@ -30,5 +30,7 @@ GATEWAY_DB_DIR=./data/gateway
# KEY_RATE_MAX=60 # requests per API key per window
# KEY_RATE_WINDOW_MS=60000
# SWEEPER_INTERVAL_MS=60000
# DEMO_SESSION_IDLE_MS=180000 # close demo chat sessions idle this long (frees the supplier slot)
# CHAT_SETTLE_MODE=full # "full" (Claim/Submit/Accept) or "ticket" (escrow post only; sweeper reclaims) — must match suppliers
# WALLET_HEALTH_INTERVAL_MS=600000
# SDK_REGISTRY_MAX=500 # max cached per-key Marketplace instances

View File

@@ -7,7 +7,8 @@
"typecheck": "tsc --noEmit",
"start": "tsx src/index.ts",
"dev": "tsx src/index.ts",
"rotate-master-key": "tsx src/cli/rotate-master-key.ts"
"rotate-master-key": "tsx src/cli/rotate-master-key.ts",
"create-demo-key": "tsx src/cli/create-demo-key.ts"
},
"dependencies": {
"@marketplace/buyer": "workspace:*",

View File

@@ -15,7 +15,7 @@ import { seal } from "../crypto/seal.js";
import { hashApiKey, requireKey } from "../middleware/apiKeyAuth.js";
import { asyncHandler } from "../middleware/http.js";
import { totalLovelace, hasCollateral } from "../onchain/preflight.js";
import { badRequest } from "../openai/errors.js";
import { badRequest, forbidden } from "../openai/errors.js";
const ACTIVE_ESCROW_STATES = new Set(["Open", "Claimed", "Submitted"]);
@@ -89,7 +89,10 @@ export function makeAccountHandler(deps: GatewayDeps) {
res.status(200).json({
key_prefix: keyRow.key_prefix,
deposit_address: keyRow.deposit_address,
// The shared demo key's wallet is operator-funded; hiding the deposit
// address keeps strangers from confusing it with their own wallet.
deposit_address: keyRow.demo ? null : keyRow.deposit_address,
...(keyRow.demo ? { demo: true } : {}),
balance: {
available_lovelace: available.toString(),
locked_in_escrow_lovelace: lockedInEscrow.toString(),
@@ -118,6 +121,11 @@ export function makeAccountHandler(deps: GatewayDeps) {
export function makeWithdrawHandler(deps: GatewayDeps) {
return asyncHandler(async (req: Request, res: Response) => {
const keyRow = requireKey(req);
if (keyRow.demo) {
// The demo key is shared publicly; without this, anyone holding it
// could drain the operator-funded wallet to their own address.
throw forbidden("demo_key_restricted", "withdrawals are disabled for the shared demo key");
}
const body = (req.body ?? {}) as Record<string, unknown>;
const toAddress = body.to_address;

View File

@@ -0,0 +1,76 @@
/**
* gateway/src/cli/create-demo-key.ts — mint the shared demo API key.
*
* Same custodial machinery as POST /signup, but sets demo=1: withdraw is
* blocked, /account hides the deposit address, chat completions run through
* the session-backed demo executor, and per-IP rate limits apply. Public
* signup can never mint a demo key — this CLI is the only path.
*
* Run inside the gateway container (hits the volume-mounted DB):
* docker exec <gateway> pnpm --filter @marketplace/gateway create-demo-key
* Or locally:
* GATEWAY_MASTER_KEY=<64hex> GATEWAY_DB_DIR=./data/gateway NETWORK_ID=1 \
* pnpm --filter @marketplace/gateway create-demo-key [--label shared-demo]
*
* The raw key is printed ONCE and never stored (only its sha256).
*/
import { randomBytes, randomUUID } from "crypto";
import { GatewayStore } from "../db/store.js";
import { genPrivKeyHex, deriveWalletKey } from "../wallet.js";
import { seal } from "../crypto/seal.js";
import { hashApiKey } from "../middleware/apiKeyAuth.js";
const HEX64 = /^[0-9a-fA-F]{64}$/;
function arg(name: string): string | undefined {
const i = process.argv.indexOf(`--${name}`);
return i >= 0 ? process.argv[i + 1] : undefined;
}
function main(): void {
const masterKey = process.env.GATEWAY_MASTER_KEY;
if (!masterKey || !HEX64.test(masterKey)) {
throw new Error("create-demo-key: GATEWAY_MASTER_KEY must be 64 hex chars");
}
const dbDir = arg("db") ?? process.env.GATEWAY_DB_DIR ?? "./data/gateway";
const networkId: 0 | 1 = process.env.NETWORK_ID === "1" ? 1 : 0;
const label = arg("label") ?? "shared-demo";
const privHex = genPrivKeyHex();
const walletKey = deriveWalletKey(privHex, networkId);
const sealed = seal(privHex, masterKey);
const rawKey = `vk_${networkId === 1 ? "live" : "test"}_${randomBytes(24).toString("hex")}`;
const store = new GatewayStore(dbDir);
store.insertKey({
id: randomUUID(),
key_hash: hashApiKey(rawKey),
key_prefix: rawKey.slice(0, 12),
label,
wallet_pkh: walletKey.pubKeyHash,
deposit_address: walletKey.address,
enc_priv_nonce: sealed.nonce,
enc_priv_ct: sealed.ct,
enc_priv_tag: sealed.tag,
master_key_version: 1,
created_at: Date.now(),
demo: 1,
});
// eslint-disable-next-line no-console
console.log(
[
`[gateway] demo key created (label=${label}, network=${networkId === 1 ? "mainnet" : "testnet"})`,
``,
` api_key: ${rawKey}`,
` deposit_address: ${walletKey.address}`,
``,
`Save the api_key now — it is shown only once. Fund the deposit address`,
`with the demo float (withdrawals are disabled for this key).`,
].join("\n"),
);
}
main();

View File

@@ -20,6 +20,10 @@
* KEY_RATE_MAX / KEY_RATE_WINDOW_MS — per-key request limiter.
* SWEEPER_INTERVAL_MS / WALLET_HEALTH_INTERVAL_MS — background tick cadences.
* SDK_REGISTRY_MAX — max cached per-key Marketplace instances (LRU).
* GATEWAY_CORS_ORIGINS — comma-separated allowlist of browser Origins that may
* call the gateway cross-origin (e.g. the marketplace
* frontend's "Generate API key" page). Default empty =
* CORS off (no Access-Control-* headers emitted).
*/
const HEX64_RE = /^[0-9a-fA-F]{64}$/;
@@ -35,9 +39,20 @@ export interface GatewayConfig {
dbDir: string;
signupRate: { max: number; windowMs: number };
keyRate: { max: number; windowMs: number };
demoIpRate: { max: number; windowMs: number };
sweeperIntervalMs: number;
/** Close demo chat sessions idle this long (frees the supplier slot).
* Keep below the supplier's CHAT_IDLE_TIMEOUT_MS so the gateway closes
* first and records usage. */
demoSessionIdleMs: number;
/** Chat-session settlement mode this deployment runs. "full" = suppliers
* Claim/Submit and the gateway Accepts; "ticket" = the escrow post at open
* is the only chain op and the sweeper Reclaims it after deliver_by.
* Must agree with the suppliers' CHAT_SETTLE_MODE. */
chatSettleMode: "full" | "ticket";
walletHealthIntervalMs: number;
sdkRegistryMax: number;
corsOrigins: string[];
}
function requireField(env: Record<string, string | undefined>, name: string): string {
@@ -48,6 +63,14 @@ function requireField(env: Record<string, string | undefined>, name: string): st
return v;
}
function chatSettleMode(env: Record<string, string | undefined>): "full" | "ticket" {
const v = env.CHAT_SETTLE_MODE ?? "full";
if (v !== "full" && v !== "ticket") {
throw new Error('loadConfig: CHAT_SETTLE_MODE must be "full" or "ticket"');
}
return v;
}
function posInt(
env: Record<string, string | undefined>,
name: string,
@@ -106,8 +129,18 @@ export function loadConfig(env: Record<string, string | undefined>): GatewayConf
max: posInt(env, "KEY_RATE_MAX", 60),
windowMs: posInt(env, "KEY_RATE_WINDOW_MS", 60 * 1000),
},
demoIpRate: {
max: posInt(env, "DEMO_IP_RATE_MAX", 20),
windowMs: posInt(env, "DEMO_IP_RATE_WINDOW_MS", 60 * 1000),
},
sweeperIntervalMs: posInt(env, "SWEEPER_INTERVAL_MS", 60 * 1000),
demoSessionIdleMs: posInt(env, "DEMO_SESSION_IDLE_MS", 3 * 60 * 1000),
chatSettleMode: chatSettleMode(env),
walletHealthIntervalMs: posInt(env, "WALLET_HEALTH_INTERVAL_MS", 10 * 60 * 1000),
sdkRegistryMax: posInt(env, "SDK_REGISTRY_MAX", 500),
corsOrigins: (env.GATEWAY_CORS_ORIGINS ?? "")
.split(",")
.map((o) => o.trim().replace(/\/+$/, ""))
.filter((o) => o !== ""),
};
}

View File

@@ -23,7 +23,8 @@ CREATE TABLE IF NOT EXISTS api_keys (
enc_priv_tag TEXT NOT NULL,
master_key_version INTEGER NOT NULL DEFAULT 1,
created_at INTEGER NOT NULL,
disabled INTEGER NOT NULL DEFAULT 0
disabled INTEGER NOT NULL DEFAULT 0,
demo INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_keys_hash ON api_keys(key_hash);

View File

@@ -24,6 +24,8 @@ export interface ApiKeyRow {
master_key_version: number;
created_at: number;
disabled: number;
/** 1 = shared demo key: withdraw blocked, session-backed completions, per-IP limits. */
demo: number;
}
export interface UsageRow {
@@ -81,13 +83,20 @@ export class GatewayStore {
this.db.pragma("journal_mode = WAL");
this.db.exec(CREATE_TABLES_SQL);
// Additive migration for DBs created before the demo flag existed
// (CREATE TABLE IF NOT EXISTS won't alter an existing table).
const keyCols = this.db.pragma("table_info(api_keys)") as Array<{ name: string }>;
if (!keyCols.some((c) => c.name === "demo")) {
this.db.exec("ALTER TABLE api_keys ADD COLUMN demo INTEGER NOT NULL DEFAULT 0");
}
this.sInsertKey = this.db.prepare(`
INSERT INTO api_keys (
id, key_hash, key_prefix, label, wallet_pkh, deposit_address,
enc_priv_nonce, enc_priv_ct, enc_priv_tag, master_key_version, created_at, disabled
enc_priv_nonce, enc_priv_ct, enc_priv_tag, master_key_version, created_at, disabled, demo
) VALUES (
@id, @key_hash, @key_prefix, @label, @wallet_pkh, @deposit_address,
@enc_priv_nonce, @enc_priv_ct, @enc_priv_tag, @master_key_version, @created_at, 0
@enc_priv_nonce, @enc_priv_ct, @enc_priv_tag, @master_key_version, @created_at, 0, @demo
)
`);
this.sGetKeyByHash = this.db.prepare(`SELECT * FROM api_keys WHERE key_hash = @key_hash`);
@@ -133,8 +142,8 @@ export class GatewayStore {
this.sListOpenSessions = this.db.prepare(`SELECT * FROM sessions WHERE state = 'open'`);
}
insertKey(row: Omit<ApiKeyRow, "disabled">): void {
this.sInsertKey.run({ ...row, label: row.label ?? null });
insertKey(row: Omit<ApiKeyRow, "disabled" | "demo"> & { demo?: number }): void {
this.sInsertKey.run({ ...row, label: row.label ?? null, demo: row.demo ?? 0 });
}
getKeyByHash(keyHash: string): ApiKeyRow | undefined {
@@ -193,4 +202,8 @@ export class GatewayStore {
listOpenSessions(): SessionRow[] {
return this.sListOpenSessions.all() as SessionRow[];
}
listOpenSessionsByKey(keyId: string): SessionRow[] {
return this.listOpenSessions().filter((s) => s.key_id === keyId);
}
}

View File

@@ -60,7 +60,28 @@ export function ipRateLimit(opts: { max: number; windowMs: number }) {
return makeLimiter(opts, (req) => req.ip ?? "unknown");
}
/** Per-API-key limiter (apply AFTER apiKeyAuth). */
/** Per-API-key limiter (apply AFTER apiKeyAuth). Demo keys are exempt: the
* shared demo key would let one abusive IP exhaust everyone's budget — the
* per-IP demoIpRateLimit is the demo control instead. */
export function keyRateLimit(opts: { max: number; windowMs: number }) {
return makeLimiter(opts, (req) => requireKey(req).id);
const limit = makeLimiter(opts, (req) => requireKey(req).id);
return (req: Request, res: Response, next: NextFunction): void => {
if (requireKey(req).demo) {
next();
return;
}
limit(req, res, next);
};
}
/** Per-IP limiter that applies ONLY to demo-key requests (after apiKeyAuth). */
export function demoIpRateLimit(opts: { max: number; windowMs: number }) {
const limit = makeLimiter(opts, (req) => req.ip ?? "unknown");
return (req: Request, res: Response, next: NextFunction): void => {
if (!requireKey(req).demo) {
next();
return;
}
limit(req, res, next);
};
}

View File

@@ -22,6 +22,8 @@ export interface EscrowRow {
utxo_ref: string;
state: string;
posted_at: number;
/** Validator reclaim floor (indexer serves it; absent on very old rows). */
deliver_by?: number;
}
export async function fetchEscrows(

View File

@@ -0,0 +1,116 @@
/**
* gateway/src/openai/affinity.ts — transcript-prefix session matching.
*
* Stateless OpenAI clients resend the FULL message history each call; sessions
* are stateful. An incoming request belongs to an open session when that
* session's mirrored transcript is a prefix of the incoming messages — the
* suffix (delta) is the next turn. Client-authored messages (user/system/tool)
* must match byte-exactly, but clients routinely reshape ASSISTANT text when
* echoing it back (strip <think> reasoning, re-render whitespace), so
* assistant content is compared normalized. Strict (byte-exact) matches are
* preferred over lenient ones to disambiguate concurrent threads that share a
* user-message prefix. The mirror itself is NEVER rewritten with client text:
* the delta is forwarded verbatim and mirrored only after a successful turn,
* so the gateway mirror and the supplier transcript stay hash-identical for
* the receipt response_hash.
*
* Pure module — no I/O — so the matching rules are unit-testable.
*/
import { chatMessagesEquivalent, toolCallsEqual, type ChatMessage } from "@marketplace/shared/tx";
export interface AffinityCandidate {
sessionId: string;
model: string;
mirror: ChatMessage[];
lastUsedAt: number;
}
export interface AffinityMatch {
sessionId: string;
/** Messages the session has not seen yet — forwarded as the next turn. */
delta: ChatMessage[];
}
/** Canonicalize assistant text the way clients mangle it when re-echoing
* history: drop <think>…</think> reasoning blocks, collapse whitespace runs,
* trim. Exported for tests. */
export function normalizeAssistantContent(content: string): string {
return content
.replace(/<think>[\s\S]*?<\/think>/gi, "")
.replace(/\s+/g, " ")
.trim();
}
/** Lenient equality: strict for client-authored roles (the client sends the
* same bytes both times); assistant content is compared normalized while
* assistant tool_calls stay strict (ids anchor tool-result turns). */
function chatMessagesLenientEquivalent(a: ChatMessage, b: ChatMessage): boolean {
if (a.role !== b.role) return false;
if (a.role !== "assistant") return chatMessagesEquivalent(a, b);
return (
normalizeAssistantContent(a.content) === normalizeAssistantContent(b.content) &&
toolCallsEqual(a.tool_calls, b.tool_calls)
);
}
type PrefixQuality = "strict" | "lenient";
function classifyPrefix(mirror: ChatMessage[], incoming: ChatMessage[]): PrefixQuality | null {
if (mirror.length >= incoming.length) return null;
let strict = true;
for (let i = 0; i < mirror.length; i++) {
if (chatMessagesEquivalent(mirror[i], incoming[i])) continue;
if (!chatMessagesLenientEquivalent(mirror[i], incoming[i])) return null;
strict = false;
}
return strict ? "strict" : "lenient";
}
function deltaRespondable(delta: ChatMessage[]): boolean {
const last = delta[delta.length - 1];
return last !== undefined && (last.role === "user" || last.role === "tool");
}
/** Drop trailing assistant/system messages so the history ends with a
* respondable message (regenerate semantics for clients that send a trailing
* assistant turn). Empty result = nothing to answer. */
export function trimToRespondable(messages: ChatMessage[]): ChatMessage[] {
let end = messages.length;
while (end > 0 && messages[end - 1].role !== "user" && messages[end - 1].role !== "tool") end--;
return messages.slice(0, end);
}
/**
* Pick the open session whose mirror best matches the incoming history.
* Strict prefixes beat lenient ones; then longest mirror wins (most
* conversation reuse); ties go to the most recently used session. Returns
* null when nothing matches — caller opens a new session and replays the full
* history. Note `incoming.length <= mirror.length` never matches (a client
* retry of an already-answered request is a miss by design: the session
* transcript must stay append-only).
*/
export function matchSessionPrefix(
candidates: AffinityCandidate[],
incoming: ChatMessage[],
model: string,
): AffinityMatch | null {
let best: { cand: AffinityCandidate; quality: PrefixQuality } | null = null;
for (const cand of candidates) {
if (cand.model !== model) continue;
const quality = classifyPrefix(cand.mirror, incoming);
if (quality === null) continue;
if (!deltaRespondable(incoming.slice(cand.mirror.length))) continue;
if (
best === null ||
(quality === "strict" && best.quality === "lenient") ||
(quality === best.quality &&
(cand.mirror.length > best.cand.mirror.length ||
(cand.mirror.length === best.cand.mirror.length && cand.lastUsedAt > best.cand.lastUsedAt)))
) {
best = { cand, quality };
}
}
if (!best) return null;
return { sessionId: best.cand.sessionId, delta: incoming.slice(best.cand.mirror.length) };
}

View File

@@ -21,6 +21,7 @@ import { preflight } from "../onchain/preflight.js";
import { resolveSubmittedRef, acceptAndConfirm } from "../onchain/settle.js";
import { ensureWalletHealthy } from "../walletHealth.js";
import { parseChatRequest, type ParsedChatRequest } from "./validate.js";
import { runDemoChat } from "./demoChat.js";
import {
genId,
buildChatCompletion,
@@ -67,8 +68,17 @@ function recordUsage(
export function makeChatCompletionsHandler(deps: GatewayDeps) {
return asyncHandler(async (req: Request, res: Response) => {
const keyRow = requireKey(req);
const parsed = parseChatRequest(req.body);
const ctx = deps.registry.getContext(keyRow);
if (keyRow.demo) {
// Demo keys run session-backed with tools passthrough, and never fall
// back to one-shot (its settle path assumes ≤1 in-flight escrow per
// wallet; the demo wallet holds one escrow per concurrent session).
// Not under the key mutex — runDemoChat takes it only for session opens.
const parsed = parseChatRequest(req.body, { allowTools: true });
await runDemoChat(deps, keyRow, ctx, parsed, res);
return;
}
const parsed = parseChatRequest(req.body);
await ctx.mutex.run(() => runOneShot(deps, keyRow, ctx, parsed, res));
});
}

View File

@@ -0,0 +1,413 @@
/**
* gateway/src/openai/demoChat.ts — session-backed /chat/completions for the
* shared demo key.
*
* Demo requests keep the standard OpenAI wire shape (tools included) but run
* against real escrow-per-session marketplace conversations: transcript-prefix
* affinity maps each stateless request onto an open session and forwards only
* the unseen delta; a miss transparently opens a new session (escrow post —
* the only chain work, under the key mutex) and replays the full history as
* the first turn. Supplier idle-closes are healed by reopening; the sweeper
* settles the abandoned escrow.
*
* Demo keys NEVER fall back to the one-shot path: resolveSubmittedRef's
* lone-Submitted-row fallback assumes ≤1 in-flight escrow per wallet, and the
* demo wallet holds one escrow per concurrent session.
*/
import { randomUUID } from "crypto";
import type { Response } from "express";
import type { ChatMessage, ToolCall } from "@marketplace/shared/tx";
import type { GatewayDeps } from "../deps.js";
import type { ApiKeyRow, SessionRow } from "../db/store.js";
import type { KeyContext } from "../sdk/registry.js";
import { transcripts, getSessionLock, dropSessionState } from "./transcripts.js";
import { matchSessionPrefix, trimToRespondable, type AffinityCandidate } from "./affinity.js";
import {
CAPABILITY,
SessionGoneError,
openSessionCore,
streamSupplierTurn,
type TurnResult,
} from "./sessions.js";
import type { ParsedChatRequest } from "./validate.js";
import { genId, buildChatCompletion, buildChunk, sseData, SSE_DONE, type Usage } from "./shapes.js";
import { GatewayError, badRequest, toGatewayError, toErrorBody } from "./errors.js";
interface DemoSessionMeta {
keyId: string;
model: string;
lastUsedAt: number;
}
/** Open sessions owned by the demo executor (sessionId → meta). Mirrors live
* in transcripts.ts; this map scopes affinity to demo-opened sessions. */
const demoSessions = new Map<string, DemoSessionMeta>();
/** Forget a session (closed/settled elsewhere, e.g. by the sweeper). */
export function dropDemoSession(sessionId: string): void {
demoSessions.delete(sessionId);
}
function candidatesFor(keyId: string, model: string): AffinityCandidate[] {
const out: AffinityCandidate[] = [];
for (const [sessionId, meta] of demoSessions) {
if (meta.keyId !== keyId || meta.model !== model) continue;
const mirror = transcripts.get(sessionId);
if (!mirror) continue;
out.push({ sessionId, model: meta.model, mirror, lastUsedAt: meta.lastUsedAt });
}
return out;
}
/** Record the session's cost when the demo executor closes its row. Full
* mode: the flat session fee (supplier Submitted; the sweeper Accepts the
* escrow afterwards). Ticket mode: 0 — nothing settles, the sweeper reclaims
* the escrow and the wallet is made whole minus fees. */
function recordSessionCost(deps: GatewayDeps, session: SessionRow): void {
deps.store.insertUsage({
id: randomUUID(),
key_id: session.key_id,
created_at: Date.now(),
kind: "chat_session",
model: session.model,
capability_id: CAPABILITY,
supplier_pkh: session.supplier_pkh,
escrow_ref: session.escrow_ref,
cost_lovelace: deps.config.chatSettleMode === "ticket" ? "0" : session.price_lovelace,
prompt_tokens: 0,
completion_tokens: 0,
status: "completed",
failure_reason: null,
});
}
function markSessionGone(deps: GatewayDeps, session: SessionRow): void {
deps.store.setSessionState(session.id, "closed", Date.now());
dropSessionState(session.id);
dropDemoSession(session.id);
recordSessionCost(deps, session);
}
/** Ask the supplier to settle (Submit) the session. NO gateway chain work —
* the sweeper Accepts the resulting Submitted escrow. The supplier releases
* its single slot as part of ending, so a truthy return means the supplier is
* free again. Best-effort: on failure the sweeper reclaims later. */
async function requestSupplierEnd(deps: GatewayDeps, session: SessionRow): Promise<boolean> {
try {
const res = await deps.fetchFn(`${session.supplier_base_url.replace(/\/+$/, "")}/v1/chat/end`, {
method: "POST",
headers: { "content-type": "application/json", "X-Escrow-Ref": session.escrow_ref },
body: JSON.stringify({}),
signal: AbortSignal.timeout(90_000),
});
await res.text().catch(() => "");
return res.ok;
} catch {
return false;
}
}
/** Close one demo session (evicted for capacity, or idle-janitored). Runs
* under the session lock so an in-flight turn finishes first; takes NO key
* mutex (no chain work here — see requestSupplierEnd). Returns the freed
* supplier_pkh, or null when the session was already gone or (janitor) was
* used again while we waited for the lock. */
async function closeDemoSession(
deps: GatewayDeps,
sessionId: string,
opts: { reason: "evicted" | "idle"; idleCutoffMs?: number; nowMs?: number },
): Promise<string | null> {
if (!demoSessions.has(sessionId)) return null;
return getSessionLock(sessionId).run(async () => {
const meta = demoSessions.get(sessionId);
if (!meta) return null; // lost a race with another closer
// Re-check idleness inside the lock: a turn that won the lock first has
// bumped lastUsedAt past the sweep's reference clock.
if (opts.idleCutoffMs !== undefined && (opts.nowMs ?? Date.now()) - meta.lastUsedAt < opts.idleCutoffMs) {
return null;
}
const session = deps.store.getSession(sessionId);
if (!session || session.state !== "open") {
dropDemoSession(sessionId);
return null;
}
await requestSupplierEnd(deps, session);
markSessionGone(deps, session);
// eslint-disable-next-line no-console
console.log(`[gateway:demo] closed session ${sessionId} (${opts.reason}), freed supplier ${session.supplier_pkh}`);
return session.supplier_pkh;
});
}
/** Least-recently-used open demo session for (key, model), excluding prior
* victims. Same-model scoping doubles as a guard: a bogus model name has no
* demo sessions, so eviction can never loop on a genuine model_not_found. */
function pickLruDemoSession(keyId: string, model: string, exclude: ReadonlySet<string>): string | null {
let lruId: string | null = null;
let lruAt = Infinity;
for (const [sessionId, meta] of demoSessions) {
if (meta.keyId !== keyId || meta.model !== model || exclude.has(sessionId)) continue;
if (meta.lastUsedAt < lruAt) {
lruAt = meta.lastUsedAt;
lruId = sessionId;
}
}
return lruId;
}
/** Close demo sessions idle past deps.config.demoSessionIdleMs so abandoned
* conversations (client gone, title-gen one-offs) release their single-slot
* suppliers. Called from the sweeper tick; sequential to keep supplier
* /v1/chat/end calls orderly. */
export async function sweepIdleDemoSessions(deps: GatewayDeps, nowMs = Date.now()): Promise<void> {
const idleMs = deps.config.demoSessionIdleMs;
for (const [sessionId, meta] of [...demoSessions]) {
if (nowMs - meta.lastUsedAt < idleMs) continue;
try {
await closeDemoSession(deps, sessionId, { reason: "idle", idleCutoffMs: idleMs, nowMs });
} catch (err) {
// eslint-disable-next-line no-console
console.error(`[gateway:demo] idle close of ${sessionId} failed:`, err instanceof Error ? err.message : err);
}
}
}
const MAX_EVICTIONS = 2;
function isNoSupplierError(err: unknown): boolean {
return err instanceof GatewayError && (err.code === "model_not_found" || err.code === "overloaded");
}
/** Open a session, evicting up to MAX_EVICTIONS LRU demo sessions when every
* supplier is pinned. Eviction frees the supplier immediately (its slot
* releases at Submit) but the indexer's status poll lags, so the retry passes
* the freed pkhs through selectCandidates' ignoreStatusFor. Lock ordering is
* safe: eviction holds only the session lock, the open holds only the key
* mutex — never nested. */
async function openSessionWithEviction(
deps: GatewayDeps,
keyRow: ApiKeyRow,
ctx: KeyContext,
model: string,
): Promise<SessionRow> {
const freedPkhs = new Set<string>();
const evicted = new Set<string>();
for (let attempt = 0; ; attempt++) {
try {
return await ctx.mutex.run(() =>
openSessionCore(deps, keyRow, ctx, model, freedPkhs.size > 0 ? { ignoreStatusFor: freedPkhs } : undefined),
);
} catch (err) {
if (attempt >= MAX_EVICTIONS || !isNoSupplierError(err)) throw err;
const victimId = pickLruDemoSession(keyRow.id, model, evicted);
if (!victimId) throw err;
evicted.add(victimId);
const pkh = await closeDemoSession(deps, victimId, { reason: "evicted" });
if (!pkh) throw err; // nothing actually freed — don't loop
freedPkhs.add(pkh);
}
}
}
type LockedTurnOutcome =
| { kind: "done"; result: TurnResult }
| { kind: "retry" } // candidate no longer valid — re-run the match
| { kind: "gone" }; // supplier lost the session — open a fresh one
/** Streaming sink shared across turn attempts on one response. */
interface StreamSink {
onToken: (token: string) => void;
emitToolCalls: (toolCalls: ToolCall[]) => void;
}
export async function runDemoChat(
deps: GatewayDeps,
keyRow: ApiKeyRow,
ctx: KeyContext,
parsed: ParsedChatRequest,
res: Response,
): Promise<void> {
// Clients sometimes send histories ending in an assistant (continue/retry
// flows) or system message; the chat.v1 supplier can only respond to a
// user/tool-terminated turn. Trim trailing unrespondable messages
// (regenerate semantics) and reject — before any chain work — only when
// nothing respondable remains.
const trimmed = trimToRespondable(parsed.messages);
if (trimmed.length === 0) {
throw badRequest("invalid_messages", "no user or tool message to respond to");
}
const effective: ParsedChatRequest =
trimmed.length === parsed.messages.length ? parsed : { ...parsed, messages: trimmed };
const id = genId();
let keepalive: ReturnType<typeof setInterval> | undefined;
let roleSent = false;
if (parsed.stream) {
res.status(200);
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache, no-transform");
res.setHeader("Connection", "keep-alive");
res.setHeader("X-Accel-Buffering", "no");
res.write(": connected\n\n");
keepalive = setInterval(() => {
try {
res.write(": keepalive\n\n");
} catch {
/* client gone */
}
}, 10_000);
}
const ensureRole = (): void => {
if (roleSent) return;
roleSent = true;
res.write(sseData(buildChunk({ id, model: parsed.model, delta: { role: "assistant" }, finishReason: null })));
};
const sink: StreamSink = {
onToken: (token) => {
if (!parsed.stream) return;
ensureRole();
res.write(sseData(buildChunk({ id, model: parsed.model, delta: { content: token }, finishReason: null })));
},
emitToolCalls: (toolCalls) => {
if (!parsed.stream) return;
ensureRole();
res.write(
sseData(
buildChunk({
id,
model: parsed.model,
delta: { tool_calls: toolCalls.map((tc, index) => ({ index, ...tc })) },
finishReason: null,
}),
),
);
},
};
try {
const result = await runWithAffinity(deps, keyRow, ctx, effective, sink);
if (keepalive) clearInterval(keepalive);
const usage: Usage = {
prompt_tokens: result.usage.prompt_tokens,
completion_tokens: result.usage.completion_tokens,
total_tokens: result.usage.prompt_tokens + result.usage.completion_tokens,
};
if (parsed.stream) {
if (result.assistantMessage.tool_calls?.length) sink.emitToolCalls(result.assistantMessage.tool_calls);
ensureRole();
res.write(sseData(buildChunk({ id, model: parsed.model, delta: {}, finishReason: result.finishReason, usage })));
res.write(SSE_DONE);
res.end();
} else {
res.status(200).json(
buildChatCompletion({
id,
model: parsed.model,
content: result.assistantMessage.content,
toolCalls: result.assistantMessage.tool_calls,
finishReason: result.finishReason,
usage,
}),
);
}
} catch (err) {
if (keepalive) clearInterval(keepalive);
if (parsed.stream && res.headersSent) {
try {
res.write(sseData(toErrorBody(toGatewayError(err))));
} catch {
/* client gone */
}
res.end();
return;
}
throw err;
}
}
async function runWithAffinity(
deps: GatewayDeps,
keyRow: ApiKeyRow,
ctx: KeyContext,
parsed: ParsedChatRequest,
sink: StreamSink,
): Promise<TurnResult> {
// Affinity: try matching sessions until one serves the turn. The match runs
// synchronously (no await between match and lock acquire, so the candidate
// set can't shift underneath us); the prefix is re-checked inside the lock
// because a concurrent request may have advanced the session first.
for (let attempt = 0; attempt < 3; attempt++) {
const match = matchSessionPrefix(candidatesFor(keyRow.id, parsed.model), parsed.messages, parsed.model);
if (!match) break;
const outcome: LockedTurnOutcome = await getSessionLock(match.sessionId).run(async () => {
const meta = demoSessions.get(match.sessionId);
const mirror = transcripts.get(match.sessionId);
const session = deps.store.getSession(match.sessionId);
if (!meta || !mirror || !session || session.state !== "open") {
dropDemoSession(match.sessionId);
return { kind: "retry" };
}
const recheck = matchSessionPrefix(
[{ sessionId: session.id, model: meta.model, mirror, lastUsedAt: meta.lastUsedAt }],
parsed.messages,
parsed.model,
);
if (!recheck) return { kind: "retry" };
try {
const result = await streamSupplierTurn(
deps,
session,
{ messages: recheck.delta, tools: parsed.tools, tool_choice: parsed.toolChoice },
sink.onToken,
);
meta.lastUsedAt = Date.now();
return { kind: "done", result };
} catch (err) {
if (err instanceof SessionGoneError) {
// Supplier idle-closed (or restarted). Close our row, record the
// fee (the escrow settles via the sweeper) and reopen fresh.
markSessionGone(deps, session);
return { kind: "gone" };
}
throw err;
}
});
if (outcome.kind === "done") return outcome.result;
if (outcome.kind === "gone") break;
// retry: candidate vanished/advanced — re-match against current state.
}
// Miss: open a new session (chain work → key mutex; evicts an LRU demo
// session when every supplier is pinned) and replay the full history as its
// first turn.
const session = await openSessionWithEviction(deps, keyRow, ctx, parsed.model);
demoSessions.set(session.id, { keyId: keyRow.id, model: parsed.model, lastUsedAt: Date.now() });
return getSessionLock(session.id).run(async () => {
try {
const result = await streamSupplierTurn(
deps,
session,
{ messages: parsed.messages, tools: parsed.tools, tool_choice: parsed.toolChoice },
sink.onToken,
);
// Refresh recency after the (possibly long) first stream so this live
// thread isn't the LRU/janitor's first victim.
const meta = demoSessions.get(session.id);
if (meta) meta.lastUsedAt = Date.now();
return result;
} catch (err) {
if (err instanceof SessionGoneError) {
// Freshly opened yet already gone (supplier crashed mid-claim).
// Surface as an upstream error rather than looping.
markSessionGone(deps, session);
}
throw err;
}
});
}

View File

@@ -36,6 +36,8 @@ export const unauthorized = (message = "invalid API key"): GatewayError =>
new GatewayError(401, "authentication_error", "invalid_api_key", message);
export const paymentRequired = (message: string, extra?: Record<string, unknown>): GatewayError =>
new GatewayError(402, "invalid_request_error", "insufficient_funds", message, extra);
export const forbidden = (code: string, message: string): GatewayError =>
new GatewayError(403, "invalid_request_error", code, message);
export const notFound = (code: string, message: string): GatewayError =>
new GatewayError(404, "invalid_request_error", code, message);
export const rateLimited = (message: string): GatewayError =>

View File

@@ -7,12 +7,20 @@
import type { Request, Response } from "express";
import type { GatewayDeps } from "../deps.js";
import { asyncHandler } from "../middleware/http.js";
import { requireKey } from "../middleware/apiKeyAuth.js";
import { listModels } from "../routing/selectSupplier.js";
import { buildModelsList } from "./shapes.js";
import { CAPABILITY as CHAT_CAPABILITY } from "./sessions.js";
export function makeModelsHandler(deps: GatewayDeps) {
return asyncHandler(async (_req: Request, res: Response) => {
const models = await listModels({ indexerUrl: deps.config.indexerUrl, fetchFn: deps.fetchFn });
return asyncHandler(async (req: Request, res: Response) => {
const keyRow = requireKey(req);
// Demo keys only reach chat-capable suppliers (session-backed executor).
const models = await listModels({
indexerUrl: deps.config.indexerUrl,
capabilityId: keyRow.demo ? CHAT_CAPABILITY : undefined,
fetchFn: deps.fetchFn,
});
res.status(200).json(buildModelsList(models));
});
}

View File

@@ -15,8 +15,9 @@
import { randomUUID } from "crypto";
import type { Request, Response } from "express";
import { TxConstructionError, type ChatMessage } from "@marketplace/shared/tx";
import { TxConstructionError, normalizeChatMessage, type ChatMessage } from "@marketplace/shared/tx";
import { SupplierError } from "@marketplace/buyer/sdk";
import { transcripts, getSessionLock, dropSessionState } from "./transcripts.js";
import type { GatewayDeps } from "../deps.js";
import type { ApiKeyRow, GatewayStore, SessionRow } from "../db/store.js";
import type { KeyContext } from "../sdk/registry.js";
@@ -36,10 +37,15 @@ import {
type Usage,
} from "./shapes.js";
const CAPABILITY = "llm.chat.v1";
export const CAPABILITY = "llm.chat.v1";
/** In-memory per-session transcript mirror (sessionId → messages). */
const transcripts = new Map<string, ChatMessage[]>();
/** Thrown when the supplier no longer knows the session (idle-closed/restart). */
export class SessionGoneError extends Error {
constructor(sessionId: string) {
super(`supplier no longer has session ${sessionId}`);
this.name = "SessionGoneError";
}
}
function refStr(ref: { txHash: string; index: number }): string {
return `${ref.txHash}#${ref.index}`;
@@ -56,22 +62,29 @@ export function makeOpenSessionHandler(deps: GatewayDeps) {
throw badRequest("invalid_model", "`model` is required");
}
const ctx = deps.registry.getContext(keyRow);
await ctx.mutex.run(() => openSession(deps, keyRow, ctx, model, res));
const session = await ctx.mutex.run(() => openSessionCore(deps, keyRow, ctx, model));
res.status(200).json({ id: session.id, object: "chat.session", model, created: nowSec() });
});
}
async function openSession(
/**
* Select a supplier, preflight the wallet, post+claim the session escrow and
* persist the SessionRow. Chain work — the CALLER must hold the key mutex
* (concurrent PostEscrow from one wallet double-spends the same UTxO).
*/
export async function openSessionCore(
deps: GatewayDeps,
keyRow: ApiKeyRow,
ctx: KeyContext,
model: string,
res: Response,
): Promise<void> {
opts?: { ignoreStatusFor?: ReadonlySet<string> },
): Promise<SessionRow> {
const candidates = await selectCandidates({
indexerUrl: deps.config.indexerUrl,
model,
capabilityId: CAPABILITY,
fetchFn: deps.fetchFn,
ignoreStatusFor: opts?.ignoreStatusFor,
});
if (candidates.length === 0) {
throw notFound("model_not_found", `no available chat (llm.chat.v1) supplier for model "${model}"`);
@@ -112,6 +125,16 @@ async function openSession(
}
if (!started || !used) throw toGatewayError(lastErr ?? new Error("no chat supplier could be engaged"));
// Deployment sanity: supplier and gateway must agree on the settle mode.
// Close behavior is driven per-session by the supplier's /end response, so
// a mismatch (rollout window) still settles correctly — just warn.
if ((started.settleMode === "ticket") !== (deps.config.chatSettleMode === "ticket")) {
// eslint-disable-next-line no-console
console.warn(
`[gateway] settle-mode mismatch: supplier ${used.supplierPkh.slice(0, 8)} ran "${started.settleMode}" but gateway is configured "${deps.config.chatSettleMode}"`,
);
}
const sessionId = randomUUID();
deps.store.insertSession({
id: sessionId,
@@ -127,7 +150,9 @@ async function openSession(
});
transcripts.set(sessionId, []);
res.status(200).json({ id: sessionId, object: "chat.session", model, created: nowSec() });
const session = deps.store.getSession(sessionId);
if (!session) throw new Error(`session ${sessionId} vanished after insert`);
return session;
}
// ─── messages (one streamed turn) ────────────────────────────────────────────
@@ -144,8 +169,12 @@ export function makeSessionMessageHandler(deps: GatewayDeps) {
const content = resolveTurnContent(session.id, body);
if (content === "") throw badRequest("content_required", "no user content in this turn");
const ctx = deps.registry.getContext(keyRow);
await ctx.mutex.run(() => streamTurn(deps, session, content, stream, res));
// Turns are chain-free (one HTTP POST to the supplier + SSE relay), so the
// per-key mutex — which exists to serialize wallet UTxO spends — is NOT
// held here. What needs serialization is this session's transcript:
// concurrent turns would interleave supplier appends and corrupt both the
// mirror and the receipt hash. Hence a per-session lock.
await getSessionLock(session.id).run(() => streamTurn(deps, session, content, stream, res));
});
}
@@ -176,6 +205,88 @@ function foldSystem(sessionId: string, content: string, body: Record<string, unk
return content;
}
/** One turn's payload to the supplier: legacy single user string, or a
* multi-message delta (e.g. tool results) with optional tools passthrough. */
export type TurnPayload =
| { content: string }
| { messages: ChatMessage[]; tools?: unknown[]; tool_choice?: unknown };
export interface TurnResult {
assistantMessage: ChatMessage;
finishReason: "stop" | "tool_calls";
usage: { prompt_tokens: number; completion_tokens: number };
}
/**
* Drive one off-chain turn against the session's supplier and mirror the
* result into the session transcript. Caller must hold the session lock.
* Throws SessionGoneError when the supplier no longer knows the session
* (idle-closed or restarted) so callers can transparently reopen.
*/
export async function streamSupplierTurn(
deps: GatewayDeps,
session: SessionRow,
payload: TurnPayload,
onToken?: (token: string) => void,
): Promise<TurnResult> {
const upstream = await deps.fetchFn(`${session.supplier_base_url.replace(/\/+$/, "")}/v1/chat/message`, {
method: "POST",
headers: { "content-type": "application/json", accept: "text/event-stream", "X-Escrow-Ref": session.escrow_ref },
body: JSON.stringify(payload),
});
if (upstream.status === 404) {
await upstream.text().catch(() => "");
throw new SessionGoneError(session.id);
}
if (!upstream.ok || !upstream.body) {
const detail = await upstream.text().catch(() => "");
throw toGatewayError(new SupplierError("supplier_http_error", { status: upstream.status, message: detail.slice(0, 200) }));
}
let assistant = "";
let errored: string | undefined;
let doneMessage: ChatMessage | null = null;
let doneFinish: string | undefined;
let usage = { prompt_tokens: 0, completion_tokens: 0 };
await readSupplierSse(upstream.body as ReadableStream<Uint8Array>, (frame) => {
if (frame.type === "token" && typeof frame.value === "string") {
assistant += frame.value;
onToken?.(frame.value);
} else if (frame.type === "done") {
// Upgraded suppliers echo the final assistant message verbatim; mirror
// that exact object so both transcripts stay hash-identical.
doneMessage = normalizeChatMessage(frame.message);
if (typeof frame.finish_reason === "string") doneFinish = frame.finish_reason;
if (frame.usage && typeof frame.usage === "object") {
const u = frame.usage as Record<string, unknown>;
usage = {
prompt_tokens: typeof u.prompt_tokens === "number" ? u.prompt_tokens : 0,
completion_tokens: typeof u.completion_tokens === "number" ? u.completion_tokens : 0,
};
}
} else if (frame.type === "error") {
errored = typeof frame.message === "string" ? frame.message : "supplier stream error";
}
});
if (errored !== undefined) {
throw toGatewayError(new SupplierError("upstream_error", { message: errored }));
}
const assistantMessage: ChatMessage = doneMessage ?? { role: "assistant", content: assistant };
const finishReason: TurnResult["finishReason"] =
doneFinish === "tool_calls" || (assistantMessage.tool_calls?.length ?? 0) > 0 ? "tool_calls" : "stop";
// Mirror the turn for endChat's response_hash verification / affinity matching.
const mirror = transcripts.get(session.id) ?? [];
if ("content" in payload) mirror.push({ role: "user", content: payload.content });
else mirror.push(...payload.messages);
mirror.push(assistantMessage);
transcripts.set(session.id, mirror);
return { assistantMessage, finishReason, usage };
}
async function streamTurn(
deps: GatewayDeps,
session: SessionRow,
@@ -184,15 +295,6 @@ async function streamTurn(
res: Response,
): Promise<void> {
const id = genId();
const upstream = await deps.fetchFn(`${session.supplier_base_url.replace(/\/+$/, "")}/v1/chat/message`, {
method: "POST",
headers: { "content-type": "application/json", accept: "text/event-stream", "X-Escrow-Ref": session.escrow_ref },
body: JSON.stringify({ content }),
});
if (!upstream.ok || !upstream.body) {
const detail = await upstream.text().catch(() => "");
throw toGatewayError(new SupplierError("supplier_http_error", { status: upstream.status, message: detail.slice(0, 200) }));
}
if (stream) {
res.status(200);
@@ -203,46 +305,43 @@ async function streamTurn(
res.write(sseData(buildChunk({ id, model: session.model, delta: { role: "assistant" }, finishReason: null })));
}
let assistant = "";
let errored: string | undefined;
await readSupplierSse(upstream.body as ReadableStream<Uint8Array>, (frame) => {
if (frame.type === "token" && typeof frame.value === "string") {
assistant += frame.value;
if (stream) res.write(sseData(buildChunk({ id, model: session.model, delta: { content: frame.value }, finishReason: null })));
} else if (frame.type === "error") {
errored = typeof frame.message === "string" ? frame.message : "supplier stream error";
let result: TurnResult;
try {
result = await streamSupplierTurn(deps, session, { content }, (token) => {
if (stream) res.write(sseData(buildChunk({ id, model: session.model, delta: { content: token }, finishReason: null })));
});
} catch (err) {
if (err instanceof SessionGoneError) {
err = toGatewayError(new SupplierError("supplier_http_error", { status: 404, message: err.message }));
}
});
if (errored !== undefined) {
if (stream) {
res.write(sseData(toErrorBody(toGatewayError(new SupplierError("upstream_error", { message: errored })))));
res.write(sseData(toErrorBody(toGatewayError(err))));
res.end();
return;
}
throw toGatewayError(new SupplierError("upstream_error", { message: errored }));
throw toGatewayError(err);
}
// Mirror the transcript for endChat's response_hash verification.
const mirror = transcripts.get(session.id) ?? [];
mirror.push({ role: "user", content });
mirror.push({ role: "assistant", content: assistant });
transcripts.set(session.id, mirror);
if (stream) {
res.write(sseData(buildChunk({ id, model: session.model, delta: {}, finishReason: "stop" })));
res.write(sseData(buildChunk({ id, model: session.model, delta: {}, finishReason: result.finishReason })));
res.write(SSE_DONE);
res.end();
} else {
const usage: Usage = { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 };
res.status(200).json(buildChatCompletion({ id, model: session.model, content: assistant, usage }));
const usage: Usage = {
prompt_tokens: result.usage.prompt_tokens,
completion_tokens: result.usage.completion_tokens,
total_tokens: result.usage.prompt_tokens + result.usage.completion_tokens,
};
res.status(200).json(buildChatCompletion({ id, model: session.model, content: result.assistantMessage.content, usage }));
}
}
interface SupplierFrame {
type?: string;
value?: string;
message?: string;
message?: unknown;
finish_reason?: unknown;
usage?: unknown;
}
/** Read a supplier SSE stream of {type:token|done|error} frames. */
@@ -308,7 +407,34 @@ async function closeSession(
});
deps.store.setSessionState(session.id, "closed", Date.now());
transcripts.delete(session.id);
dropSessionState(session.id);
if (result.settleMode === "ticket") {
// Nothing was Submitted/Accepted: no receipt, no charge (the escrow
// returns via the sweeper's reclaim), and no wallet churn to tidy up.
deps.store.insertUsage({
id: randomUUID(),
key_id: keyRow.id,
created_at: Date.now(),
kind: "chat_session",
model: session.model,
capability_id: CAPABILITY,
supplier_pkh: session.supplier_pkh,
escrow_ref: session.escrow_ref,
cost_lovelace: "0",
prompt_tokens: 0,
completion_tokens: 0,
status: "completed",
failure_reason: null,
});
res.status(200).json({
status: "ended",
escrow_ref: session.escrow_ref,
settle_mode: "ticket",
model: session.model,
});
return;
}
const prompt = result.receipt.prompt_tokens ?? 0;
const completion = result.receipt.completion_tokens ?? 0;

View File

@@ -7,6 +7,7 @@
import { randomBytes } from "crypto";
import type { Receipt } from "@marketplace/shared/receipt";
import type { ToolCall } from "@marketplace/shared/tx";
export function genId(): string {
return `chatcmpl-${randomBytes(16).toString("hex")}`;
@@ -40,6 +41,8 @@ export function buildChatCompletion(args: {
model: string;
content: string;
usage: Usage;
toolCalls?: ToolCall[];
finishReason?: "stop" | "tool_calls";
vector?: VectorReceipt;
}): Record<string, unknown> {
return {
@@ -50,8 +53,12 @@ export function buildChatCompletion(args: {
choices: [
{
index: 0,
message: { role: "assistant", content: args.content },
finish_reason: "stop",
message: {
role: "assistant",
content: args.content,
...(args.toolCalls?.length ? { tool_calls: args.toolCalls } : {}),
},
finish_reason: args.finishReason ?? "stop",
},
],
usage: args.usage,
@@ -63,8 +70,10 @@ export function buildChatCompletion(args: {
export function buildChunk(args: {
id: string;
model: string;
delta: { role?: string; content?: string } | Record<string, never>;
finishReason: "stop" | null;
delta:
| { role?: string; content?: string; tool_calls?: Array<{ index: number } & ToolCall> }
| Record<string, never>;
finishReason: "stop" | "tool_calls" | null;
usage?: Usage;
vector?: VectorReceipt;
}): Record<string, unknown> {

View File

@@ -0,0 +1,35 @@
/**
* gateway/src/openai/transcripts.ts — in-memory per-session state shared by
* the session routes (sessions.ts), the demo affinity executor (demoChat.ts)
* and the sweeper's housekeeping.
*
* transcripts: mirror of each open session's transcript (sessionId → messages).
* Kept hash-identical to the supplier's authoritative transcript (endChat's
* response_hash check). Lost on restart — abandoned sessions are recovered
* by the supplier idle watchdog + gateway sweeper.
* sessionLocks: per-session Mutex serializing turns on ONE session (concurrent
* turns would interleave supplier transcript appends). Distinct from the
* per-key Mutex, which only serializes on-chain wallet work.
*/
import type { ChatMessage } from "@marketplace/shared/tx";
import { Mutex } from "../sdk/registry.js";
export const transcripts = new Map<string, ChatMessage[]>();
const sessionLocks = new Map<string, Mutex>();
export function getSessionLock(sessionId: string): Mutex {
let lock = sessionLocks.get(sessionId);
if (!lock) {
lock = new Mutex();
sessionLocks.set(sessionId, lock);
}
return lock;
}
/** Forget all in-memory state for a session (closed/gone). */
export function dropSessionState(sessionId: string): void {
transcripts.delete(sessionId);
sessionLocks.delete(sessionId);
}

View File

@@ -2,7 +2,7 @@
* gateway/src/openai/validate.ts — request body validation for chat endpoints.
*/
import type { ChatMessage } from "@marketplace/shared/tx";
import { normalizeChatMessage, type ChatMessage } from "@marketplace/shared/tx";
import { badRequest } from "./errors.js";
export interface ParsedChatRequest {
@@ -10,21 +10,30 @@ export interface ParsedChatRequest {
messages: ChatMessage[];
maxTokens?: number;
stream: boolean;
tools?: unknown[];
toolChoice?: unknown;
}
const ROLES = new Set(["system", "user", "assistant"]);
/** Validate an OpenAI chat.completions-style body. Rejects tools/functions;
* silently ignores temperature/n/stop/top_p/etc (documented). */
export function parseChatRequest(body: unknown): ParsedChatRequest {
/** Validate an OpenAI chat.completions-style body. Default path rejects
* tools/functions; `allowTools` (demo keys) accepts tools/tool_choice, the
* `tool` role, null content and assistant tool_calls, normalizing every
* message into canonical mirror shape. Silently ignores temperature/n/stop/
* top_p/etc (documented). */
export function parseChatRequest(body: unknown, opts?: { allowTools?: boolean }): ParsedChatRequest {
const allowTools = opts?.allowTools === true;
if (typeof body !== "object" || body === null) {
throw badRequest("invalid_body", "request body must be a JSON object");
}
const b = body as Record<string, unknown>;
if ("tools" in b || "tool_choice" in b || "functions" in b || "function_call" in b) {
if (!allowTools && ("tools" in b || "tool_choice" in b || "functions" in b || "function_call" in b)) {
throw badRequest("unsupported_parameter", "tools and function-calling are not supported by this gateway");
}
if (allowTools && ("functions" in b || "function_call" in b)) {
throw badRequest("unsupported_parameter", "legacy functions/function_call are not supported; use tools");
}
const model = b.model;
if (typeof model !== "string" || model === "") {
@@ -37,6 +46,14 @@ export function parseChatRequest(body: unknown): ParsedChatRequest {
}
const messages: ChatMessage[] = [];
for (const m of rawMessages) {
if (allowTools) {
const msg = normalizeChatMessage(m);
if (!msg) {
throw badRequest("invalid_messages", "each message must be OpenAI-shaped {role, content, tool_calls?, tool_call_id?}");
}
messages.push(msg);
continue;
}
if (
typeof m !== "object" ||
m === null ||
@@ -50,6 +67,13 @@ export function parseChatRequest(body: unknown): ParsedChatRequest {
messages.push({ role: mm.role, content: mm.content });
}
let tools: unknown[] | undefined;
let toolChoice: unknown;
if (allowTools && Array.isArray(b.tools) && b.tools.length > 0) {
tools = b.tools;
toolChoice = b.tool_choice;
}
let maxTokens: number | undefined;
const mt = b.max_tokens ?? b.max_completion_tokens;
if (mt !== undefined && mt !== null) {
@@ -60,5 +84,5 @@ export function parseChatRequest(body: unknown): ParsedChatRequest {
}
const stream = b.stream === true;
return { model, messages, maxTokens, stream };
return { model, messages, maxTokens, stream, tools, toolChoice };
}

View File

@@ -56,6 +56,11 @@ export interface SelectSupplierOpts {
model: string;
capabilityId: string;
fetchFn?: typeof globalThis.fetch;
/** supplier_pkhs eligible regardless of polled status — used after the demo
* executor frees a supplier the indexer poller (~20 s cadence) still reports
* busy. sdk.startChat's live /status pre-check still guards genuinely-busy
* suppliers. */
ignoreStatusFor?: ReadonlySet<string>;
}
/**
@@ -79,7 +84,9 @@ export async function selectCandidates(opts: SelectSupplierOpts): Promise<Suppli
if (raw.capability_id !== opts.capabilityId) continue;
if (raw.model !== opts.model) continue;
if (raw.advert_status !== "Active") continue;
if (raw.status !== "free" && raw.status !== "unknown") continue;
if (raw.status !== "free" && raw.status !== "unknown" && !opts.ignoreStatusFor?.has(raw.supplier_pkh)) {
continue;
}
const advertRef = parseRef(raw.utxo_ref);
if (!advertRef) continue;
candidates.push({
@@ -101,19 +108,27 @@ export async function selectCandidates(opts: SelectSupplierOpts): Promise<Suppli
return candidates;
}
/** Distinct models across Active suppliers (for GET /openai/v1/models). */
/** Distinct models across Active suppliers (for GET /openai/v1/models).
* Pass capabilityId to list only models under one capability (demo keys see
* only llm.chat.v1 models — the session-backed executor's routing target). */
export async function listModels(opts: {
indexerUrl: string;
capabilityId?: string;
fetchFn?: typeof globalThis.fetch;
}): Promise<string[]> {
const fetchFn = opts.fetchFn ?? globalThis.fetch;
const res = await fetchFn(`${opts.indexerUrl}/suppliers`);
const url = opts.capabilityId
? `${opts.indexerUrl}/suppliers?capability_id=${encodeURIComponent(opts.capabilityId)}`
: `${opts.indexerUrl}/suppliers`;
const res = await fetchFn(url);
if (!res.ok) throw new Error(`indexer /suppliers returned ${res.status}`);
const body = (await res.json()) as unknown;
if (!Array.isArray(body)) throw new Error("indexer /suppliers did not return an array");
const models = new Set<string>();
for (const raw of body as SupplierView[]) {
if (raw.advert_status === "Active" && typeof raw.model === "string") models.add(raw.model);
if (raw.advert_status !== "Active" || typeof raw.model !== "string") continue;
if (opts.capabilityId !== undefined && raw.capability_id !== opts.capabilityId) continue;
models.add(raw.model);
}
return [...models].sort();
}

View File

@@ -7,10 +7,10 @@
* POST /openai/v1/chat/sessions[/:id/messages|/close].
*/
import express, { type Express, type Request, type Response } from "express";
import express, { type Express, type NextFunction, type Request, type Response } from "express";
import type { GatewayDeps } from "./deps.js";
import { makeApiKeyAuth } from "./middleware/apiKeyAuth.js";
import { ipRateLimit, keyRateLimit } from "./middleware/rateLimit.js";
import { ipRateLimit, keyRateLimit, demoIpRateLimit } from "./middleware/rateLimit.js";
import { sendError } from "./middleware/http.js";
import { notFound } from "./openai/errors.js";
import { makeChatCompletionsHandler } from "./openai/chatCompletions.js";
@@ -23,10 +23,39 @@ import {
import { makeSignupHandler, makeAccountHandler, makeWithdrawHandler } from "./account/routes.js";
import { INDEX_HTML } from "./ui/page.js";
/**
* CORS for browser clients on an allowlisted Origin (e.g. the marketplace
* frontend's "Generate API key" page, served from a different host than the
* gateway). Only echoes Access-Control-* headers when the request's Origin is
* in the configured allowlist; an empty allowlist disables CORS entirely.
* No credentials are allowed — the signup call is a plain bearer-less POST, so
* cookies never ride along cross-origin. Preflight OPTIONS is answered 204.
*/
function corsMiddleware(allowOrigins: string[]) {
const allow = new Set(allowOrigins);
return (req: Request, res: Response, next: NextFunction): void => {
const origin = req.headers.origin;
if (typeof origin === "string" && allow.has(origin)) {
res.setHeader("Access-Control-Allow-Origin", origin);
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-Max-Age", "600");
}
if (req.method === "OPTIONS") {
res.status(204).end();
return;
}
next();
};
}
export function createApp(deps: GatewayDeps): Express {
const app = express();
// Behind traefik/Cloudflare — trust the proxy chain so req.ip is the client.
app.set("trust proxy", true);
// CORS before body-parsing so preflight OPTIONS short-circuits cheaply.
app.use(corsMiddleware(deps.config.corsOrigins));
app.use(express.json({ limit: "1mb" }));
app.get("/healthz", (_req: Request, res: Response) => {
@@ -40,10 +69,12 @@ export function createApp(deps: GatewayDeps): Express {
// Public signup, rate-limited per IP.
app.post("/signup", ipRateLimit(deps.config.signupRate), makeSignupHandler(deps));
// Bearer-gated routes: auth → per-key rate limit → handler.
// Bearer-gated routes: auth → per-key rate limit (skips demo keys) →
// per-IP demo limit (demo keys only) → handler.
const auth = makeApiKeyAuth(deps.store);
const krl = keyRateLimit(deps.config.keyRate);
const gated = [auth, krl] as const;
const dirl = demoIpRateLimit(deps.config.demoIpRate);
const gated = [auth, krl, dirl] as const;
app.get("/account", auth, makeAccountHandler(deps));
app.post("/account/withdraw", auth, makeWithdrawHandler(deps));

View File

@@ -10,26 +10,120 @@
* state and never blind-reclaim a Submitted escrow. The on-chain validator is
* the source of truth for deadlines, so premature/late attempts simply fail and
* are caught here; the age thresholds just avoid racing in-flight requests.
*
* Chat sessions get special handling:
* - Their deliver_by is ~30 min (vs ~10 min one-shot), so rows belonging to
* an OPEN session use CHAT_RECLAIM_AFTER_MS — otherwise every open session
* older than 11 min would trigger a doomed Reclaim attempt per sweep while
* holding the key's mutex.
* - When a session's escrow reaches a terminal state (Accepted here, spent
* elsewhere, or Reclaimed), the session row is closed, its in-memory state
* dropped, and a usage row recorded — abandoned demo sessions (nobody calls
* /close) settle and get billed through this path.
*
* Escrow↔session lineage: the indexer keeps one row per utxo_ref per state
* (terminal spends update nothing), and the validator preserves posted_at
* across states — so a session's lineage is "rows sharing the posted_at of the
* session's original Open ref row".
*
* Because terminal spends never update indexer rows, every settled escrow
* would be retried (one doomed chain query each) on every sweep forever. The
* in-process `settledRefs` memory skips refs this process has already settled
* or found spent; a restart just causes one harmless retry burst.
*
* Ticket mode (config.chatSettleMode === "ticket"): session escrows are never
* Claimed — they sit Open until reclaimed here after deliver_by. Reclaiming
* an Open ticket does NOT close a still-open session row: nothing re-checks
* the escrow after /v1/chat/start, so the conversation (and its affinity
* mirror) lives on, bounded only by the idle janitor.
*/
import { randomUUID } from "crypto";
import type { GatewayDeps } from "./deps.js";
import { fetchEscrows, acceptAndConfirm, reclaimAndConfirm } from "./onchain/settle.js";
import type { SessionRow } from "./db/store.js";
import { fetchEscrows, acceptAndConfirm, reclaimAndConfirm, type EscrowRow } from "./onchain/settle.js";
import { parseRef } from "./routing/selectSupplier.js";
import { dropSessionState } from "./openai/transcripts.js";
import { dropDemoSession, sweepIdleDemoSessions } from "./openai/demoChat.js";
import { CAPABILITY as CHAT_CAPABILITY } from "./openai/sessions.js";
// Don't touch escrows newer than this — an in-flight request is settling them.
const MIN_AGE_MS = 90_000;
// Reclaim Open/Claimed escrows older than this (past deliver_by ~10m + buffer).
const RECLAIM_AFTER_MS = 11 * 60 * 1000;
// Chat-session escrows: deliver_by ≈ posted_at + 30 min, so wait longer.
const CHAT_RECLAIM_AFTER_MS = 35 * 60 * 1000;
// Reclaim only fires once the validator allows it (validity lower bound >=
// deliver_by); attempting earlier is a guaranteed-doomed tx build.
const DELIVER_BY_BUFFER_MS = 60_000;
/** Refs this process already settled (Accepted/Reclaimed) or found spent.
* The indexer never updates terminal rows, so without this every historical
* escrow is retried each sweep. Bounded FIFO; restart = one retry burst. */
const settledRefs = new Set<string>();
const SETTLED_REFS_MAX = 5_000;
function markSettled(ref: string): void {
settledRefs.add(ref);
if (settledRefs.size > SETTLED_REFS_MAX) {
const first = settledRefs.values().next().value;
if (first !== undefined) settledRefs.delete(first);
}
}
function log(msg: string): void {
// eslint-disable-next-line no-console
console.log(`[gateway:sweeper] ${msg}`);
}
export async function runSweepOnce(deps: GatewayDeps): Promise<void> {
function closeSessionRow(
deps: GatewayDeps,
session: SessionRow,
outcome: { status: "completed" | "reclaimed"; costLovelace: string | null },
): void {
// Re-read: an in-flight close (route handler or demo executor) may have won.
const current = deps.store.getSession(session.id);
if (!current || current.state !== "open") return;
deps.store.setSessionState(session.id, "closed", Date.now());
dropSessionState(session.id);
dropDemoSession(session.id);
deps.store.insertUsage({
id: randomUUID(),
key_id: session.key_id,
created_at: Date.now(),
kind: "chat_session",
model: session.model,
capability_id: CHAT_CAPABILITY,
supplier_pkh: session.supplier_pkh,
escrow_ref: session.escrow_ref,
cost_lovelace: outcome.costLovelace,
prompt_tokens: 0,
completion_tokens: 0,
status: outcome.status,
failure_reason: null,
});
log(`closed abandoned session ${session.id} (${outcome.status})`);
}
/** Chain-settle functions, injectable for tests. */
export interface SweepSettleFns {
accept: typeof acceptAndConfirm;
reclaim: typeof reclaimAndConfirm;
}
export async function runSweepOnce(
deps: GatewayDeps,
settle: SweepSettleFns = { accept: acceptAndConfirm, reclaim: reclaimAndConfirm },
): Promise<void> {
// Idle demo sessions first: closing them makes the supplier Submit, so this
// sweep (or the next) can Accept the escrow right after.
await sweepIdleDemoSessions(deps).catch((err) =>
log(`demo idle janitor: ${err instanceof Error ? err.message : String(err)}`),
);
const now = Date.now();
for (const keyRow of deps.store.listKeys()) {
let rows;
let rows: EscrowRow[];
try {
rows = await fetchEscrows(deps.config.indexerUrl, keyRow.wallet_pkh, deps.fetchFn);
} catch (err) {
@@ -37,20 +131,71 @@ export async function runSweepOnce(deps: GatewayDeps): Promise<void> {
continue;
}
// Lineage-match open sessions: the session stores its ORIGINAL Open ref;
// rows sharing that row's posted_at are the same escrow in later states.
const sessionByPostedAt = new Map<number, SessionRow>();
for (const session of deps.store.listOpenSessionsByKey(keyRow.id)) {
const origin = rows.find((r) => r.utxo_ref === session.escrow_ref);
if (origin) sessionByPostedAt.set(origin.posted_at, session);
}
const ctx = deps.registry.getContext(keyRow);
for (const row of rows) {
const ref = parseRef(row.utxo_ref);
if (!ref) continue;
if (settledRefs.has(row.utxo_ref)) continue;
const ageMs = now - (row.posted_at ?? 0);
if (ageMs < MIN_AGE_MS) continue;
const chatSession = sessionByPostedAt.get(row.posted_at);
try {
if (row.state === "Submitted") {
await ctx.mutex.run(() => acceptAndConfirm(deps.chain, ctx.walletKey, ref));
log(`accepted stranded Submitted escrow ${row.utxo_ref}`);
} else if ((row.state === "Open" || row.state === "Claimed") && ageMs > RECLAIM_AFTER_MS) {
await ctx.mutex.run(() => reclaimAndConfirm(deps.chain, ctx.walletKey, ref));
log(`reclaimed stranded ${row.state} escrow ${row.utxo_ref}`);
try {
await ctx.mutex.run(() => settle.accept(deps.chain, ctx.walletKey, ref));
log(`accepted stranded Submitted escrow ${row.utxo_ref}`);
markSettled(row.utxo_ref);
if (chatSession) {
closeSessionRow(deps, chatSession, { status: "completed", costLovelace: chatSession.price_lovelace });
}
} catch (err) {
// Accept fails routinely when the UTxO is already spent (settled
// by an in-flight request, an earlier sweep, or supplier Release).
// For session lineages, confirm the spend on-chain before closing
// the row — an Accept that failed for another reason (timeout)
// must leave the session open for the next sweep.
if ((await deps.chain.queryUtxo(ref)) === null) {
markSettled(row.utxo_ref);
if (chatSession) {
closeSessionRow(deps, chatSession, { status: "completed", costLovelace: chatSession.price_lovelace });
}
} else if (!chatSession) {
// Preserve the historical quiet-skip for one-shot escrows.
void err;
}
}
} else if (row.state === "Open" || row.state === "Claimed") {
const reclaimAfterMs = chatSession ? CHAT_RECLAIM_AFTER_MS : RECLAIM_AFTER_MS;
// The validator only allows Reclaim at/after deliver_by — skip
// attempts that cannot succeed yet (early-closed ticket sessions).
const pastDeliverBy =
typeof row.deliver_by === "number" ? now > row.deliver_by + DELIVER_BY_BUFFER_MS : true;
if (ageMs > reclaimAfterMs && pastDeliverBy) {
await ctx.mutex.run(() => settle.reclaim(deps.chain, ctx.walletKey, ref));
log(`reclaimed stranded ${row.state} escrow ${row.utxo_ref}`);
markSettled(row.utxo_ref);
if (chatSession) {
const ticketOpen = deps.config.chatSettleMode === "ticket" && row.state === "Open";
if (ticketOpen) {
// Ticket escrow reclaimed out from under a LIVE session —
// by design. Keep the session row + mirror; only the idle
// janitor / close paths end ticket sessions.
log(`ticket mode: escrow reclaimed, session ${chatSession.id} stays open`);
} else {
// Funds returned — the session never settled; record no cost.
closeSessionRow(deps, chatSession, { status: "reclaimed", costLovelace: null });
}
}
}
}
} catch {
// Expected when the validator rejects (wrong window/state) or the UTxO
@@ -58,9 +203,6 @@ export async function runSweepOnce(deps: GatewayDeps): Promise<void> {
}
}
}
// Mark closed/settled chat sessions whose escrow is no longer active.
// (Best-effort housekeeping; the escrow recovery above is the load-bearing part.)
}
/** Start the periodic sweeper. Returns a stop() function. */

View File

@@ -0,0 +1,143 @@
/**
* affinity.test.ts — transcript-prefix session matching rules.
*/
import { describe, it, expect } from "vitest";
import {
matchSessionPrefix,
normalizeAssistantContent,
trimToRespondable,
type AffinityCandidate,
} from "../src/openai/affinity.js";
import type { ChatMessage } from "@marketplace/shared/tx";
const u = (content: string): ChatMessage => ({ role: "user", content });
const a = (content: string): ChatMessage => ({ role: "assistant", content });
const TOOL_CALL = { id: "call_1", type: "function" as const, function: { name: "get_time", arguments: "{}" } };
function cand(sessionId: string, mirror: ChatMessage[], lastUsedAt = 0, model = "m"): AffinityCandidate {
return { sessionId, model, mirror, lastUsedAt };
}
describe("matchSessionPrefix", () => {
it("matches an exact prefix and returns the delta", () => {
const got = matchSessionPrefix([cand("s1", [u("hi"), a("hello")])], [u("hi"), a("hello"), u("more")], "m");
expect(got).toEqual({ sessionId: "s1", delta: [u("more")] });
});
it("accepts a multi-message tool delta ending with a tool result", () => {
const mirror = [u("hi"), { role: "assistant" as const, content: "", tool_calls: [TOOL_CALL] }];
const incoming = [...mirror, { role: "tool" as const, content: "3pm", tool_call_id: "call_1" }];
const got = matchSessionPrefix([cand("s1", mirror)], incoming, "m");
expect(got?.sessionId).toBe("s1");
expect(got?.delta).toHaveLength(1);
expect(got?.delta[0].role).toBe("tool");
});
it("rejects deltas that end with a non-respondable role", () => {
const got = matchSessionPrefix([cand("s1", [u("hi")])], [u("hi"), a("stray assistant")], "m");
expect(got).toBeNull();
});
it("never matches when incoming is not longer than the mirror (retry of answered request)", () => {
const mirror = [u("hi"), a("hello")];
expect(matchSessionPrefix([cand("s1", mirror)], mirror, "m")).toBeNull();
expect(matchSessionPrefix([cand("s1", mirror)], [u("hi")], "m")).toBeNull();
});
it("rejects diverged histories (prefix mismatch)", () => {
const got = matchSessionPrefix([cand("s1", [u("hi"), a("hello")])], [u("hi"), a("DIFFERENT"), u("more")], "m");
expect(got).toBeNull();
});
it("requires the model to match", () => {
expect(matchSessionPrefix([cand("s1", [u("hi")], 0, "other")], [u("hi"), u("more")], "m")).toBeNull();
});
it("prefers the longest mirror, then most recently used", () => {
const incoming = [u("hi"), a("hello"), u("more")];
const longest = matchSessionPrefix(
[cand("short", [u("hi")], 99), cand("long", [u("hi"), a("hello")], 0)],
incoming,
"m",
);
expect(longest?.sessionId).toBe("long");
const tie = matchSessionPrefix(
[cand("old", [u("hi"), a("hello")], 1), cand("recent", [u("hi"), a("hello")], 2)],
incoming,
"m",
);
expect(tie?.sessionId).toBe("recent");
});
it("matches an empty mirror (fresh session) against any respondable history", () => {
const incoming = [{ role: "system" as const, content: "be nice" }, u("hi")];
const got = matchSessionPrefix([cand("fresh", [])], incoming, "m");
expect(got).toEqual({ sessionId: "fresh", delta: incoming });
});
it("treats content null-normalization as equivalent (tool-call assistant echo)", () => {
const mirror = [u("hi"), { role: "assistant" as const, content: "", tool_calls: [TOOL_CALL] }];
// Client echoes the assistant message back with content:"" (normalized from null upstream).
const incoming = [u("hi"), { role: "assistant" as const, content: "", tool_calls: [{ ...TOOL_CALL }] }, { role: "tool" as const, content: "x", tool_call_id: "call_1" }];
expect(matchSessionPrefix([cand("s1", mirror)], incoming, "m")?.sessionId).toBe("s1");
});
it("matches leniently when the client reshapes the assistant echo (think-strip + whitespace)", () => {
const mirror = [u("hi"), a("<think>plan it</think>Hello there\n")];
const got = matchSessionPrefix([cand("s1", mirror)], [u("hi"), a("Hello there"), u("more")], "m");
expect(got).toEqual({ sessionId: "s1", delta: [u("more")] });
});
it("stays strict on user/tool divergence even with lenient assistant matching", () => {
const mirror = [u("hi "), a("Hello")];
// User content differs by whitespace — client-authored roles must be byte-exact.
expect(matchSessionPrefix([cand("s1", mirror)], [u("hi"), a("Hello"), u("more")], "m")).toBeNull();
});
it("rejects assistant tool_calls divergence under lenient matching", () => {
const mirror = [u("hi"), { role: "assistant" as const, content: "x", tool_calls: [TOOL_CALL] }];
const changed = { ...TOOL_CALL, function: { name: "get_time", arguments: '{"tz":"utc"}' } };
const incoming = [u("hi"), { role: "assistant" as const, content: "x ", tool_calls: [changed] }, { role: "tool" as const, content: "t", tool_call_id: "call_1" }];
expect(matchSessionPrefix([cand("s1", mirror)], incoming, "m")).toBeNull();
});
it("prefers a strict match over a lenient one regardless of recency", () => {
const incoming = [u("hi"), a("Hello"), u("more")];
const got = matchSessionPrefix(
[
cand("lenient", [u("hi"), a("Hello ")], 99),
cand("strict", [u("hi"), a("Hello")], 1),
],
incoming,
"m",
);
expect(got?.sessionId).toBe("strict");
});
});
describe("normalizeAssistantContent", () => {
it("strips think blocks, collapses whitespace and trims", () => {
expect(normalizeAssistantContent("<think>a\nb</think> Hi\n\nthere friend ")).toBe("Hi there friend");
expect(normalizeAssistantContent("<THINK>x</THINK>ok")).toBe("ok");
expect(normalizeAssistantContent("plain")).toBe("plain");
});
});
describe("trimToRespondable", () => {
it("drops trailing assistant/system messages", () => {
expect(trimToRespondable([u("hi"), a("partial")])).toEqual([u("hi")]);
expect(trimToRespondable([u("hi"), a("x"), { role: "system" as const, content: "nudge" }])).toEqual([u("hi")]);
});
it("returns empty when nothing respondable remains", () => {
expect(trimToRespondable([a("hello")])).toEqual([]);
expect(trimToRespondable([])).toEqual([]);
});
it("keeps already-respondable histories unchanged", () => {
const msgs = [u("hi"), a("x"), u("more")];
expect(trimToRespondable(msgs)).toEqual(msgs);
});
});

View File

@@ -1,23 +1,37 @@
import { describe, it, expect } from "vitest";
import { describe, it, expect, vi } from "vitest";
import { tmpdir } from "os";
import { join } from "path";
import { randomUUID } from "crypto";
import { randomBytes, randomUUID } from "crypto";
import request from "supertest";
import { createApp } from "../src/server.js";
import { sweepIdleDemoSessions } from "../src/openai/demoChat.js";
import { GatewayStore } from "../src/db/store.js";
import { SdkRegistry } from "../src/sdk/registry.js";
import { genPrivKeyHex, deriveWalletKey } from "../src/wallet.js";
import { seal } from "../src/crypto/seal.js";
import { hashApiKey } from "../src/middleware/apiKeyAuth.js";
import type { GatewayConfig } from "../src/config.js";
import type { GatewayDeps } from "../src/deps.js";
const MASTER = "ab".repeat(32);
const SUPPLIERS = [
{ utxo_ref: "aa".repeat(32) + "#0", supplier_pkh: "s1", capability_id: "llm.text.generate.v1", model: "qwen", max_output_tokens: 256, price_lovelace: "1000000", supplier_bond_lovelace: "1000000", buyer_bond_lovelace: "1000000", endpoint_url: "http://sup", advert_status: "Active", status: "free" },
{ utxo_ref: "bb".repeat(32) + "#0", supplier_pkh: "s2", capability_id: "llm.chat.v1", model: "kimi", max_output_tokens: 256, price_lovelace: "1000000", supplier_bond_lovelace: "1000000", buyer_bond_lovelace: "1000000", endpoint_url: "http://sup", advert_status: "Active", status: "free" },
];
function makeDeps(fetchFn?: typeof globalThis.fetch): GatewayDeps {
function makeDeps(
fetchFn?: typeof globalThis.fetch,
overrides?: {
demoIpRate?: { max: number; windowMs: number };
chain?: GatewayDeps["chain"];
chatSettleMode?: "full" | "ticket";
},
): GatewayDeps {
const dbDir = join(tmpdir(), `gw-test-${randomUUID()}`);
const store = new GatewayStore(dbDir);
const config: GatewayConfig = {
masterKeyHex: "ab".repeat(32),
masterKeyHex: MASTER,
indexerUrl: "http://ix",
ogmiosUrl: "http://og",
networkId: 0,
@@ -26,11 +40,15 @@ function makeDeps(fetchFn?: typeof globalThis.fetch): GatewayDeps {
dbDir,
signupRate: { max: 1000, windowMs: 60_000 },
keyRate: { max: 1000, windowMs: 60_000 },
demoIpRate: overrides?.demoIpRate ?? { max: 1000, windowMs: 60_000 },
sweeperIntervalMs: 60_000,
demoSessionIdleMs: 180_000,
chatSettleMode: overrides?.chatSettleMode ?? "full",
walletHealthIntervalMs: 600_000,
sdkRegistryMax: 100,
corsOrigins: [],
};
const chain = { queryUtxosByAddress: async () => [] } as unknown as GatewayDeps["chain"];
const chain = overrides?.chain ?? ({ queryUtxosByAddress: async () => [] } as unknown as GatewayDeps["chain"]);
const registry = new SdkRegistry({
chain,
indexerUrl: config.indexerUrl,
@@ -103,3 +121,450 @@ describe("gateway HTTP", () => {
expect(res.body.error.code).toBe("unsupported_parameter");
});
});
// ─── Shared demo key (session-backed completions) ───────────────────────────
const TOOLS = [{ type: "function", function: { name: "get_time", parameters: {} } }];
const TOOL_CALL = { id: "call_1", type: "function", function: { name: "get_time", arguments: "{}" } };
/** A funded wallet: covers price+bonds+collateral+fee and has pure-ADA ≥ 5. */
const fundedChain = {
queryUtxosByAddress: async () => [
{ ref: { txHash: "ff".repeat(32), index: 0 }, address: "a", lovelace: 20_000_000n, assets: {}, datumHex: null, scriptRef: null },
],
} as unknown as GatewayDeps["chain"];
function supplierSse(frames: unknown[]): Response {
const body = frames.map((f) => `data: ${JSON.stringify(f)}\n\n`).join("");
return new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } });
}
function doneFrame(message: Record<string, unknown>, finishReason: string, usage = { prompt_tokens: 3, completion_tokens: 2 }) {
return { type: "done", message, finish_reason: finishReason, usage };
}
/** fetchFn serving the indexer supplier list and scripted /v1/chat/message
* responses. Pass `suppliers` to serve a mutable list (eviction tests flip
* status mid-test); /v1/chat/end calls are recorded and 200-acked. */
function demoFetch(messageResponses: Array<() => Response>, suppliers?: () => unknown[]) {
const messageCalls: Array<Record<string, unknown>> = [];
const endCalls: string[] = [];
let i = 0;
const fetchFn = (async (url: unknown, init?: { body?: string; headers?: Record<string, string> }) => {
const u = String(url);
if (u.includes("/v1/chat/message")) {
messageCalls.push(init?.body ? (JSON.parse(init.body) as Record<string, unknown>) : {});
const make = messageResponses[Math.min(i, messageResponses.length - 1)];
i += 1;
return make();
}
if (u.includes("/v1/chat/end")) {
endCalls.push(init?.headers?.["X-Escrow-Ref"] ?? "");
return new Response(JSON.stringify({ status: "submitted" }), { status: 200 });
}
if (u.includes("/suppliers")) return new Response(JSON.stringify(suppliers ? suppliers() : SUPPLIERS), { status: 200 });
return new Response("[]", { status: 200 });
}) as unknown as typeof globalThis.fetch;
return { fetchFn, messageCalls, endCalls };
}
/** Insert a demo-flagged key directly (public signup can't mint one) and stub
* the SDK's chain-touching startChat on the pre-warmed per-key context. */
function setupDemoKey(deps: GatewayDeps) {
const rawKey = `vk_test_${randomBytes(24).toString("hex")}`;
const privHex = genPrivKeyHex();
const walletKey = deriveWalletKey(privHex, 0);
const sealed = seal(privHex, MASTER);
deps.store.insertKey({
id: randomUUID(),
key_hash: hashApiKey(rawKey),
key_prefix: rawKey.slice(0, 12),
label: "shared-demo",
wallet_pkh: walletKey.pubKeyHash,
deposit_address: walletKey.address,
enc_priv_nonce: sealed.nonce,
enc_priv_ct: sealed.ct,
enc_priv_tag: sealed.tag,
master_key_version: 1,
created_at: Date.now(),
demo: 1,
});
const keyRow = deps.store.getKeyByHash(hashApiKey(rawKey))!;
const ctx = deps.registry.getContext(keyRow);
const startChat = vi.fn(async () => ({
escrowRef: { txHash: randomBytes(32).toString("hex"), index: 0 },
sessionNonce: "nonce",
supplierBaseUrl: "http://sup",
}));
(ctx.sdk as { startChat: unknown }).startChat = startChat;
return { rawKey, keyRow, startChat };
}
describe("gateway demo key", () => {
it("runs a session-backed completion with tools and reuses the session via affinity", async () => {
const { fetchFn, messageCalls } = demoFetch([
() => supplierSse([doneFrame({ role: "assistant", content: "", tool_calls: [TOOL_CALL] }, "tool_calls")]),
() => supplierSse([{ type: "token", value: "It is 3pm" }, doneFrame({ role: "assistant", content: "It is 3pm" }, "stop")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const app = createApp(deps);
// Turn 1: model asks for a tool call.
const history: Array<Record<string, unknown>> = [{ role: "user", content: "what time is it?" }];
const res1 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: history, tools: TOOLS });
expect(res1.status).toBe(200);
expect(res1.body.choices[0].message.tool_calls).toEqual([TOOL_CALL]);
expect(res1.body.choices[0].finish_reason).toBe("tool_calls");
expect(startChat).toHaveBeenCalledTimes(1);
// First turn replays the full history and forwards tools.
expect((messageCalls[0].messages as unknown[]).length).toBe(1);
expect(messageCalls[0].tools).toEqual(TOOLS);
// Turn 2: client echoes the assistant tool_calls + tool result.
history.push({ role: "assistant", content: "", tool_calls: [TOOL_CALL] });
history.push({ role: "tool", content: "3pm", tool_call_id: "call_1" });
const res2 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: history, tools: TOOLS });
expect(res2.status).toBe(200);
expect(res2.body.choices[0].message.content).toBe("It is 3pm");
expect(res2.body.choices[0].finish_reason).toBe("stop");
// Affinity hit: no new session, only the unseen delta is forwarded.
expect(startChat).toHaveBeenCalledTimes(1);
const delta = messageCalls[1].messages as Array<{ role: string }>;
expect(delta).toHaveLength(1);
expect(delta[0].role).toBe("tool");
});
it("streams tool_calls chunks with finish_reason tool_calls", async () => {
const { fetchFn } = demoFetch([
() => supplierSse([doneFrame({ role: "assistant", content: "", tool_calls: [TOOL_CALL] }, "tool_calls")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey } = setupDemoKey(deps);
const res = await request(createApp(deps))
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "time?" }], tools: TOOLS, stream: true });
expect(res.status).toBe(200);
const chunks = res.text
.split("\n\n")
.filter((l) => l.startsWith("data: ") && !l.includes("[DONE]"))
.map((l) => JSON.parse(l.slice(6)));
const toolChunk = chunks.find((c) => c.choices?.[0]?.delta?.tool_calls);
expect(toolChunk.choices[0].delta.tool_calls[0].function.name).toBe("get_time");
expect(chunks.at(-1).choices[0].finish_reason).toBe("tool_calls");
expect(res.text).toContain("data: [DONE]");
});
it("transparently reopens when the supplier lost the session (idle-close)", async () => {
const { fetchFn, messageCalls } = demoFetch([
() => supplierSse([{ type: "token", value: "Hi" }, doneFrame({ role: "assistant", content: "Hi" }, "stop")]),
() => new Response("not found", { status: 404 }),
() => supplierSse([{ type: "token", value: "Sure" }, doneFrame({ role: "assistant", content: "Sure" }, "stop")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const app = createApp(deps);
const res1 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res1.status).toBe(200);
// Extended history matches the session, but the supplier 404s the turn →
// the executor closes the session and reopens, replaying full history.
const history = [
{ role: "user", content: "hi" },
{ role: "assistant", content: "Hi" },
{ role: "user", content: "more please" },
];
const res2 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: history });
expect(res2.status).toBe(200);
expect(res2.body.choices[0].message.content).toBe("Sure");
expect(startChat).toHaveBeenCalledTimes(2);
expect((messageCalls[2].messages as unknown[]).length).toBe(3);
// The dead session's fee was recorded for spend audit.
const usage = deps.store.listUsage(deps.store.listAllKeys().find((k) => k.demo === 1)!.id, 10);
expect(usage.some((u) => u.kind === "chat_session" && u.cost_lovelace === "1000000")).toBe(true);
});
it("rejects histories with nothing respondable before any chain work", async () => {
const deps = makeDeps(undefined, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const res = await request(createApp(deps))
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "assistant", content: "hello" }] });
expect(res.status).toBe(400);
expect(res.body.error.code).toBe("invalid_messages");
expect(startChat).not.toHaveBeenCalled();
});
it("answers a trailing-assistant history by trimming to the last user message (regenerate)", async () => {
const { fetchFn, messageCalls } = demoFetch([
() => supplierSse([{ type: "token", value: "Again" }, doneFrame({ role: "assistant", content: "Again" }, "stop")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const res = await request(createApp(deps))
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({
model: "kimi",
messages: [
{ role: "user", content: "hi" },
{ role: "assistant", content: "partial answer" },
],
});
expect(res.status).toBe(200);
expect(res.body.choices[0].message.content).toBe("Again");
expect(startChat).toHaveBeenCalledTimes(1);
// The trailing assistant message is dropped from the replay.
expect((messageCalls[0].messages as Array<{ role: string }>).map((m) => m.role)).toEqual(["user"]);
});
it("keeps the session when the client reshapes the assistant echo (lenient affinity)", async () => {
const { fetchFn, messageCalls } = demoFetch([
() => supplierSse([{ type: "token", value: "<think>hm</think>Hi there" }, doneFrame({ role: "assistant", content: "<think>hm</think>Hi there" }, "stop")]),
() => supplierSse([{ type: "token", value: "Sure" }, doneFrame({ role: "assistant", content: "Sure" }, "stop")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const app = createApp(deps);
const res1 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res1.status).toBe(200);
// Client echoes the assistant reply with the think block stripped and
// whitespace collapsed — still the same session, delta only.
const res2 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({
model: "kimi",
messages: [
{ role: "user", content: "hi" },
{ role: "assistant", content: "Hi there" },
{ role: "user", content: "more" },
],
});
expect(res2.status).toBe(200);
expect(res2.body.choices[0].message.content).toBe("Sure");
expect(startChat).toHaveBeenCalledTimes(1);
expect((messageCalls[1].messages as unknown[]).length).toBe(1);
});
it("evicts the LRU demo session when every supplier is pinned", async () => {
const suppliers = [{ ...SUPPLIERS[1] }]; // single chat supplier, mutable status
const { fetchFn, messageCalls, endCalls } = demoFetch(
[
() => supplierSse([{ type: "token", value: "Hi" }, doneFrame({ role: "assistant", content: "Hi" }, "stop")]),
() => supplierSse([{ type: "token", value: "New" }, doneFrame({ role: "assistant", content: "New" }, "stop")]),
],
() => suppliers,
);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const app = createApp(deps);
const res1 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res1.status).toBe(200);
// The lone supplier is now pinned by session 1 (indexer sees it working).
suppliers[0] = { ...suppliers[0], status: "working" };
// Unrelated history → affinity miss → no free supplier → evict session 1
// (supplier /v1/chat/end) and retry, bypassing the stale indexer status.
const res2 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "different topic" }] });
expect(res2.status).toBe(200);
expect(res2.body.choices[0].message.content).toBe("New");
expect(startChat).toHaveBeenCalledTimes(2);
expect(endCalls).toHaveLength(1);
expect((messageCalls[1].messages as unknown[]).length).toBe(1);
// Evicted session closed + billed.
const keyId = deps.store.listAllKeys().find((k) => k.demo === 1)!.id;
const usage = deps.store.listUsage(keyId, 10);
expect(usage.some((u) => u.kind === "chat_session" && u.cost_lovelace === "1000000")).toBe(true);
expect(deps.store.listOpenSessionsByKey(keyId)).toHaveLength(1); // only session 2
});
it("surfaces the original no-supplier error when there is nothing to evict", async () => {
const suppliers = [{ ...SUPPLIERS[1], status: "working" }];
const { fetchFn, endCalls } = demoFetch([() => new Response("unused", { status: 500 })], () => suppliers);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const res = await request(createApp(deps))
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res.status).toBe(404);
expect(res.body.error.code).toBe("model_not_found");
expect(startChat).not.toHaveBeenCalled();
expect(endCalls).toHaveLength(0);
});
it("ticket mode: session close settles nothing and bills zero", async () => {
const { fetchFn } = demoFetch([
() => supplierSse([{ type: "token", value: "Hi" }, doneFrame({ role: "assistant", content: "Hi" }, "stop")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain, chatSettleMode: "ticket" });
const { rawKey, keyRow } = setupDemoKey(deps);
const ctx = deps.registry.getContext(keyRow);
const endChat = vi.fn(async (opts: { escrowRef: unknown }) => ({
settleMode: "ticket" as const,
escrowRef: opts.escrowRef,
}));
(ctx.sdk as { endChat: unknown }).endChat = endChat;
const app = createApp(deps);
const res1 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res1.status).toBe(200);
const session = deps.store.listOpenSessionsByKey(keyRow.id)[0];
const close = await request(app)
.post(`/openai/v1/chat/sessions/${session.id}/close`)
.set("authorization", `Bearer ${rawKey}`)
.send({});
expect(close.status).toBe(200);
expect(close.body.settle_mode).toBe("ticket");
expect(close.body.accepted_ref).toBeUndefined();
expect(endChat).toHaveBeenCalledTimes(1);
const usage = deps.store.listUsage(keyRow.id, 10);
expect(usage.some((u) => u.kind === "chat_session" && u.cost_lovelace === "0")).toBe(true);
});
it("ticket mode: janitor-closed demo sessions bill zero", async () => {
const { fetchFn, endCalls } = demoFetch([
() => supplierSse([{ type: "token", value: "Hi" }, doneFrame({ role: "assistant", content: "Hi" }, "stop")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain, chatSettleMode: "ticket" });
const { rawKey, keyRow } = setupDemoKey(deps);
const app = createApp(deps);
const res = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res.status).toBe(200);
await sweepIdleDemoSessions(deps, Date.now() + deps.config.demoSessionIdleMs + 1);
expect(endCalls.length).toBeGreaterThan(0);
const usage = deps.store.listUsage(keyRow.id, 10);
expect(usage.some((u) => u.kind === "chat_session" && u.cost_lovelace === "0")).toBe(true);
});
it("janitor closes idle demo sessions so the next request opens fresh", async () => {
const { fetchFn, messageCalls, endCalls } = demoFetch([
() => supplierSse([{ type: "token", value: "Hi" }, doneFrame({ role: "assistant", content: "Hi" }, "stop")]),
() => supplierSse([{ type: "token", value: "Fresh" }, doneFrame({ role: "assistant", content: "Fresh" }, "stop")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const app = createApp(deps);
const res1 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res1.status).toBe(200);
// Pretend demoSessionIdleMs elapsed: the sweeper tick closes the session.
await sweepIdleDemoSessions(deps, Date.now() + deps.config.demoSessionIdleMs + 1);
expect(endCalls).toHaveLength(1);
const keyId = deps.store.listAllKeys().find((k) => k.demo === 1)!.id;
expect(deps.store.listOpenSessionsByKey(keyId)).toHaveLength(0);
// Continuing the thread now misses affinity and reopens with full replay.
const res2 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({
model: "kimi",
messages: [
{ role: "user", content: "hi" },
{ role: "assistant", content: "Hi" },
{ role: "user", content: "more" },
],
});
expect(res2.status).toBe(200);
expect(startChat).toHaveBeenCalledTimes(2);
expect((messageCalls[1].messages as unknown[]).length).toBe(3);
});
it("blocks withdraw and hides the deposit address for demo keys", async () => {
const deps = makeDeps();
const { rawKey } = setupDemoKey(deps);
const app = createApp(deps);
const wd = await request(app)
.post("/account/withdraw")
.set("authorization", `Bearer ${rawKey}`)
.send({ to_address: "addr_test1abc" });
expect(wd.status).toBe(403);
expect(wd.body.error.code).toBe("demo_key_restricted");
const acct = await request(app).get("/account").set("authorization", `Bearer ${rawKey}`);
expect(acct.status).toBe(200);
expect(acct.body.deposit_address).toBeNull();
expect(acct.body.demo).toBe(true);
});
it("lists only chat-capable models for demo keys", async () => {
const fetchFn = (async (url: unknown) =>
new Response(JSON.stringify(String(url).includes("/suppliers") ? SUPPLIERS.filter((s) => !String(url).includes("capability_id") || s.capability_id === "llm.chat.v1") : []), { status: 200 })) as unknown as typeof globalThis.fetch;
const deps = makeDeps(fetchFn);
const { rawKey } = setupDemoKey(deps);
const app = createApp(deps);
const demoModels = await request(app).get("/openai/v1/models").set("authorization", `Bearer ${rawKey}`);
expect(demoModels.body.data.map((m: { id: string }) => m.id)).toEqual(["kimi"]);
const signup = await request(app).post("/signup").send({});
const normalModels = await request(app).get("/openai/v1/models").set("authorization", `Bearer ${signup.body.api_key as string}`);
expect(normalModels.body.data.map((m: { id: string }) => m.id).sort()).toEqual(["kimi", "qwen"]);
});
it("rate-limits demo requests per IP (and not per key)", async () => {
const deps = makeDeps(undefined, { demoIpRate: { max: 1, windowMs: 60_000 } });
const { rawKey } = setupDemoKey(deps);
const app = createApp(deps);
// Model with no chat suppliers → cheap 404 AFTER passing the limiter.
const body = { model: "nope", messages: [{ role: "user", content: "hi" }] };
const first = await request(app).post("/openai/v1/chat/completions").set("authorization", `Bearer ${rawKey}`).send(body);
expect(first.status).toBe(404);
const second = await request(app).post("/openai/v1/chat/completions").set("authorization", `Bearer ${rawKey}`).send(body);
expect(second.status).toBe(429);
// Non-demo keys are untouched by the demo IP limiter.
const signup = await request(app).post("/signup").send({});
const normal = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${signup.body.api_key as string}`)
.send(body);
expect(normal.status).not.toBe(429);
});
});

View File

@@ -0,0 +1,79 @@
/**
* migration.test.ts — the demo-column migration on pre-existing databases.
*
* GatewayStore has no migration framework (pure CREATE TABLE IF NOT EXISTS),
* so opening a DB created before the `demo` column must ALTER it in place.
*/
import { describe, it, expect } from "vitest";
import { tmpdir } from "os";
import { join } from "path";
import { mkdirSync } from "fs";
import { randomUUID } from "crypto";
import Database from "better-sqlite3";
import { GatewayStore } from "../src/db/store.js";
const LEGACY_API_KEYS_SQL = `
CREATE TABLE IF NOT EXISTS api_keys (
id TEXT PRIMARY KEY,
key_hash TEXT NOT NULL UNIQUE,
key_prefix TEXT NOT NULL,
label TEXT,
wallet_pkh TEXT NOT NULL,
deposit_address TEXT NOT NULL,
enc_priv_nonce TEXT NOT NULL,
enc_priv_ct TEXT NOT NULL,
enc_priv_tag TEXT NOT NULL,
master_key_version INTEGER NOT NULL DEFAULT 1,
created_at INTEGER NOT NULL,
disabled INTEGER NOT NULL DEFAULT 0
);
`;
function keyRow(id: string) {
return {
id,
key_hash: `hash-${id}`,
key_prefix: "vk_test_xxxx",
label: null,
wallet_pkh: "pkh",
deposit_address: "addr_test1x",
enc_priv_nonce: "n",
enc_priv_ct: "c",
enc_priv_tag: "t",
master_key_version: 1,
created_at: Date.now(),
};
}
describe("GatewayStore demo-column migration", () => {
it("adds the demo column to a legacy DB and defaults existing keys to 0", () => {
const dir = join(tmpdir(), `gw-mig-${randomUUID()}`);
mkdirSync(dir, { recursive: true });
const raw = new Database(join(dir, "gateway.db"));
raw.exec(LEGACY_API_KEYS_SQL);
raw
.prepare(
`INSERT INTO api_keys (id, key_hash, key_prefix, label, wallet_pkh, deposit_address,
enc_priv_nonce, enc_priv_ct, enc_priv_tag, master_key_version, created_at, disabled)
VALUES (@id, @key_hash, @key_prefix, @label, @wallet_pkh, @deposit_address,
@enc_priv_nonce, @enc_priv_ct, @enc_priv_tag, @master_key_version, @created_at, 0)`,
)
.run(keyRow("legacy"));
raw.close();
const store = new GatewayStore(dir);
const legacy = store.getKeyByHash("hash-legacy");
expect(legacy?.demo).toBe(0);
store.insertKey({ ...keyRow("demo-key"), demo: 1 });
expect(store.getKeyByHash("hash-demo-key")?.demo).toBe(1);
});
it("persists the demo flag on fresh databases (default 0)", () => {
const dir = join(tmpdir(), `gw-fresh-${randomUUID()}`);
const store = new GatewayStore(dir);
store.insertKey(keyRow("plain"));
expect(store.getKeyByHash("hash-plain")?.demo).toBe(0);
});
});

View File

@@ -0,0 +1,186 @@
/**
* sweeper.test.ts — reclaim/close decoupling + settled-refs memory.
*
* Ticket mode: reclaiming a live session's Open escrow must NOT close the
* session row or drop its transcript mirror (the conversation continues on a
* spent ticket). Full mode keeps close-on-reclaim. Claimed rows close even
* under ticket config (they can only come from full-mode suppliers). The
* settled-refs memory prevents re-attempting refs already settled, and the
* deliver_by gate skips reclaims the validator would reject.
*/
import { describe, it, expect, vi } from "vitest";
import { tmpdir } from "os";
import { join } from "path";
import { randomBytes, randomUUID } from "crypto";
import { GatewayStore } from "../src/db/store.js";
import { SdkRegistry } from "../src/sdk/registry.js";
import { genPrivKeyHex, deriveWalletKey } from "../src/wallet.js";
import { seal } from "../src/crypto/seal.js";
import { hashApiKey } from "../src/middleware/apiKeyAuth.js";
import { runSweepOnce } from "../src/sweeper.js";
import { transcripts } from "../src/openai/transcripts.js";
import type { GatewayConfig } from "../src/config.js";
import type { GatewayDeps } from "../src/deps.js";
import type { EscrowRow } from "../src/onchain/settle.js";
const MASTER = "ab".repeat(32);
function makeSweepDeps(opts: { chatSettleMode: "full" | "ticket"; escrows: EscrowRow[] }) {
const dbDir = join(tmpdir(), `gw-sweep-${randomUUID()}`);
const store = new GatewayStore(dbDir);
const config: GatewayConfig = {
masterKeyHex: MASTER,
indexerUrl: "http://ix",
ogmiosUrl: "http://og",
networkId: 0,
liveChain: true,
port: 0,
dbDir,
signupRate: { max: 1000, windowMs: 60_000 },
keyRate: { max: 1000, windowMs: 60_000 },
demoIpRate: { max: 1000, windowMs: 60_000 },
sweeperIntervalMs: 60_000,
demoSessionIdleMs: 180_000,
chatSettleMode: opts.chatSettleMode,
walletHealthIntervalMs: 600_000,
sdkRegistryMax: 100,
corsOrigins: [],
};
const chain = { queryUtxo: async () => null } as unknown as GatewayDeps["chain"];
const registry = new SdkRegistry({
chain,
indexerUrl: config.indexerUrl,
networkId: config.networkId,
masterKeyHex: config.masterKeyHex,
max: config.sdkRegistryMax,
});
const fetchFn = (async (url: unknown) => {
if (String(url).includes("/escrows")) return new Response(JSON.stringify(opts.escrows), { status: 200 });
return new Response("[]", { status: 200 });
}) as unknown as typeof globalThis.fetch;
const deps: GatewayDeps = { config, store, chain, registry, fetchFn };
const rawKey = `vk_test_${randomBytes(24).toString("hex")}`;
const privHex = genPrivKeyHex();
const walletKey = deriveWalletKey(privHex, 0);
const sealed = seal(privHex, MASTER);
store.insertKey({
id: randomUUID(),
key_hash: hashApiKey(rawKey),
key_prefix: rawKey.slice(0, 12),
label: "sweep-test",
wallet_pkh: walletKey.pubKeyHash,
deposit_address: walletKey.address,
enc_priv_nonce: sealed.nonce,
enc_priv_ct: sealed.ct,
enc_priv_tag: sealed.tag,
master_key_version: 1,
created_at: Date.now(),
demo: 1,
});
const keyRow = store.getKeyByHash(hashApiKey(rawKey))!;
return { deps, keyRow };
}
function insertOpenSession(deps: GatewayDeps, keyId: string, escrowRef: string): string {
const id = randomUUID();
deps.store.insertSession({
id,
key_id: keyId,
escrow_ref: escrowRef,
session_nonce: "nonce",
supplier_base_url: "http://sup",
supplier_pkh: "s1",
model: "kimi",
price_lovelace: "200000",
state: "open",
opened_at: Date.now() - 40 * 60 * 1000,
});
return id;
}
function openRow(refChar: string, opts?: Partial<EscrowRow>): EscrowRow {
return {
utxo_ref: `${refChar.repeat(64)}#0`,
state: "Open",
posted_at: Date.now() - 40 * 60 * 1000,
deliver_by: Date.now() - 5 * 60 * 1000,
...opts,
};
}
const settleOk = () => ({
accept: vi.fn(async () => undefined) as never,
reclaim: vi.fn(async () => undefined) as never,
});
describe("sweeper — ticket-mode reclaim/close decoupling", () => {
it("reclaims a live session's Open ticket but keeps the session (and mirror) alive", async () => {
const row = openRow("1");
const { deps, keyRow } = makeSweepDeps({ chatSettleMode: "ticket", escrows: [row] });
const sessionId = insertOpenSession(deps, keyRow.id, row.utxo_ref);
transcripts.set(sessionId, [{ role: "user", content: "hi" }]);
const settle = settleOk();
await runSweepOnce(deps, settle);
expect(settle.reclaim).toHaveBeenCalledTimes(1);
expect(deps.store.getSession(sessionId)!.state).toBe("open"); // NOT closed
expect(transcripts.get(sessionId)).toBeDefined(); // mirror intact
expect(deps.store.listUsage(keyRow.id, 10)).toHaveLength(0); // nothing billed
transcripts.delete(sessionId);
});
it("full mode: reclaim closes the session row", async () => {
const row = openRow("2");
const { deps, keyRow } = makeSweepDeps({ chatSettleMode: "full", escrows: [row] });
const sessionId = insertOpenSession(deps, keyRow.id, row.utxo_ref);
const settle = settleOk();
await runSweepOnce(deps, settle);
expect(settle.reclaim).toHaveBeenCalledTimes(1);
expect(deps.store.getSession(sessionId)!.state).toBe("closed");
const usage = deps.store.listUsage(keyRow.id, 10);
expect(usage.some((u) => u.status === "reclaimed")).toBe(true);
});
it("ticket config still closes sessions behind Claimed rows (full-mode suppliers)", async () => {
const row = openRow("3", { state: "Claimed" });
const { deps, keyRow } = makeSweepDeps({ chatSettleMode: "ticket", escrows: [row] });
const sessionId = insertOpenSession(deps, keyRow.id, row.utxo_ref);
const settle = settleOk();
await runSweepOnce(deps, settle);
expect(settle.reclaim).toHaveBeenCalledTimes(1);
expect(deps.store.getSession(sessionId)!.state).toBe("closed");
});
});
describe("sweeper — reclaim gating + settled-refs memory", () => {
it("skips reclaim while deliver_by has not passed", async () => {
const row = openRow("4", { deliver_by: Date.now() + 10 * 60 * 1000 });
const { deps, keyRow } = makeSweepDeps({ chatSettleMode: "ticket", escrows: [row] });
insertOpenSession(deps, keyRow.id, row.utxo_ref);
const settle = settleOk();
await runSweepOnce(deps, settle);
expect(settle.reclaim).not.toHaveBeenCalled();
});
it("does not re-attempt refs it already settled (in-process memory)", async () => {
const row = openRow("5");
const { deps, keyRow } = makeSweepDeps({ chatSettleMode: "ticket", escrows: [row] });
insertOpenSession(deps, keyRow.id, row.utxo_ref);
const settle = settleOk();
await runSweepOnce(deps, settle);
await runSweepOnce(deps, settle);
expect(settle.reclaim).toHaveBeenCalledTimes(1);
});
});

View File

@@ -84,10 +84,31 @@ describe("routing/selectSupplier", () => {
expect(await listModels({ indexerUrl: "http://ix", fetchFn: jsonResponse(rows) })).toEqual(["m", "other"]);
});
it("lists only one capability's models when capabilityId is given", async () => {
expect(await listModels({ indexerUrl: "http://ix", capabilityId: "llm.chat.v1", fetchFn: jsonResponse(rows) })).toEqual(["m"]);
});
it("parseRef", () => {
expect(parseRef("aa".repeat(32) + "#3")).toEqual({ txHash: "aa".repeat(32), index: 3 });
expect(parseRef("nope")).toBeNull();
});
it("ignoreStatusFor readmits a busy supplier by pkh (stale indexer status)", async () => {
const busy = [
{ ...rows[0], status: "working" },
{ ...rows[1], status: "offline" },
];
const base = { indexerUrl: "http://ix", model: "m", capabilityId: "llm.text.generate.v1" };
const excluded = await selectCandidates({ ...base, fetchFn: jsonResponse(busy) });
expect(excluded).toEqual([]);
const readmitted = await selectCandidates({
...base,
fetchFn: jsonResponse(busy),
ignoreStatusFor: new Set(["s1"]),
});
expect(readmitted.map((c) => c.supplierPkh)).toEqual(["s1"]);
});
});
describe("openai/validate", () => {
@@ -104,6 +125,37 @@ describe("openai/validate", () => {
expect(() => parseChatRequest({ model: "m", messages: [] })).toThrow();
expect(() => parseChatRequest({ messages: [{ role: "user", content: "x" }] })).toThrow();
});
const TOOL_CALL = { id: "call_1", type: "function", function: { name: "get_time", arguments: "{}" } };
it("allowTools accepts tools, tool role, null content and assistant tool_calls", () => {
const p = parseChatRequest(
{
model: "m",
messages: [
{ role: "user", content: "hi" },
{ role: "assistant", content: null, tool_calls: [TOOL_CALL] },
{ role: "tool", content: "3pm", tool_call_id: "call_1" },
],
tools: [{ type: "function", function: { name: "get_time", parameters: {} } }],
tool_choice: "auto",
},
{ allowTools: true },
);
expect(p.tools).toHaveLength(1);
expect(p.toolChoice).toBe("auto");
expect(p.messages[1]).toEqual({ role: "assistant", content: "", tool_calls: [TOOL_CALL] });
expect(p.messages[2]).toEqual({ role: "tool", content: "3pm", tool_call_id: "call_1" });
});
it("allowTools still rejects legacy functions and the default path still rejects tool role", () => {
expect(() =>
parseChatRequest({ model: "m", messages: [{ role: "user", content: "x" }], functions: [] }, { allowTools: true }),
).toThrow(GatewayError);
expect(() =>
parseChatRequest({ model: "m", messages: [{ role: "tool", content: "x", tool_call_id: "c" }] }),
).toThrow(GatewayError);
});
});
describe("openai/shapes", () => {
@@ -121,6 +173,24 @@ describe("openai/shapes", () => {
expect(ch.object).toBe("chat.completion.chunk");
expect((ch.choices as any)[0].delta.content).toBe("tok");
});
it("tool_calls chunk + finish_reason tool_calls", () => {
const tc = { index: 0, id: "call_1", type: "function" as const, function: { name: "f", arguments: "{}" } };
const ch = buildChunk({ id: "id", model: "m", delta: { tool_calls: [tc] }, finishReason: "tool_calls" });
expect((ch.choices as any)[0].delta.tool_calls).toEqual([tc]);
expect((ch.choices as any)[0].finish_reason).toBe("tool_calls");
});
it("completion with tool_calls and finish_reason", () => {
const cc = buildChatCompletion({
id: "id",
model: "m",
content: "",
usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 },
toolCalls: [{ id: "call_1", type: "function", function: { name: "f", arguments: "{}" } }],
finishReason: "tool_calls",
});
expect((cc.choices as any)[0].message.tool_calls).toHaveLength(1);
expect((cc.choices as any)[0].finish_reason).toBe("tool_calls");
});
it("renderMessages folds roles", () => {
expect(renderMessages([{ role: "system", content: "be nice" }, { role: "user", content: "hi" }])).toBe("System: be nice\n\nUser: hi");
});

View File

@@ -0,0 +1,88 @@
/**
* tx/chatMessage.ts — canonical ChatMessage normalization + equivalence.
*
* The supplier's session transcript and the gateway's in-memory mirror must be
* hash-identical (receipt response_hash = sha256(canonicalize(transcript))),
* so both sides build messages through normalizeChatMessage: optional fields
* are truly absent when empty, content is never null, and vendor extras are
* stripped. chatMessagesEquivalent is the tolerant comparison used by the
* demo-endpoint transcript-prefix affinity matcher.
*/
import type { ChatMessage, ToolCall } from "./types.js";
const ROLES = new Set(["system", "user", "assistant", "tool"]);
function normalizeToolCall(raw: unknown): ToolCall | null {
if (typeof raw !== "object" || raw === null) return null;
const tc = raw as Record<string, unknown>;
const fn = tc.function as Record<string, unknown> | undefined;
if (typeof tc.id !== "string" || tc.id === "") return null;
if (tc.type !== "function") return null;
if (typeof fn !== "object" || fn === null) return null;
if (typeof fn.name !== "string" || fn.name === "") return null;
if (typeof fn.arguments !== "string") return null;
return { id: tc.id, type: "function", function: { name: fn.name, arguments: fn.arguments } };
}
/**
* Validate and canonicalize an untrusted message-shaped value.
* Returns null when the value is not a usable ChatMessage.
*/
export function normalizeChatMessage(raw: unknown): ChatMessage | null {
if (typeof raw !== "object" || raw === null) return null;
const m = raw as Record<string, unknown>;
if (typeof m.role !== "string" || !ROLES.has(m.role)) return null;
const role = m.role as ChatMessage["role"];
let content: string;
if (typeof m.content === "string") content = m.content;
else if (m.content === null || m.content === undefined) content = "";
else return null;
const msg: ChatMessage = { role, content };
if (m.tool_calls !== undefined && m.tool_calls !== null) {
if (role !== "assistant" || !Array.isArray(m.tool_calls)) return null;
const calls: ToolCall[] = [];
for (const raw of m.tool_calls) {
const tc = normalizeToolCall(raw);
if (!tc) return null;
calls.push(tc);
}
if (calls.length > 0) msg.tool_calls = calls;
}
if (m.tool_call_id !== undefined && m.tool_call_id !== null) {
if (role !== "tool" || typeof m.tool_call_id !== "string" || m.tool_call_id === "") return null;
msg.tool_call_id = m.tool_call_id;
}
if (role === "tool" && msg.tool_call_id === undefined) return null;
return msg;
}
export function toolCallsEqual(a?: ToolCall[], b?: ToolCall[]): boolean {
const la = a ?? [];
const lb = b ?? [];
if (la.length !== lb.length) return false;
return la.every((tc, i) =>
tc.id === lb[i].id &&
tc.function.name === lb[i].function.name &&
tc.function.arguments === lb[i].function.arguments,
);
}
/**
* Tolerant equality for affinity prefix matching. Both sides are expected to
* already be normalized; comparison ignores incidental key order and treats
* missing tool fields as empty.
*/
export function chatMessagesEquivalent(a: ChatMessage, b: ChatMessage): boolean {
return (
a.role === b.role &&
a.content === b.content &&
(a.tool_call_id ?? "") === (b.tool_call_id ?? "") &&
toolCallsEqual(a.tool_calls, b.tool_calls)
);
}

View File

@@ -2,8 +2,9 @@
* tx/index.ts — public exports for M1-B tx builders.
*/
export type { ChatMessage, WalletKey, BuildResult, PostAdvertBuildResult, PostEscrowBuildResult } from "./types.js";
export type { ChatMessage, ToolCall, WalletKey, BuildResult, PostAdvertBuildResult, PostEscrowBuildResult } from "./types.js";
export { TxConstructionError } from "./types.js";
export { normalizeChatMessage, chatMessagesEquivalent, toolCallsEqual } from "./chatMessage.js";
export { mockSlotToWallclockMs, mockWallclockMsToSlot, NETWORK_BUFFER_MS } from "./internal/constants.js";
export { detectCborBackend, type CborBackend } from "./internal/cborBackend.js";
export type { Signer } from "./signer.js";

View File

@@ -358,6 +358,45 @@ function rethrowAsTxError(err: unknown, fallback: string): never {
throw new TxConstructionError(fallback, msg);
}
/** Choose PostEscrow inputs that PRESERVE a pure-ADA collateral UTxO for the
* later Accept/Reclaim script-spend. PostEscrow is NOT a script spend, so lucid
* is otherwise free to consume the dedicated ≥5-ADA collateral UTxO as an
* ordinary input — leaving the wallet with no collateral candidate and
* stranding the escrow when Accept later runs (the fragmentation bug observed
* on back-to-back requests).
*
* Reserve the SMALLEST pure-ADA UTxO ≥ collateral floor and fund the escrow
* from the remaining UTxOs. Returns null (→ default lucid coin selection) when
* reservation isn't possible or needed:
* - a single UTxO (nothing to reserve),
* - no qualifying pure-ADA collateral UTxO, or
* - excluding it would underfund the escrow. The large-single-UTxO case is
* safe under default selection anyway: the change output is itself a
* pure-ADA UTxO well above the collateral floor. */
export function reserveCollateralInputs(
utxos: LucidUTxO[],
lockedLovelace: bigint,
): LucidUTxO[] | null {
if (utxos.length < 2) return null;
const candidates = utxos
.filter(
(u) =>
u.assets.lovelace >= COLLATERAL_MIN_LOVELACE &&
Object.keys(u.assets).every((k) => k === "lovelace"),
)
.sort((a, b) => (a.assets.lovelace < b.assets.lovelace ? -1 : 1));
if (candidates.length === 0) return null;
const reserved = candidates[0];
const remaining = utxos.filter(
(u) => !(u.txHash === reserved.txHash && u.outputIndex === reserved.outputIndex),
);
if (remaining.length === 0) return null;
const remainingLovelace = remaining.reduce((s, u) => s + u.assets.lovelace, 0n);
// Cover the locked output + tx fee + a min-change cushion; else fall back.
if (remainingLovelace < lockedLovelace + 2_000_000n) return null;
return remaining;
}
// ─── PostEscrow ──────────────────────────────────────────────────────
export async function buildLiveTxForEscrow(
@@ -382,6 +421,12 @@ export async function buildLiveTxForEscrow(
const lockedLovelace = totalLocked < MIN_UTXO_LOVELACE ? MIN_UTXO_LOVELACE : totalLocked;
// Preserve a pure-ADA collateral UTxO so the buyer's later Accept/Reclaim
// script-spend can find collateral. Without this, PostEscrow coin selection
// can consume the 5-ADA collateral UTxO and strand the escrow at Accept.
const walletUtxos = await lucid.wallet().getUtxos();
const reservedInputs = reserveCollateralInputs(walletUtxos, lockedLovelace);
let signed;
try {
const txBuilder = lucid
@@ -404,7 +449,12 @@ export async function buildLiveTxForEscrow(
// becomes change. Same fix as Claim/Submit/Accept paths.
.setMinFee(500_000n);
const completed = await txBuilder.complete();
// When we can reserve collateral, restrict coin selection to the remaining
// (working) UTxOs so the collateral candidate survives for Accept. Else
// fall back to lucid's default selection (single/large-UTxO cases are safe).
const completed = reservedInputs
? await txBuilder.complete({ presetWalletInputs: reservedInputs, localUPLCEval: false })
: await txBuilder.complete();
signed = await completed.sign.withWallet().complete();
} catch (err) {
rethrowAsTxError(err, "post-escrow build failed");

View File

@@ -0,0 +1,42 @@
import { describe, it, expect } from "vitest";
import { reserveCollateralInputs } from "./liveCbor.js";
// Minimal LucidUTxO-shaped fixtures (the helper only reads txHash/outputIndex/assets).
const pure = (id: string, lovelace: bigint) =>
({ txHash: id, outputIndex: 0, assets: { lovelace } }) as never;
const mixed = (id: string, lovelace: bigint) =>
({ txHash: id, outputIndex: 0, assets: { lovelace, "policy.tok": 1n } }) as never;
const LOCKED = 2_200_000n; // price + buyer_bond + supplier_bond
const ids = (r: ReturnType<typeof reserveCollateralInputs>) =>
(r ?? []).map((u: { txHash: string }) => u.txHash);
describe("reserveCollateralInputs", () => {
it("reserves a 5-ADA collateral UTxO and funds from the rest", () => {
const r = reserveCollateralInputs([pure("collat", 5_000_000n), pure("work", 6_000_000n)], LOCKED);
expect(ids(r)).toEqual(["work"]); // collat (smallest ≥5) reserved
});
it("reserves the SMALLEST qualifying candidate", () => {
const r = reserveCollateralInputs([pure("big", 9_000_000n), pure("small", 5_000_000n)], LOCKED);
expect(ids(r)).toEqual(["big"]); // small reserved as collateral
});
it("only reserves PURE-ADA UTxOs (ignores ones carrying native assets)", () => {
const r = reserveCollateralInputs([mixed("mixed", 10_000_000n), pure("purecol", 6_000_000n)], LOCKED);
expect(ids(r)).toEqual(["mixed"]); // purecol reserved, fund from the mixed UTxO
});
it("returns null for a single UTxO (large change stays as collateral under default selection)", () => {
expect(reserveCollateralInputs([pure("only", 12_000_000n)], LOCKED)).toBeNull();
});
it("returns null when no UTxO qualifies as collateral", () => {
expect(reserveCollateralInputs([pure("a", 4_000_000n), pure("b", 4_000_000n)], LOCKED)).toBeNull();
});
it("returns null when reserving collateral would underfund the escrow", () => {
// {5 collateral, 3 working}: working can't cover locked + cushion → fall back.
expect(reserveCollateralInputs([pure("collat", 5_000_000n), pure("work", 3_000_000n)], LOCKED)).toBeNull();
});
});

View File

@@ -11,14 +11,29 @@
import type { OutputReference } from "../chain/ChainProvider.js";
/**
* ToolCall — OpenAI-compatible function tool call emitted by an assistant turn.
* `function.arguments` is the raw JSON string as produced by the model.
*/
export interface ToolCall {
id: string;
type: "function";
function: { name: string; arguments: string };
}
/**
* ChatMessage — OpenAI-compatible chat message.
* v1 supports user/system/assistant roles. The wire shape is intentionally
* narrow; future capabilities (tools, multimodal) will extend or replace it.
* `content` is always a string ("" for tool-call-only assistant messages —
* OpenAI's `content: null` is normalized on ingest). `tool_calls` /
* `tool_call_id` MUST be omitted (not set to undefined) when absent: the
* receipt hash canonicalizes the transcript objects, so field presence is
* part of the identity shared between supplier transcript and gateway mirror.
*/
export interface ChatMessage {
role: "system" | "user" | "assistant";
role: "system" | "user" | "assistant" | "tool";
content: string;
tool_calls?: ToolCall[];
tool_call_id?: string;
}
/**

View File

@@ -126,3 +126,9 @@ ADVERT_DETAIL_HASH=
# `docker compose down && up -d` to apply.
# ESCROW_REF_UTXO=<txhash>#0
# ADVERT_REF_UTXO=<txhash>#1
# ─── Optional: chat-session tuning (CAPABILITY_KIND=chat-session only) ─────
# CHAT_IDLE_TIMEOUT_MS=300000 # auto-end sessions idle this long
# MAX_CHAT_SESSIONS=1 # concurrent session slots (default 1)
# CHAT_SETTLE_MODE=full # "full" (Claim/Submit) or "ticket" (no supplier
# # chain ops; buyer reclaims the Open escrow)

View File

@@ -0,0 +1,114 @@
# supplier/.env.huggingface-chat.example
#
# Preset for a chat supplier that uses the HuggingFace Inference Providers router
# (https://router.huggingface.co) as its COMPUTE BACKEND to fulfil bonded
# inference work commissioned on Vector. It is the OpenAI-compatible on-ramp:
# the router is OpenAI-compatible, so this reuses the existing LLM_BACKEND=openai
# path with NO code changes — it is config only.
#
# Copy to supplier/.env on the host (chmod 600) and fill in real values.
# .env is gitignored — NEVER commit a populated copy.
#
# Runbook: docs/HUGGINGFACE_ROUTER_SETUP.md
#
# To generate a fresh Ed25519 wallet seed: openssl rand -hex 32
# Lines marked SECRET must never be shared, logged, or committed.
# ─── Required ─────────────────────────────────────────────────────────────
# 64-char hex Ed25519 private key seed for this supplier's wallet. SECRET.
SUPPLIER_PRIV_KEY_HEX=PASTE_64_HEX_CHARS_HERE
# Ogmios WebSocket URL. Testnet shown; use the mainnet URL for NETWORK_ID=1:
# testnet: wss://ogmios.vector.testnet.apexfusion.org
# mainnet: wss://ogmios.vector.mainnet.apexfusion.org
OGMIOS_URL=wss://ogmios.vector.testnet.apexfusion.org
# OutputReference of this supplier's on-chain advertisement UTxO.
# Format "<64-hex-tx-hash>#<int>". Post it first (see tx:post-advert below),
# then paste the printed "<txHash>#0" here.
ADVERT_REF=PASTE_TXHASH_HERE#0
# Cardano network ID. "0" = testnet, "1" = mainnet. Keep "0" until you have
# validated the setup end-to-end on testnet.
NETWORK_ID=0
# ─── Backend: HuggingFace router (OpenAI-compatible) ──────────────────────
# Route the chat job runner at the HF Inference Providers router instead of a
# local Ollama. The supplier POSTs to ${OPENAI_BASE_URL}/v1/chat/completions.
LLM_BACKEND=openai
# HF router base URL. The supplier appends /v1/chat/completions itself, so do
# NOT include the /v1 suffix here.
OPENAI_BASE_URL=https://router.huggingface.co
# HuggingFace access token (starts with "hf_"). Create one at
# https://huggingface.co/settings/tokens with "Make calls to Inference
# Providers" permission. Forwarded as `Authorization: Bearer <token>`. SECRET.
OPENAI_API_KEY=PASTE_HF_TOKEN_HERE
# First-token latency across hosted providers can exceed local Ollama — keep a
# generous request timeout.
OPENAI_TIMEOUT_MS=180000
# ─── IMPORTANT: do NOT set OPENAI_REASONING here ──────────────────────────
# `reasoning:{enabled:false}` is an OpenRouter-only param. The HF router
# rejects it with HTTP 400 on every request. The supplier refuses to boot if
# OPENAI_REASONING is disabled while OPENAI_BASE_URL points at the HF router.
# (This is the #1 footgun when copying an OpenRouter env — leave it unset.)
# OPENAI_REASONING=
# Optional hard ceiling forwarded as `max_tokens`. Unset = provider grants the
# max available (context − prompt). Keep below the model's context window.
# OPENAI_MAX_TOKENS=
# ─── Enable real chain submissions ────────────────────────────────────────
# LIVE_CHAIN: only the literal "1" opts in to real Claim/Submit txs. Leave
# unset for a dry run (read-only, no chain writes) while testing.
LIVE_CHAIN=
# HTTP listen port inside the container. Compose maps this to the host.
PORT=8080
# ─── tx:post-advert — run once before first deploy ────────────────────────
# Execute (from repo root):
# pnpm --filter @marketplace/supplier tx:post-advert \
# --capability-id llm.text.generate.v1 \
# --model deepseek-ai/DeepSeek-V3:fastest \
# --max-output-tokens 512 \
# --max-processing-ms 60000 \
# --price-lovelace 2000000 \
# --endpoint-url https://mp-supplier.vector.testnet.apexfusion.org
# Copy the printed "<txHash>#0" into ADVERT_REF above.
#
# NOTE on --model: HF uses PROVIDER-QUALIFIED model names. The bare id picks a
# default provider; a ":<hint>" suffix pins routing:
# deepseek-ai/DeepSeek-V3:fastest # cheapest/fastest provider for the model
# deepseek-ai/DeepSeek-V3:together # pin a specific provider
# openai/gpt-oss-120b # open-weight, HF-hosted
# Browse models + providers at https://huggingface.co/models?inference_provider=all
# The value lives on-chain in AdvertDatum.model — it is data, not code, so you
# can advertise any router-supported model without touching the supplier.
ADVERT_CAPABILITY_ID=llm.text.generate.v1
ADVERT_MODEL=deepseek-ai/DeepSeek-V3:fastest
ADVERT_MAX_OUTPUT_TOKENS=512
ADVERT_MAX_PROCESSING_MS=60000
# Flat AP3X amount committed in escrow per job (1 AP3X = 1_000_000 lovelace).
# The HF backend meters PER TOKEN while this committed amount is fixed per job,
# so size it for your worst-case prompt length and cap output tightly via
# ADVERT_MAX_OUTPUT_TOKENS. See docs/HUGGINGFACE_ROUTER_SETUP.md §4.
ADVERT_PRICE_LOVELACE=2000000
ADVERT_SUPPLIER_BOND_LOVELACE=1000000
ADVERT_BUYER_BOND_LOVELACE=1000000
# HTTPS endpoint buyers will call (must be publicly reachable).
ADVERT_ENDPOINT_URL=https://mp-supplier.vector.testnet.apexfusion.org
# Optional off-chain detail pointer + its sha256 (leave empty for v1).
ADVERT_DETAIL_URI=
ADVERT_DETAIL_HASH=
# ─── Optional: CIP-33 reference scripts (recommended on mainnet) ──────────
# ESCROW_REF_UTXO=<txhash>#0
# ADVERT_REF_UTXO=<txhash>#1

View File

@@ -8,9 +8,10 @@
* turns that run fully OFF-CHAIN, and is settled at /v1/chat/end (or by the
* idle watchdog) when the supplier Submits a receipt over the whole transcript.
*
* Single-slot: the supplier serves one paid chat at a time (SupplierState
* mutex), so at most one record is "active". Ended records are retained so
* /v1/chat/end is idempotent (returns the same receipt).
* Concurrency: up to MAX_CHAT_SESSIONS records are "active" at once
* (SupplierState slot admission). Ended records are retained so /v1/chat/end
* is idempotent (returns the same receipt) and so a used ticket-mode escrow
* ref can never open a second session.
*
* Timer handles (idle + hard-cap) live on the record; server.ts arms them and
* endChatSession clears them.
@@ -31,7 +32,11 @@ export interface ChatSessionEndResult {
export interface ChatSessionRecord {
escrowRef: string;
claimedRef: OutputReference;
/** Absent in ticket mode (the escrow is never Claimed). */
claimedRef?: OutputReference;
/** "full" = Claim/Submit lifecycle; "ticket" = no chain ops after the
* buyer's escrow post. */
settleMode: "full" | "ticket";
advert: AdvertDatum;
escrowDatum: EscrowDatum;
/** Full ordered conversation (user + assistant turns). Hashed at End. */
@@ -49,7 +54,8 @@ export interface ChatSessionRecord {
export interface CreateChatSessionParams {
escrowRef: string;
claimedRef: OutputReference;
claimedRef?: OutputReference;
settleMode?: "full" | "ticket";
advert: AdvertDatum;
escrowDatum: EscrowDatum;
}
@@ -62,6 +68,7 @@ export class ChatSessionStore {
const record: ChatSessionRecord = {
escrowRef: params.escrowRef,
claimedRef: params.claimedRef,
settleMode: params.settleMode ?? "full",
advert: params.advert,
escrowDatum: params.escrowDatum,
transcript: [],
@@ -92,19 +99,37 @@ export class ChatSessionStore {
r.lastActivityMs = Date.now();
}
/** Append a whole turn delta (e.g. user msg, or tool results) verbatim. */
appendMessages(escrowRef: string, messages: ChatMessage[]): void {
const r = this.records.get(escrowRef);
if (!r) return;
r.transcript.push(...messages);
r.lastActivityMs = Date.now();
}
appendAssistant(
escrowRef: string,
content: string,
message: ChatMessage,
usage: { prompt_tokens: number; completion_tokens: number },
): void {
const r = this.records.get(escrowRef);
if (!r) return;
r.transcript.push({ role: "assistant", content });
r.transcript.push(message);
r.promptTokens += usage.prompt_tokens;
r.completionTokens += usage.completion_tokens;
r.lastActivityMs = Date.now();
}
/** Roll back a failed turn: truncate the transcript to `length`. The gateway
* mirror only appends after a successful turn, so the delta appended at turn
* start must be removed on LLM failure to keep both sides hash-identical. */
truncateTranscript(escrowRef: string, length: number): void {
const r = this.records.get(escrowRef);
if (!r || length < 0 || r.transcript.length <= length) return;
r.transcript.length = length;
r.lastActivityMs = Date.now();
}
touch(escrowRef: string): void {
const r = this.records.get(escrowRef);
if (r) r.lastActivityMs = Date.now();

View File

@@ -1,20 +1,22 @@
/**
* supplier/src/chatSessionRunner.ts — settles an open chat session.
* supplier/src/chatSessionRunner.ts — ends an open chat session.
*
* endChatSession is the chat-session analogue of runChatJob's receipt+Submit
* tail (jobRunner.ts steps 3-7). It is invoked from BOTH /v1/chat/end (user
* clicked "End chat") and the idle/hard-cap watchdog (abandoned session). It:
* 1. atomically flips the session "active" → "ending" (idempotent re-entry)
* 2. clears the idle + hard-cap timers
* 3. builds + signs a receipt over the FULL transcript
* (response_hash = sha256(canonical(transcript)); the buyer recomputes
* this from its local transcript mirror to verify)
* 4. buildSubmitTx against the Claimed UTxO → awaitTx (60s)
* 5. stores the terminal result and ALWAYS releases the supplier lock
* endChatSession is invoked from BOTH /v1/chat/end (user clicked "End chat")
* and the idle/hard-cap watchdog (abandoned session). Behavior branches on
* the record's settleMode:
*
* full — chat-session analogue of runChatJob's receipt+Submit tail:
* build + sign a receipt over the FULL transcript
* (response_hash = sha256(canonical(transcript))), then
* buildSubmitTx against the Claimed UTxO → awaitTx (60s), inside
* the wallet mutex (never race another wallet spend).
* ticket — no chain work at all: mark ended and free the slot. The escrow
* was never Claimed; the buyer's sweeper Reclaims it after
* deliver_by.
*
* Never throws. On any failure the session is marked "ended" with endFailure
* and the lock is released (the escrow then lingers Claimed until the buyer
* Reclaims after deliver_by — same fate as a failed one-off job).
* and the slot is released (a full-mode escrow then lingers Claimed until the
* buyer Reclaims after deliver_by — same fate as a failed one-off job).
*/
import { canonicalize } from "@marketplace/shared/cbor";
@@ -41,7 +43,7 @@ export interface EndChatSessionParams {
}
/**
* Settle a chat session. Returns the (now terminal) record, or null if no
* End a chat session. Returns the (now terminal) record, or null if no
* session exists for the ref. Idempotent: a second call returns the same
* record without re-submitting.
*/
@@ -59,6 +61,21 @@ export async function endChatSession(
deps.sessions.clearTimers(record);
try {
if (record.settleMode === "ticket") {
// No chain work: the un-Claimed escrow is the buyer's to reclaim.
record.status = "ended";
return record;
}
const claimedRef = record.claimedRef;
if (!claimedRef) {
// Full-mode record without a Claim ref should be impossible; end
// defensively rather than throw (this path must never leak the slot).
console.warn(`[chat_end_failed] escrow=${escrowRef} trigger=${trigger ?? "end"} reason=missing_claimed_ref`);
record.endFailure = { reason: "missing_claimed_ref", message: "full-mode session has no claimedRef" };
record.status = "ended";
return record;
}
// ── Receipt over the full transcript ──────────────────────────────────
const transcriptHash = sha256Hex(canonicalize(record.transcript));
const receipt = buildReceipt({
@@ -74,27 +91,35 @@ export async function endChatSession(
const signed = signReceipt(receipt, deps.supplierKey.privateKeyHex);
const resultHash = receiptResultHash(signed);
// ── Submit (Claimed → Submitted) ──────────────────────────────────────
let buildResult: { txCborHex: string; expectedTxHash: string };
try {
buildResult = await buildSubmitTx({
chain: deps.chain,
supplierKey: deps.supplierKey,
escrowRef: record.claimedRef,
receiptHash: resultHash,
});
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
// ── Submit (Claimed → Submitted): one wallet critical section ─────────
const submitOutcome = await deps.state.walletMutex.run(async () => {
let built: { txCborHex: string; expectedTxHash: string };
try {
built = await buildSubmitTx({
chain: deps.chain,
supplierKey: deps.supplierKey,
escrowRef: claimedRef,
receiptHash: resultHash,
});
} catch (err) {
return { kind: "build_failed" as const, err };
}
try {
await deps.chain.awaitTx(built.expectedTxHash, 60_000);
} catch (err) {
return { kind: "await_failed" as const, err };
}
return { kind: "ok" as const, built };
});
if (submitOutcome.kind === "build_failed") {
const message = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err);
console.warn(`[chat_end_failed] escrow=${escrowRef} trigger=${trigger ?? "end"} reason=submit_failed msg=${message}`);
record.endFailure = { reason: "submit_failed", message: `Submit tx failed: ${message}` };
record.status = "ended";
return record;
}
try {
await deps.chain.awaitTx(buildResult.expectedTxHash, 60_000);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
if (submitOutcome.kind === "await_failed") {
const message = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err);
console.warn(`[chat_end_failed] escrow=${escrowRef} trigger=${trigger ?? "end"} reason=submit_timeout msg=${message}`);
record.endFailure = { reason: "submit_timeout", message: `Submit awaitTx failed: ${message}` };
record.status = "ended";
@@ -105,14 +130,14 @@ export async function endChatSession(
receipt: signed.receipt as unknown as Record<string, unknown>,
receipt_signature: signed.signature,
// The Submit tx output (index 0) is the new Submitted UTxO the buyer Accepts.
submitted_ref: `${buildResult.expectedTxHash}#0`,
submitted_ref: `${submitOutcome.built.expectedTxHash}#0`,
};
record.status = "ended";
return record;
} finally {
// Always release the single-slot lock so the next chat can start.
// Always release this session's slot so the next chat can start.
try {
deps.state.release();
deps.state.release(record.escrowRef);
} catch {
/* never let lock-release escape */
}

View File

@@ -15,7 +15,10 @@
*
* Required for CAPABILITY_KIND="chat" + LLM_BACKEND="openai":
* OPENAI_BASE_URL — http(s):// URL for an OpenAI-compatible /v1/chat/completions
* host (e.g. ChatMock localhost proxy backed by Codex OAuth).
* host. Examples: ChatMock localhost proxy (Codex OAuth),
* OpenRouter (https://openrouter.ai/api), or the HuggingFace
* Inference Providers router (https://router.huggingface.co).
* See docs/HUGGINGFACE_ROUTER_SETUP.md for the HF preset.
*
* Required for CAPABILITY_KIND="tts":
* PIPER_URL — http(s):// URL for the openedai-speech-min PiperTTS host
@@ -51,6 +54,15 @@ 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*$/;
/** Lowercased hostname of a URL, or "" if it doesn't parse. */
function hostnameOf(url: string): string {
try {
return new URL(url).hostname.toLowerCase();
} catch {
return "";
}
}
export interface AdvertRef {
txHash: string;
index: number;
@@ -66,6 +78,13 @@ export type CapabilityKind = "chat" | "tts" | "chat-session";
/** LLM backend selected for the chat capability. */
export type LlmBackend = "ollama" | "openai";
/** Chain settlement mode for chat sessions.
* "full" — Claim at start, Submit receipt at end (buyer Accepts). Default.
* "ticket" — the buyer's escrow post is the only chain op; the supplier
* verifies it as an entry ticket but never Claims/Submits, and
* the buyer's sweeper Reclaims the Open escrow after deliver_by. */
export type ChatSettleMode = "full" | "ticket";
export interface SupplierConfig {
supplierPrivKeyHex: string;
ogmiosUrl: string;
@@ -79,6 +98,19 @@ export interface SupplierConfig {
* proxies). Required for hosted OpenAI-compatible APIs (DeepSeek, OpenAI, etc.).
*/
openaiApiKey: string;
/**
* When true, upstream OpenAI/OpenRouter calls send `reasoning:{enabled:false}`
* to disable "thinking" tokens (faster/cheaper straight answers; avoids
* reasoning starving the completion into a length-truncated empty answer).
* Set via OPENAI_REASONING=off. Default false. OpenRouter-only.
*/
openaiReasoningDisabled: boolean;
/**
* Optional `max_tokens` ceiling forwarded on upstream calls. 0 = omit (the
* provider then grants the max available = context − prompt). Must stay below
* the model context or the provider 400s (e.g. kimi context = 262144).
*/
openaiMaxTokens: number;
/** Empty string when capabilityKind="chat". */
piperUrl: string;
advertRef: AdvertRef;
@@ -96,6 +128,13 @@ export interface SupplierConfig {
* 300_000 (5 min).
*/
chatIdleTimeoutMs: number;
/**
* Concurrent chat-session slots (capabilityKind="chat-session" only).
* Default 1 = legacy single-slot behavior.
*/
maxChatSessions: number;
/** Chain settlement mode for chat sessions. Default "full". */
chatSettleMode: ChatSettleMode;
/**
* When true, the supplier boots a LiveOgmiosProvider (real submitTx/awaitTx).
* Default false → ReadOnlyOgmiosProvider (safe; no chain writes).
@@ -230,6 +269,49 @@ export function loadConfig(env: Record<string, string | undefined>): SupplierCon
const openaiApiKey = env.OPENAI_API_KEY ?? "";
// OPENAI_REASONING — "off"/"false"/"0"/"no" disables OpenRouter reasoning
// ("thinking") by sending reasoning:{enabled:false} on every upstream call.
// Unset/any other value leaves the model's own default untouched (no param
// sent). Only valid for OpenRouter-backed suppliers — do NOT set for
// ChatMock/DeepSeek-direct, which reject the param.
const reasoningRaw = (env.OPENAI_REASONING ?? "").trim().toLowerCase();
const openaiReasoningDisabled =
reasoningRaw === "off" ||
reasoningRaw === "false" ||
reasoningRaw === "0" ||
reasoningRaw === "no";
// HuggingFace router guard. The `reasoning:{enabled:false}` param is
// OpenRouter-specific; the HF Inference Providers router
// (router.huggingface.co) rejects it with HTTP 400 on EVERY request. This is
// the easy footgun when copying an OpenRouter env to the HF preset, so fail
// fast at boot with a clear message instead of 400-ing every job.
// See docs/HUGGINGFACE_ROUTER_SETUP.md.
if (
llmBackend === "openai" &&
openaiReasoningDisabled &&
hostnameOf(openaiBaseUrl) === "router.huggingface.co"
) {
throw new Error(
"loadConfig: OPENAI_REASONING must be unset for the HuggingFace router " +
"(OPENAI_BASE_URL=https://router.huggingface.co) — it rejects " +
"reasoning:{enabled:false} with HTTP 400. Remove OPENAI_REASONING.",
);
}
// OPENAI_MAX_TOKENS — optional hard ceiling forwarded as `max_tokens`. Unset
// or non-positive omits it (provider grants max available = context − prompt).
// Keep it below the model context (kimi 262144) or the provider 400s when
// prompt+max_tokens exceeds context.
const maxTokStr = env.OPENAI_MAX_TOKENS;
let openaiMaxTokens = 0;
if (maxTokStr !== undefined && maxTokStr !== "") {
if (!POS_INT_RE.test(maxTokStr)) {
throw new Error("loadConfig: OPENAI_MAX_TOKENS must be a positive integer");
}
openaiMaxTokens = Number(maxTokStr);
}
// CHAT_IDLE_TIMEOUT_MS — idle auto-end window for chat sessions (default 5 min).
const chatIdleStr = env.CHAT_IDLE_TIMEOUT_MS;
let chatIdleTimeoutMs = 300_000;
@@ -240,6 +322,32 @@ export function loadConfig(env: Record<string, string | undefined>): SupplierCon
chatIdleTimeoutMs = Number(chatIdleStr);
}
// MAX_CHAT_SESSIONS — concurrent chat-session slots (default 1 = legacy
// single-slot). Only meaningful for capabilityKind="chat-session".
const maxSessionsStr = env.MAX_CHAT_SESSIONS;
let maxChatSessions = 1;
if (maxSessionsStr !== undefined && maxSessionsStr !== "") {
if (!POS_INT_RE.test(maxSessionsStr)) {
throw new Error("loadConfig: MAX_CHAT_SESSIONS must be a positive integer");
}
maxChatSessions = Number(maxSessionsStr);
}
if (maxChatSessions > 1 && capabilityKind !== "chat-session") {
throw new Error('loadConfig: MAX_CHAT_SESSIONS > 1 requires CAPABILITY_KIND="chat-session"');
}
// CHAT_SETTLE_MODE — "full" (default; Claim at start, Submit receipt at end)
// or "ticket" (the buyer's escrow post is the only chain op; the supplier
// never Claims/Submits and the buyer reclaims the escrow after deliver_by).
const settleModeStr = env.CHAT_SETTLE_MODE ?? "full";
if (settleModeStr !== "full" && settleModeStr !== "ticket") {
throw new Error('loadConfig: CHAT_SETTLE_MODE must be "full" or "ticket"');
}
if (settleModeStr === "ticket" && capabilityKind !== "chat-session") {
throw new Error('loadConfig: CHAT_SETTLE_MODE="ticket" requires CAPABILITY_KIND="chat-session"');
}
const chatSettleMode: ChatSettleMode = settleModeStr;
const piperTimeoutStr = env.PIPER_TIMEOUT_MS;
let piperTimeoutMs = 120_000;
if (piperTimeoutStr !== undefined && piperTimeoutStr !== "") {
@@ -270,6 +378,8 @@ export function loadConfig(env: Record<string, string | undefined>): SupplierCon
ollamaUrl,
openaiBaseUrl,
openaiApiKey,
openaiReasoningDisabled,
openaiMaxTokens,
piperUrl,
advertRef,
networkId,
@@ -280,6 +390,8 @@ export function loadConfig(env: Record<string, string | undefined>): SupplierCon
capabilityKind,
llmBackend,
chatIdleTimeoutMs,
maxChatSessions,
chatSettleMode,
liveChain,
walletHealthIntervalMs,
};

View File

@@ -42,7 +42,8 @@ async function main(): Promise<void> {
// M1-F-2: provider selection is driven by cfg.liveChain (LIVE_CHAIN=1).
// ReadOnly is the safe default; live mode is explicit opt-in.
const chain = buildChainProvider(cfg);
const state = new SupplierState();
// 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();
const deps: SupplierDeps = { chain, state, config: cfg, supplierKey, jobs };

View File

@@ -92,6 +92,8 @@ export async function runChatJob(params: RunChatJobParams): Promise<void> {
messages: requestBody.messages,
timeoutMs: deps.config.openaiTimeoutMs,
apiKey: deps.config.openaiApiKey,
maxTokens: deps.config.openaiMaxTokens,
disableReasoning: deps.config.openaiReasoningDisabled,
});
} else {
inference = await ollama.callOllama({
@@ -141,17 +143,28 @@ export async function runChatJob(params: RunChatJobParams): Promise<void> {
const signed = signReceipt(receipt, deps.supplierKey.privateKeyHex);
const resultHash = receiptResultHash(signed);
// ── 4. Build + submit Submit tx ─────────────────────────────────────
let buildResult: { txCborHex: string; expectedTxHash: string };
try {
buildResult = await buildSubmitTx({
chain: deps.chain,
supplierKey: deps.supplierKey,
escrowRef: claimedRef,
receiptHash: resultHash,
});
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
// ── 4+5. Submit tx: build + confirm as ONE wallet critical section ──
const submitOutcome = await deps.state.walletMutex.run(async () => {
let built: { txCborHex: string; expectedTxHash: string };
try {
built = await buildSubmitTx({
chain: deps.chain,
supplierKey: deps.supplierKey,
escrowRef: claimedRef,
receiptHash: resultHash,
});
} catch (err) {
return { kind: "build_failed" as const, err };
}
try {
await deps.chain.awaitTx(built.expectedTxHash, 60_000);
} catch (err) {
return { kind: "await_failed" as const, err };
}
return { kind: "ok" as const };
});
if (submitOutcome.kind === "build_failed") {
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}`,
);
@@ -162,12 +175,8 @@ export async function runChatJob(params: RunChatJobParams): Promise<void> {
});
return;
}
// ── 5. Submit confirmation (awaitTx, 60s budget) ───────────────────
try {
await deps.chain.awaitTx(buildResult.expectedTxHash, 60_000);
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
if (submitOutcome.kind === "await_failed") {
const msg = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err);
console.warn(
`[job_failed] jobId=${jobId} reason=submit_timeout httpStatus=502 msg=${msg}`,
);
@@ -199,9 +208,9 @@ export async function runChatJob(params: RunChatJobParams): Promise<void> {
};
deps.jobs.complete(jobId, payload);
} finally {
// ── 7. Always release the lock ──────────────────────────────────────
// ── 7. Always release this job's slot ───────────────────────────────
try {
deps.state.release();
deps.state.release(escrowRef);
} catch {
// never let a lock-release error escape the runner
}
@@ -290,17 +299,28 @@ export async function runTtsJob(params: RunTtsJobParams): Promise<void> {
const signed = signReceipt(receipt, deps.supplierKey.privateKeyHex);
const resultHash = receiptResultHash(signed);
// ── 4. Build + submit Submit tx ───────────────────────────────────
let buildResult: { txCborHex: string; expectedTxHash: string };
try {
buildResult = await buildSubmitTx({
chain: deps.chain,
supplierKey: deps.supplierKey,
escrowRef: claimedRef,
receiptHash: resultHash,
});
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
// ── 4+5. Submit tx: build + confirm as ONE wallet critical section ──
const submitOutcome = await deps.state.walletMutex.run(async () => {
let built: { txCborHex: string; expectedTxHash: string };
try {
built = await buildSubmitTx({
chain: deps.chain,
supplierKey: deps.supplierKey,
escrowRef: claimedRef,
receiptHash: resultHash,
});
} catch (err) {
return { kind: "build_failed" as const, err };
}
try {
await deps.chain.awaitTx(built.expectedTxHash, 60_000);
} catch (err) {
return { kind: "await_failed" as const, err };
}
return { kind: "ok" as const };
});
if (submitOutcome.kind === "build_failed") {
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}`,
);
@@ -311,12 +331,8 @@ export async function runTtsJob(params: RunTtsJobParams): Promise<void> {
});
return;
}
// ── 5. Submit confirmation ────────────────────────────────────────
try {
await deps.chain.awaitTx(buildResult.expectedTxHash, 60_000);
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
if (submitOutcome.kind === "await_failed") {
const msg = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err);
console.warn(
`[job_failed] jobId=${jobId} reason=submit_timeout httpStatus=502 msg=${msg}`,
);
@@ -342,7 +358,7 @@ export async function runTtsJob(params: RunTtsJobParams): Promise<void> {
deps.jobs.complete(jobId, payload);
} finally {
try {
deps.state.release();
deps.state.release(escrowRef);
} catch {
/* never let lock-release escape */
}

View File

@@ -26,19 +26,40 @@
* - Empty-string content is treated as malformed (matches ollama.ts).
*/
import type { ChatMessage } from "@marketplace/shared/tx";
import type { ChatMessage, ToolCall } from "@marketplace/shared/tx";
export interface CallOpenAiParams {
baseUrl: string;
model: string;
messages: ChatMessage[];
timeoutMs: number;
/**
* OpenAI-compatible `tools` / `tool_choice` passthrough (chat-session
* capability only). Forwarded verbatim when present; the upstream provider
* validates the schema.
*/
tools?: unknown[];
toolChoice?: unknown;
/**
* Optional bearer token. When non-empty, an `authorization: Bearer <apiKey>`
* header is sent. Omit (or pass "") for endpoints that don't require auth,
* e.g. the ChatMock localhost proxy.
*/
apiKey?: string;
/**
* Forwarded as `max_tokens` when a positive number; omitted otherwise. The
* caller is responsible for keeping prompt+max_tokens within the model's
* context window (some providers 400 when it exceeds context).
*/
maxTokens?: number;
/**
* When true, sends `reasoning: { enabled: false }` to disable OpenRouter
* "thinking" tokens — faster/cheaper straight answers, and avoids reasoning
* starving the completion budget into a length-truncated (empty) answer.
* Only valid for OpenRouter endpoints; leave false for ChatMock / direct
* DeepSeek, which don't accept the param.
*/
disableReasoning?: boolean;
}
export interface OpenAiResult {
@@ -46,6 +67,10 @@ export interface OpenAiResult {
prompt_tokens: number;
completion_tokens: number;
wallclock_ms: number;
/** Present when the model requested tool calls (streaming path only). */
tool_calls?: ToolCall[];
/** Upstream finish_reason when reported (e.g. "stop", "tool_calls"). */
finish_reason?: string;
}
export type OpenAiErrorReason = "openai_failure" | "openai_timeout" | "openai_malformed";
@@ -68,9 +93,12 @@ function isAbortError(err: unknown): boolean {
}
export async function callOpenAi(params: CallOpenAiParams): Promise<OpenAiResult> {
const { baseUrl, model, messages, timeoutMs, apiKey } = params;
const { baseUrl, model, messages, timeoutMs, apiKey, maxTokens, disableReasoning } = params;
const url = `${baseUrl}/v1/chat/completions`;
const body = JSON.stringify({ model, messages, stream: false });
const payload: Record<string, unknown> = { model, messages, stream: false };
if (typeof maxTokens === "number" && maxTokens > 0) payload.max_tokens = maxTokens;
if (disableReasoning) payload.reasoning = { enabled: false };
const body = JSON.stringify(payload);
const headers: Record<string, string> = { "content-type": "application/json" };
if (apiKey) {
@@ -187,14 +215,21 @@ export async function callOpenAiStream(
params: CallOpenAiParams,
onToken: (delta: string) => void,
): Promise<OpenAiResult> {
const { baseUrl, model, messages, timeoutMs, apiKey } = params;
const { baseUrl, model, messages, timeoutMs, apiKey, maxTokens, disableReasoning, tools, toolChoice } = params;
const url = `${baseUrl}/v1/chat/completions`;
const body = JSON.stringify({
const payload: Record<string, unknown> = {
model,
messages,
stream: true,
stream_options: { include_usage: true },
});
};
if (typeof maxTokens === "number" && maxTokens > 0) payload.max_tokens = maxTokens;
if (disableReasoning) payload.reasoning = { enabled: false };
if (Array.isArray(tools) && tools.length > 0) {
payload.tools = tools;
if (toolChoice !== undefined) payload.tool_choice = toolChoice;
}
const body = JSON.stringify(payload);
const headers: Record<string, string> = {
"content-type": "application/json",
@@ -248,7 +283,28 @@ export async function callOpenAiStream(
let accumulated = "";
let promptTokens = 0;
let completionTokens = 0;
let finishReason: string | undefined;
let buffer = "";
// delta.tool_calls arrives as indexed fragments (id/name once, arguments in
// pieces) — accumulate per index, emit complete calls after the stream ends.
const toolCallParts = new Map<number, { id: string; name: string; arguments: string }>();
const handleToolCallDeltas = (raw: unknown): void => {
if (!Array.isArray(raw)) return;
for (const item of raw) {
if (!item || typeof item !== "object") continue;
const frag = item as Record<string, unknown>;
const index = typeof frag.index === "number" ? frag.index : 0;
const part = toolCallParts.get(index) ?? { id: "", name: "", arguments: "" };
if (typeof frag.id === "string" && frag.id.length > 0) part.id = frag.id;
const fn = frag.function as Record<string, unknown> | undefined;
if (fn && typeof fn === "object") {
if (typeof fn.name === "string" && fn.name.length > 0) part.name = fn.name;
if (typeof fn.arguments === "string") part.arguments += fn.arguments;
}
toolCallParts.set(index, part);
}
};
const handleData = (payload: string): void => {
const trimmed = payload.trim();
@@ -264,14 +320,15 @@ export async function callOpenAiStream(
const obj = parsed as Record<string, unknown>;
const choices = obj.choices;
if (Array.isArray(choices) && choices.length > 0) {
const delta = (choices[0] as Record<string, unknown>)?.delta as
| Record<string, unknown>
| undefined;
const choice = choices[0] as Record<string, unknown>;
const delta = choice?.delta as Record<string, unknown> | undefined;
const content = delta?.content;
if (typeof content === "string" && content.length > 0) {
accumulated += content;
onToken(content);
}
handleToolCallDeltas(delta?.tool_calls);
if (typeof choice?.finish_reason === "string") finishReason = choice.finish_reason;
}
const usageRaw = obj.usage;
if (usageRaw && typeof usageRaw === "object") {
@@ -316,14 +373,24 @@ export async function callOpenAiStream(
clearTimeout(timer);
}
if (accumulated.length === 0) {
const toolCalls: ToolCall[] = [...toolCallParts.entries()]
.sort(([a], [b]) => a - b)
.filter(([, p]) => p.id.length > 0 && p.name.length > 0)
.map(([, p]) => ({ id: p.id, type: "function" as const, function: { name: p.name, arguments: p.arguments } }));
// A pure tool-call turn legitimately has no content; only an empty stream
// with neither content nor tool calls is malformed.
if (accumulated.length === 0 && toolCalls.length === 0) {
throw new OpenAiError("openai_malformed", "OpenAI stream produced no content");
}
return {
const result: OpenAiResult = {
content: accumulated,
prompt_tokens: promptTokens,
completion_tokens: completionTokens,
wallclock_ms: Date.now() - startedAt,
};
if (toolCalls.length > 0) result.tool_calls = toolCalls;
if (finishReason !== undefined) result.finish_reason = finishReason;
return result;
}

View File

@@ -37,8 +37,14 @@ import { createHash } from "crypto";
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 { WalletKey } from "@marketplace/shared/tx";
import { buildClaimTx, chatSessionPromptHash, mockSlotToWallclockMs, detectCborBackend } from "@marketplace/shared/tx";
import type { ChatMessage, WalletKey } from "@marketplace/shared/tx";
import {
buildClaimTx,
chatSessionPromptHash,
mockSlotToWallclockMs,
detectCborBackend,
normalizeChatMessage,
} from "@marketplace/shared/tx";
import type { SupplierState } from "./state.js";
import type { SupplierConfig } from "./config.js";
@@ -141,6 +147,8 @@ function makeStatusHandler(deps: ResolvedDeps) {
const payload: Record<string, unknown> = {
status: snap.status,
last_seen: snap.lastSeenIso,
active_sessions: snap.activeSessions,
max_sessions: snap.maxSessions,
};
if (snap.status === "working" && snap.currentEscrowRef) {
payload.current_escrow_ref = snap.currentEscrowRef;
@@ -184,6 +192,9 @@ async function fetchActiveAdvert(deps: ResolvedDeps): Promise<AdvertResult> {
function makeChatHandler(deps: ResolvedDeps) {
return async (req: Request, res: Response, next: NextFunction) => {
// 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;
try {
// ── 1. Header validation ──────────────────────────────────────────
const headerVal = req.header("X-Escrow-Ref");
@@ -297,23 +308,37 @@ function makeChatHandler(deps: ResolvedDeps) {
`now ${nowMs} >= deliver_by ${escrowDatum.deliver_by}`);
}
// ── 8. Acquire single-slot lock ──────────────────────────────────
// ── 8. Acquire session slot ──────────────────────────────────────
if (!deps.state.tryAcquire(escrowRefStr)) {
return jsonError(res, 409, "supplier_busy", "supplier is already working another job");
}
// From here on, every error path MUST release the lock until the
// runner takes ownership of the lock.
acquiredRef = escrowRefStr;
// From here on, every error path MUST release the slot until the
// runner takes ownership of it.
// ── 9. Submit Claim tx ───────────────────────────────────────────
let claimResult;
try {
claimResult = await buildClaimTx({
chain: deps.chain,
supplierKey: deps.supplierKey,
escrowRef,
});
} catch (err) {
deps.state.release();
// ── 9+10. Claim tx: build + confirm as ONE wallet critical section ─
// (splitting them would let another wallet spend coin-select mid-flight)
const claimOutcome = await deps.state.walletMutex.run(async () => {
let built;
try {
built = await buildClaimTx({
chain: deps.chain,
supplierKey: deps.supplierKey,
escrowRef,
});
} catch (err) {
return { kind: "build_failed" as const, err };
}
try {
await deps.chain.awaitTx(built.expectedTxHash, 60_000);
} catch (err) {
return { kind: "await_failed" as const, err };
}
return { kind: "ok" as const, built };
});
if (claimOutcome.kind === "build_failed") {
deps.state.release(escrowRefStr);
acquiredRef = null;
// Backstop the periodic wallet-health ticker: most Claim build failures
// we see in practice are wallet-shape problems (fragmentation, dust).
// Fire-and-forget a debounced consolidate so the NEXT buyer retry
@@ -324,17 +349,15 @@ function makeChatHandler(deps: ResolvedDeps) {
supplierKey: deps.supplierKey,
});
return jsonError(res, 503, "chain_submit_failed",
`Claim tx submit failed: ${(err as Error).message}`);
`Claim tx submit failed: ${(claimOutcome.err as Error).message}`);
}
// ── 10. Await Claim confirmation ─────────────────────────────────
try {
await deps.chain.awaitTx(claimResult.expectedTxHash, 60_000);
} catch (err) {
deps.state.release();
if (claimOutcome.kind === "await_failed") {
deps.state.release(escrowRefStr);
acquiredRef = null;
return jsonError(res, 504, "claim_timeout",
`Claim awaitTx failed: ${(err as Error).message}`);
`Claim awaitTx failed: ${(claimOutcome.err as Error).message}`);
}
const claimResult = claimOutcome.built;
// ── 11. Continuing-output ref + create job + fire-and-forget ─────
const claimedRef: OutputReference = {
@@ -343,7 +366,8 @@ function makeChatHandler(deps: ResolvedDeps) {
};
const jobId = deps.jobs.create(escrowRefStr);
// Lock release now happens inside runChatJob's finally.
// Slot release now happens inside runChatJob's finally.
acquiredRef = null;
void runChatJob({
deps: {
chain: deps.chain,
@@ -367,7 +391,9 @@ function makeChatHandler(deps: ResolvedDeps) {
escrow_ref: escrowRefStr,
});
} catch (err) {
try { deps.state.release(); } catch { /* ignore */ }
if (acquiredRef) {
try { deps.state.release(acquiredRef); } catch { /* ignore */ }
}
next(err);
return;
}
@@ -441,6 +467,8 @@ const ALLOWED_TTS_FORMATS = new Set(["mp3", "wav", "opus", "aac", "flac"]);
function makeTtsHandler(deps: ResolvedDeps) {
return async (req: Request, res: Response, next: NextFunction) => {
// Slot held by THIS request (null until acquired / after hand-off).
let acquiredRef: string | null = null;
try {
// ── 1. Header ───────────────────────────────────────────────────
const headerVal = req.header("X-Escrow-Ref");
@@ -544,39 +572,53 @@ function makeTtsHandler(deps: ResolvedDeps) {
`now ${nowMs} >= deliver_by ${escrowDatum.deliver_by}`);
}
// ── 8. Lock ─────────────────────────────────────────────────────
// ── 8. Acquire session slot ─────────────────────────────────────
if (!deps.state.tryAcquire(escrowRefStr)) {
return jsonError(res, 409, "supplier_busy", "supplier is already working another job");
}
acquiredRef = escrowRefStr;
// ── 9. Claim tx ─────────────────────────────────────────────────
let claimResult;
try {
claimResult = await buildClaimTx({
chain: deps.chain,
supplierKey: deps.supplierKey,
escrowRef,
});
} catch (err) {
deps.state.release();
// ── 9. Claim tx: build + confirm as ONE wallet critical section ──
const claimOutcome = await deps.state.walletMutex.run(async () => {
let built;
try {
built = await buildClaimTx({
chain: deps.chain,
supplierKey: deps.supplierKey,
escrowRef,
});
} catch (err) {
return { kind: "build_failed" as const, err };
}
try {
await deps.chain.awaitTx(built.expectedTxHash, 60_000);
} catch (err) {
return { kind: "await_failed" as const, err };
}
return { kind: "ok" as const, built };
});
if (claimOutcome.kind === "build_failed") {
deps.state.release(escrowRefStr);
acquiredRef = null;
return jsonError(res, 503, "chain_submit_failed",
`Claim tx submit failed: ${(err as Error).message}`);
`Claim tx submit failed: ${(claimOutcome.err as Error).message}`);
}
try {
await deps.chain.awaitTx(claimResult.expectedTxHash, 60_000);
} catch (err) {
deps.state.release();
if (claimOutcome.kind === "await_failed") {
deps.state.release(escrowRefStr);
acquiredRef = null;
return jsonError(res, 504, "claim_timeout",
`Claim awaitTx failed: ${(err as Error).message}`);
`Claim awaitTx failed: ${(claimOutcome.err as Error).message}`);
}
// ── 10. Spawn runner + 202 ──────────────────────────────────────
const claimedRef: OutputReference = {
txHash: claimResult.expectedTxHash,
txHash: claimOutcome.built.expectedTxHash,
index: 0,
};
const jobId = deps.jobs.create(escrowRefStr);
// Slot release now happens inside runTtsJob's finally.
acquiredRef = null;
void runTtsJob({
deps: {
chain: deps.chain,
@@ -599,7 +641,9 @@ function makeTtsHandler(deps: ResolvedDeps) {
escrow_ref: escrowRefStr,
});
} catch (err) {
try { deps.state.release(); } catch { /* ignore */ }
if (acquiredRef) {
try { deps.state.release(acquiredRef); } catch { /* ignore */ }
}
next(err);
return;
}
@@ -699,6 +743,9 @@ function makeChatSessionHandlers(deps: ResolvedDeps) {
}
const start = async (req: Request, res: Response, next: NextFunction) => {
// Slot held by THIS request (null until acquired / after the session
// record takes ownership) — the catch-all only releases its own.
let acquiredRef: string | null = null;
try {
// ── 1. Header ───────────────────────────────────────────────────
const headerVal = req.header("X-Escrow-Ref");
@@ -711,6 +758,15 @@ function makeChatSessionHandlers(deps: ResolvedDeps) {
}
const escrowRefStr = `${escrowRef.txHash}#${escrowRef.index}`;
// ── 1.5 Duplicate-session guard ───────────────────────────────────
// Ended records are retained, so this also blocks replaying a used
// ticket-mode escrow (which stays Open on-chain — the state check below
// wouldn't catch it). In full mode the on-chain Claim already prevents
// reuse; this check just fails faster.
if (deps.chatSessions.get(escrowRefStr)) {
return jsonError(res, 409, "session_exists", "a chat session already exists for this escrow ref");
}
// ── 2. Body (session_nonce; no messages) ─────────────────────────
const body = (req.body ?? {}) as ChatStartBody;
const sessionNonce = typeof body.session_nonce === "string" ? body.session_nonce : "";
@@ -774,36 +830,68 @@ function makeChatSessionHandlers(deps: ResolvedDeps) {
return jsonError(res, 408, "past_deliver_by", `now ${nowMs} >= deliver_by ${escrowDatum.deliver_by}`);
}
// ── 8. Single-slot lock ──────────────────────────────────────────
// ── 8. Acquire session slot ──────────────────────────────────────
if (!deps.state.tryAcquire(escrowRefStr)) {
return jsonError(res, 409, "supplier_busy", "supplier is already working another chat");
}
acquiredRef = escrowRefStr;
// ── 9. Claim (Open → Claimed) — hold the lock on success ─────────
let claimResult;
try {
claimResult = await buildClaimTx({ chain: deps.chain, supplierKey: deps.supplierKey, escrowRef });
} catch (err) {
deps.state.release();
triggerOnFailureConsolidate({ chain: deps.chain, state: deps.state, supplierKey: deps.supplierKey });
return jsonError(res, 503, "chain_submit_failed", `Claim tx submit failed: ${(err as Error).message}`);
if (deps.config.chatSettleMode === "ticket") {
// ── 9-ticket. No Claim: the verified Open escrow IS the entry
// ticket. No hard-cap timer either — there is no Submit deadline, so
// the session may outlive deliver_by (the buyer reclaims the escrow
// independently). The idle timer is the only lifetime bound.
const record = deps.chatSessions.create({
escrowRef: escrowRefStr,
settleMode: "ticket",
advert,
escrowDatum,
});
acquiredRef = null; // record owns the slot; endChatSession releases it
armIdleTimer(record);
return res.status(200).json({ status: "ticket", escrow_ref: escrowRefStr, settle_mode: "ticket" });
}
try {
await deps.chain.awaitTx(claimResult.expectedTxHash, 60_000);
} catch (err) {
deps.state.release();
return jsonError(res, 504, "claim_timeout", `Claim awaitTx failed: ${(err as Error).message}`);
// ── 9-full. Claim (Open → Claimed): build + confirm as ONE wallet
// critical section — hold the slot on success through Submit at end.
const claimOutcome = await deps.state.walletMutex.run(async () => {
let built;
try {
built = await buildClaimTx({ chain: deps.chain, supplierKey: deps.supplierKey, escrowRef });
} catch (err) {
return { kind: "build_failed" as const, err };
}
try {
await deps.chain.awaitTx(built.expectedTxHash, 60_000);
} catch (err) {
return { kind: "await_failed" as const, err };
}
return { kind: "ok" as const, built };
});
if (claimOutcome.kind === "build_failed") {
deps.state.release(escrowRefStr);
acquiredRef = null;
triggerOnFailureConsolidate({ chain: deps.chain, state: deps.state, supplierKey: deps.supplierKey });
return jsonError(res, 503, "chain_submit_failed", `Claim tx submit failed: ${(claimOutcome.err as Error).message}`);
}
if (claimOutcome.kind === "await_failed") {
deps.state.release(escrowRefStr);
acquiredRef = null;
return jsonError(res, 504, "claim_timeout", `Claim awaitTx failed: ${(claimOutcome.err as Error).message}`);
}
// ── 10. Create session + arm watchdog ────────────────────────────
const claimedRef: OutputReference = { txHash: claimResult.expectedTxHash, index: 0 };
const record = deps.chatSessions.create({ escrowRef: escrowRefStr, claimedRef, advert, escrowDatum });
const claimedRef: OutputReference = { txHash: claimOutcome.built.expectedTxHash, index: 0 };
const record = deps.chatSessions.create({ escrowRef: escrowRefStr, claimedRef, settleMode: "full", advert, escrowDatum });
acquiredRef = null; // record owns the slot; endChatSession releases it
armIdleTimer(record);
armHardCapTimer(record, nowMs);
return res.status(200).json({ status: "claimed", escrow_ref: escrowRefStr });
} catch (err) {
try { deps.state.release(); } catch { /* ignore */ }
if (acquiredRef) {
try { deps.state.release(acquiredRef); } catch { /* ignore */ }
}
next(err);
return;
}
@@ -825,16 +913,47 @@ function makeChatSessionHandlers(deps: ResolvedDeps) {
return jsonError(res, 404, "chat_session_not_found",
`no active chat session for ${escrowRefStr}`);
}
const body = (req.body ?? {}) as { content?: unknown };
const content = typeof body.content === "string" ? body.content : "";
if (content.length === 0) {
return jsonError(res, 400, "content_required", "body.content must be a non-empty string");
// Two body shapes: legacy {content} (single user turn) and
// {messages, tools?, tool_choice?} (multi-message delta — e.g. tool
// results — appended to the transcript as one turn).
const body = (req.body ?? {}) as {
content?: unknown;
messages?: unknown;
tools?: unknown;
tool_choice?: unknown;
};
let delta: ChatMessage[];
if (Array.isArray(body.messages)) {
const normalized: ChatMessage[] = [];
for (const raw of body.messages) {
const msg = normalizeChatMessage(raw);
if (!msg) {
return jsonError(res, 400, "invalid_messages",
"body.messages must be OpenAI-shaped {role, content, tool_calls?, tool_call_id?} objects");
}
normalized.push(msg);
}
const last = normalized[normalized.length - 1];
if (!last || (last.role !== "user" && last.role !== "tool")) {
return jsonError(res, 400, "invalid_messages",
"body.messages must end with a user or tool message");
}
delta = normalized;
} else {
const content = typeof body.content === "string" ? body.content : "";
if (content.length === 0) {
return jsonError(res, 400, "content_required", "body.content must be a non-empty string");
}
delta = [{ role: "user", content }];
}
const tools = Array.isArray(body.tools) && body.tools.length > 0 ? body.tools : undefined;
const toolChoice = tools !== undefined ? body.tool_choice : undefined;
// Pause the idle timer while a turn is in flight; re-arm when it ends so
// a long generation never trips the auto-end mid-stream.
if (record.idleTimer) { clearTimeout(record.idleTimer); record.idleTimer = undefined; }
deps.chatSessions.appendUser(escrowRefStr, content);
const transcriptLenBefore = record.transcript.length;
deps.chatSessions.appendMessages(escrowRefStr, delta);
res.status(200);
res.setHeader("Content-Type", "text/event-stream");
@@ -856,15 +975,36 @@ function makeChatSessionHandlers(deps: ResolvedDeps) {
messages: record.transcript,
timeoutMs: deps.config.openaiTimeoutMs,
apiKey: deps.config.openaiApiKey,
maxTokens: deps.config.openaiMaxTokens,
disableReasoning: deps.config.openaiReasoningDisabled,
tools,
toolChoice,
},
(delta) => sse({ type: "token", value: delta }),
(tok) => sse({ type: "token", value: tok }),
);
deps.chatSessions.appendAssistant(escrowRefStr, result.content, {
// Build the assistant message once and send it VERBATIM in the done
// frame: the gateway mirrors this object so both transcripts stay
// hash-identical for the receipt (field presence matters).
const assistantMsg: ChatMessage = { role: "assistant", content: result.content };
if (result.tool_calls && result.tool_calls.length > 0) assistantMsg.tool_calls = result.tool_calls;
deps.chatSessions.appendAssistant(escrowRefStr, assistantMsg, {
prompt_tokens: result.prompt_tokens,
completion_tokens: result.completion_tokens,
});
sse({ type: "done" });
const finishReason =
result.finish_reason === "tool_calls" || (result.tool_calls?.length ?? 0) > 0
? "tool_calls"
: "stop";
sse({
type: "done",
message: assistantMsg,
finish_reason: finishReason,
usage: { prompt_tokens: result.prompt_tokens, completion_tokens: result.completion_tokens },
});
} catch (err) {
// Failed turn: the gateway mirror never saw this delta, so drop it here
// too — otherwise a retry re-sends it and the transcripts diverge.
deps.chatSessions.truncateTranscript(escrowRefStr, transcriptLenBefore);
const errMsg = err instanceof Error ? err.message : String(err);
sse({ type: "error", message: errMsg });
} finally {
@@ -896,6 +1036,11 @@ function makeChatSessionHandlers(deps: ResolvedDeps) {
if (!record) {
return jsonError(res, 404, "chat_session_not_found", `no chat session for ${escrowRefStr}`);
}
if (record.settleMode === "ticket") {
// No receipt/Submit in ticket mode — the buyer reclaims the escrow.
// Idempotent: a re-end returns the same shape.
return res.status(200).json({ status: "ended", escrow_ref: escrowRefStr, settle_mode: "ticket" });
}
if (record.endFailure) {
return jsonError(res, 502, record.endFailure.reason, record.endFailure.message);
}

View File

@@ -1,35 +1,68 @@
/**
* supplier/src/state.ts — In-process single-slot concurrency lock.
* supplier/src/state.ts — In-process session-slot admission + wallet mutex.
*
* SupplierState tracks whether the node is free, working, or offline.
* SupplierState admits up to `capacity` concurrent sessions (capacity 1 for
* one-shot chat/tts kinds; MAX_CHAT_SESSIONS for chat-session kind). Status is
* free-until-full: "working" only when every slot is taken, so the indexer
* status poller and the buyer SDK's pre-check keep their existing semantics.
* Cross-process coordination is M1-D; this is purely in-memory.
*
* Methods:
* tryAcquire(escrowRef): boolean — atomic test-and-set; returns false unless free
* release() — transition working → free (no-op from free/offline)
* markOffline() — transition any state → offline; idempotent
* snapshot() — current status snapshot (defensive copy)
* walletMutex serializes every wallet-spending chain op (Claim/Submit/
* consolidate): the wallet holds one working UTxO + one collateral UTxO, so
* concurrent txs would double-spend regardless of how many session slots
* exist. Hold it across build+awaitTx as one critical section.
*
* Invariants:
* - currentEscrowRef is set IFF status === "working"
* - markOffline() clears currentEscrowRef
* - tryAcquire never overwrites a held lock
* - a ref can be held at most once (duplicate tryAcquire returns false)
* - release(ref) frees exactly that ref's slot
* - markOffline() clears all slots and refuses new acquires
* - snapshot.currentEscrowRef is present IFF status === "working" (legacy
* consumers; holds the oldest active ref)
* - lastSeenIso updates on every state-changing operation
*/
export type SupplierStatus = "free" | "working" | "offline";
/** Tiny promise-chain mutex: run(fn) executes fns strictly one at a time. */
export class Mutex {
private tail: Promise<unknown> = Promise.resolve();
run<T>(fn: () => Promise<T>): Promise<T> {
const result = this.tail.then(fn, fn);
// Keep the chain alive even if fn rejects; swallow here so the next
// run() isn't poisoned by a prior rejection.
this.tail = result.then(
() => undefined,
() => undefined,
);
return result;
}
}
export interface SupplierSnapshot {
status: SupplierStatus;
/** Oldest active ref; present ONLY when status === "working" (back-compat). */
currentEscrowRef?: string;
activeEscrowRefs: string[];
activeSessions: number;
maxSessions: number;
lastSeenIso: string;
}
export class SupplierState {
private status: SupplierStatus = "free";
private currentEscrowRef: string | undefined = undefined;
/** Serializes ALL wallet-spending chain ops (Claim/Submit/consolidate). */
readonly walletMutex = new Mutex();
private readonly active = new Set<string>(); // insertion-ordered
private offline = false;
private lastSeenIso: string = new Date().toISOString();
constructor(readonly capacity: number = 1) {
if (!Number.isInteger(capacity) || capacity < 1) {
throw new Error(`SupplierState: capacity must be a positive integer, got ${capacity}`);
}
}
private touch(): void {
// ISO 8601 with millisecond resolution. Coarse clocks may produce equal
// values in tight succession — tests use >= rather than strict greater.
@@ -37,40 +70,45 @@ export class SupplierState {
}
tryAcquire(escrowRef: string): boolean {
if (this.status !== "free") return false;
this.status = "working";
this.currentEscrowRef = escrowRef;
if (this.offline) return false;
if (this.active.has(escrowRef)) return false;
if (this.active.size >= this.capacity) return false;
this.active.add(escrowRef);
this.touch();
return true;
}
release(): void {
if (this.status === "working") {
this.status = "free";
this.currentEscrowRef = undefined;
release(escrowRef: string): void {
if (this.active.delete(escrowRef)) {
this.touch();
}
}
markOffline(): void {
this.status = "offline";
this.currentEscrowRef = undefined;
this.offline = true;
this.active.clear();
this.touch();
}
snapshot(): SupplierSnapshot {
// Omit currentEscrowRef key entirely when not working so
// ('currentEscrowRef' in snap) === false (matches test expectation).
if (this.status === "working" && this.currentEscrowRef !== undefined) {
return {
status: this.status,
currentEscrowRef: this.currentEscrowRef,
lastSeenIso: this.lastSeenIso,
};
}
return {
status: this.status,
const refs = [...this.active];
const status: SupplierStatus = this.offline
? "offline"
: refs.length >= this.capacity
? "working"
: "free";
const snap: SupplierSnapshot = {
status,
activeEscrowRefs: refs,
activeSessions: refs.length,
maxSessions: this.capacity,
lastSeenIso: this.lastSeenIso,
};
// Omit currentEscrowRef key entirely when not working so
// ('currentEscrowRef' in snap) === false (matches legacy consumers).
if (status === "working" && refs.length > 0) {
snap.currentEscrowRef = refs[0];
}
return snap;
}
}

View File

@@ -16,9 +16,10 @@
* multiple consolidates. Returns 503 to the buyer for THIS request; the
* next buyer retry (~30-60s later) finds the wallet healthy.
*
* State-lock discipline: both layers use state.tryAcquire with a synthetic
* escrowRef so they share the existing single-slot lock with real Claims.
* If a Claim is in flight, the consolidation skips this tick.
* Wallet-lock discipline: consolidation spends EVERY wallet UTxO, so it runs
* inside state.walletMutex — the same mutex that serializes Claims/Submits —
* queueing behind any in-flight chain op instead of skipping. Session slots
* are not consumed: consolidation is a wallet op, not a session.
*/
import type { ChainProvider } from "@marketplace/shared/chain";
@@ -31,7 +32,6 @@ import {
import type { SupplierState } from "./state.js";
const WALLET_HEALTH_LOCK_REF = "__wallet_consolidate__";
const DEFAULT_COLLATERAL_LOVELACE = 5_000_000n;
const DEFAULT_AWAIT_TIMEOUT_MS = 90_000;
const DEFAULT_ON_FAILURE_DEBOUNCE_MS = 5 * 60_000;
@@ -77,20 +77,20 @@ async function consolidateOnce(
opts.log("skip: another consolidate already in flight");
return;
}
if (!state.tryAcquire(WALLET_HEALTH_LOCK_REF)) {
opts.log("skip: supplier busy");
return;
}
inFlight.add(state);
const consolidate: ConsolidateFn = deps.consolidate ?? runConsolidateWallet;
try {
const result = await consolidate({
chain: deps.chain,
walletKey: deps.supplierKey,
collateralLovelace: opts.collateralLovelace,
awaitTimeoutMs: opts.awaitTimeoutMs,
log: opts.log,
});
// Queue behind any in-flight Claim/Submit — consolidation reshapes the
// whole wallet and must never race another wallet spend.
const result = await state.walletMutex.run(() =>
consolidate({
chain: deps.chain,
walletKey: deps.supplierKey,
collateralLovelace: opts.collateralLovelace,
awaitTimeoutMs: opts.awaitTimeoutMs,
log: opts.log,
}),
);
if (result.reason === "already-healthy") {
opts.log("already-healthy");
} else {
@@ -100,7 +100,6 @@ async function consolidateOnce(
opts.log(`consolidate failed: ${(err as Error).message}`);
} finally {
inFlight.delete(state);
state.release();
}
}

View File

@@ -31,6 +31,8 @@ export function buildSampleConfig(): SupplierConfig {
piperUrl: "",
piperTimeoutMs: 120_000,
capabilityKind: "chat",
maxChatSessions: 1,
chatSettleMode: "full",
liveChain: false,
};
}

View File

@@ -0,0 +1,118 @@
/**
* buyer-sdk-chat-session.test.ts — settle-mode handling in startChat/endChat.
*
* Ticket-mode suppliers answer /v1/chat/start with {status:"ticket"} and
* /v1/chat/end with {status:"ended", settle_mode:"ticket"}. The SDK must map
* those to settleMode:"ticket" — and, at end, skip receipt verification AND
* the Accept tx entirely (zero chain calls). Full-shaped responses keep the
* existing behavior.
*/
import { describe, it, expect, vi } from "vitest";
import { Marketplace } from "../../buyer/src/sdk/Marketplace.js";
import { MockChainProvider } from "../../packages/shared/src/chain/MockChainProvider.js";
import { encodeAdvertDatum } from "../../packages/shared/src/cbor/AdvertDatum.js";
import type { AdvertDatum } from "../../packages/shared/src/cbor/types.js";
import { buildBuyerWalletKey } from "../fixtures/buyer-side/wallet-keys.js";
const ADVERT_TX_HASH = "b".repeat(64);
const ESCROW_REF = { txHash: "f".repeat(64), index: 0 };
const ESCROW_REF_STR = `${ESCROW_REF.txHash}#0`;
function advertDatum(): AdvertDatum {
return {
supplier_pkh: "abcdef0123456789abcdef0123456789abcdef0123456789abcdef01",
capability_id: "llm.chat.v1",
model: "kimi",
max_output_tokens: 512,
max_processing_ms: 1_800_000,
price_lovelace: 200_000n,
supplier_bond_lovelace: 1_000_000n,
buyer_bond_lovelace: 1_000_000n,
endpoint_url: "http://supplier.test",
detail_uri: "ipfs://QmTest",
detail_hash: "a".repeat(64),
advertised_at: 1_745_500_000_000,
status: "Active",
};
}
/** fetch stub answering the supplier's /status, /v1/chat/start, /v1/chat/end. */
function supplierFetch(responses: { start?: unknown; end?: unknown }) {
return vi.fn(async (url: RequestInfo | URL) => {
const u = String(url);
if (u.includes("/status")) {
return new Response(JSON.stringify({ status: "free" }), { status: 200 });
}
if (u.includes("/v1/chat/start")) {
return new Response(JSON.stringify(responses.start ?? { status: "claimed" }), { status: 200 });
}
if (u.includes("/v1/chat/end")) {
return new Response(JSON.stringify(responses.end ?? {}), { status: 200 });
}
if (u.includes("/capability")) {
return new Response(JSON.stringify({ supplier_pkh: "", model: "kimi" }), { status: 200 });
}
return new Response("{}", { status: 200 });
}) as unknown as typeof globalThis.fetch;
}
function makeMp(fetchImpl: typeof globalThis.fetch, chain = new MockChainProvider()) {
chain.seed({
ref: { txHash: ADVERT_TX_HASH, index: 0 },
address: "addr_test1wfakeadvert",
lovelace: 2_000_000n,
assets: {},
datumHex: encodeAdvertDatum(advertDatum()),
scriptRef: null,
});
return { mp: new Marketplace({
chain,
indexerUrl: "http://indexer.test",
walletKey: buildBuyerWalletKey(),
networkParams: { networkId: 0 },
_fetch: fetchImpl,
}), chain };
}
describe("Marketplace.startChat — settle-mode mapping", () => {
it("maps {status:'ticket'} to settleMode ticket", async () => {
const { mp } = makeMp(supplierFetch({ start: { status: "ticket", settle_mode: "ticket" } }));
const result = await mp.startChat({ advertRef: { txHash: ADVERT_TX_HASH, index: 0 }, payment_lovelace: 200_000n });
expect(result.settleMode).toBe("ticket");
});
it("maps {status:'claimed'} (and legacy responses) to settleMode full", async () => {
const { mp } = makeMp(supplierFetch({ start: { status: "claimed" } }));
const result = await mp.startChat({ advertRef: { txHash: ADVERT_TX_HASH, index: 0 }, payment_lovelace: 200_000n });
expect(result.settleMode).toBe("full");
});
});
describe("Marketplace.endChat — ticket branch", () => {
it("returns the ticket result with zero chain calls and no receipt verification", async () => {
const chain = new MockChainProvider();
const submitSpy = vi.spyOn(chain, "submitTx");
const { mp } = makeMp(
supplierFetch({ end: { status: "ended", escrow_ref: ESCROW_REF_STR, settle_mode: "ticket" } }),
chain,
);
const result = await mp.endChat({
escrowRef: ESCROW_REF,
sessionNonce: "nonce-abc",
supplierBaseUrl: "http://supplier.test",
});
expect(result.settleMode).toBe("ticket");
expect("receipt" in result).toBe(false);
expect(submitSpy).not.toHaveBeenCalled(); // no Accept tx
});
it("still rejects a receipt-less response that does not declare ticket mode", async () => {
const { mp } = makeMp(supplierFetch({ end: { status: "submitted" } }));
await expect(
mp.endChat({ escrowRef: ESCROW_REF, sessionNonce: "nonce-abc", supplierBaseUrl: "http://supplier.test" }),
).rejects.toThrow(/receipt/);
});
});

View File

@@ -0,0 +1,112 @@
/**
* shared-chat-message.test.ts — normalizeChatMessage / chatMessagesEquivalent.
*
* The supplier transcript and the gateway mirror must canonicalize to the SAME
* string (receipt response_hash = sha256(canonicalize(transcript))), so the
* normalizer's field-presence behavior is load-bearing: absent optional fields
* must be truly absent, content is never null, vendor extras are stripped.
*/
import { describe, it, expect } from "vitest";
import { normalizeChatMessage, chatMessagesEquivalent, toolCallsEqual } from "../../packages/shared/src/tx/chatMessage.js";
import { canonicalize } from "../../packages/shared/src/cbor/canonical.js";
const TOOL_CALL = { id: "call_1", type: "function", function: { name: "get_time", arguments: "{}" } };
describe("normalizeChatMessage", () => {
it("passes through plain user/system/assistant messages", () => {
expect(normalizeChatMessage({ role: "user", content: "hi" })).toEqual({ role: "user", content: "hi" });
expect(normalizeChatMessage({ role: "system", content: "be nice" })).toEqual({ role: "system", content: "be nice" });
});
it("normalizes null/undefined content to empty string", () => {
expect(normalizeChatMessage({ role: "assistant", content: null, tool_calls: [TOOL_CALL] })).toEqual({
role: "assistant",
content: "",
tool_calls: [TOOL_CALL],
});
});
it("keeps optional fields ABSENT when not provided (hash identity)", () => {
const msg = normalizeChatMessage({ role: "assistant", content: "x" })!;
expect("tool_calls" in msg).toBe(false);
expect("tool_call_id" in msg).toBe(false);
});
it("strips vendor extras from tool calls", () => {
const msg = normalizeChatMessage({
role: "assistant",
content: "",
tool_calls: [{ ...TOOL_CALL, extra_vendor_field: 42, function: { ...TOOL_CALL.function, parsed: {} } }],
})!;
expect(msg.tool_calls).toEqual([TOOL_CALL]);
});
it("requires tool role to carry tool_call_id", () => {
expect(normalizeChatMessage({ role: "tool", content: "result" })).toBeNull();
expect(normalizeChatMessage({ role: "tool", content: "result", tool_call_id: "call_1" })).toEqual({
role: "tool",
content: "result",
tool_call_id: "call_1",
});
});
it("rejects tool_calls on non-assistant roles and malformed shapes", () => {
expect(normalizeChatMessage({ role: "user", content: "x", tool_calls: [TOOL_CALL] })).toBeNull();
expect(normalizeChatMessage({ role: "assistant", content: "", tool_calls: [{ id: "", type: "function", function: { name: "f", arguments: "" } }] })).toBeNull();
expect(normalizeChatMessage({ role: "bogus", content: "x" })).toBeNull();
expect(normalizeChatMessage("nope")).toBeNull();
expect(normalizeChatMessage({ role: "user", content: 42 })).toBeNull();
});
it("canonicalizes identically regardless of input key order (mirror vs transcript)", () => {
const a = normalizeChatMessage({ role: "assistant", content: "", tool_calls: [TOOL_CALL] })!;
const b = normalizeChatMessage({
tool_calls: [{ function: { arguments: "{}", name: "get_time" }, type: "function", id: "call_1" }],
content: null,
role: "assistant",
})!;
expect(canonicalize([a])).toBe(canonicalize([b]));
});
});
describe("chatMessagesEquivalent", () => {
it("matches normalized equals and rejects differences", () => {
const a = normalizeChatMessage({ role: "assistant", content: "", tool_calls: [TOOL_CALL] })!;
const b = normalizeChatMessage({ role: "assistant", content: null, tool_calls: [TOOL_CALL] })!;
expect(chatMessagesEquivalent(a, b)).toBe(true);
const c = normalizeChatMessage({ role: "assistant", content: "hi" })!;
expect(chatMessagesEquivalent(a, c)).toBe(false);
const d = normalizeChatMessage({
role: "assistant",
content: "",
tool_calls: [{ ...TOOL_CALL, function: { name: "get_time", arguments: '{"tz":"utc"}' } }],
})!;
expect(chatMessagesEquivalent(a, d)).toBe(false);
});
it("compares tool results by tool_call_id", () => {
const a = normalizeChatMessage({ role: "tool", content: "3pm", tool_call_id: "call_1" })!;
const b = normalizeChatMessage({ role: "tool", content: "3pm", tool_call_id: "call_1" })!;
const c = normalizeChatMessage({ role: "tool", content: "3pm", tool_call_id: "call_2" })!;
expect(chatMessagesEquivalent(a, b)).toBe(true);
expect(chatMessagesEquivalent(a, c)).toBe(false);
});
});
describe("toolCallsEqual", () => {
const CALL = { id: "call_1", type: "function" as const, function: { name: "get_time", arguments: "{}" } };
it("treats missing and empty lists as equal (exported for the affinity matcher)", () => {
expect(toolCallsEqual(undefined, undefined)).toBe(true);
expect(toolCallsEqual(undefined, [])).toBe(true);
expect(toolCallsEqual([CALL], [{ ...CALL }])).toBe(true);
});
it("rejects id/name/arguments differences and length mismatches", () => {
expect(toolCallsEqual([CALL], [])).toBe(false);
expect(toolCallsEqual([CALL], [{ ...CALL, id: "call_2" }])).toBe(false);
expect(toolCallsEqual([CALL], [{ ...CALL, function: { name: "get_time", arguments: '{"tz":"utc"}' } }])).toBe(false);
});
});

View File

@@ -0,0 +1,122 @@
/**
* supplier-chat-session-rollback.test.ts — failed-turn transcript rollback.
*
* The chat-session message handler appends the incoming delta BEFORE calling
* the LLM backend; the gateway mirror only appends after a successful turn.
* On backend failure the supplier must truncate the delta back out, so a
* retried turn doesn't duplicate messages and both transcripts stay
* hash-identical for the receipt.
*/
import { describe, it, expect, vi, afterEach } from "vitest";
import request from "supertest";
import { MockChainProvider } from "../../packages/shared/src/chain/MockChainProvider.js";
import type { AdvertDatum, EscrowDatum } from "../../packages/shared/src/cbor/types.js";
import type { SupplierConfig } from "../../supplier/src/config.js";
import { SupplierState } from "../../supplier/src/state.js";
import { JobStore } from "../../supplier/src/jobs.js";
import { ChatSessionStore } from "../../supplier/src/chatSession.js";
import { createApp } from "../../supplier/src/server.js";
import { buildSampleConfig } from "../fixtures/supplier-side/sample-config.js";
import { buildSupplierWalletKey } from "../fixtures/supplier-side/wallet-keys.js";
const ESCROW_REF = `${"e".repeat(64)}#0`;
function chatSessionConfig(): SupplierConfig {
return {
...buildSampleConfig(),
capabilityKind: "chat-session",
llmBackend: "openai",
ollamaUrl: "",
openaiBaseUrl: "http://up",
openaiApiKey: "",
openaiTimeoutMs: 5_000,
openaiMaxTokens: 0,
openaiReasoningDisabled: false,
chatIdleTimeoutMs: 60_000,
} as SupplierConfig;
}
function makeAppWithSession() {
const chatSessions = new ChatSessionStore();
const record = chatSessions.create({
escrowRef: ESCROW_REF,
claimedRef: { txHash: "c".repeat(64), index: 0 },
advert: { model: "kimi" } as AdvertDatum,
escrowDatum: {} as EscrowDatum,
});
const app = createApp({
chain: new MockChainProvider(),
state: new SupplierState(),
config: chatSessionConfig(),
supplierKey: buildSupplierWalletKey(),
jobs: new JobStore(),
chatSessions,
});
return { app, chatSessions, record };
}
function sseBody(frames: unknown[]): string {
return frames.map((f) => `data: ${typeof f === "string" ? f : JSON.stringify(f)}\n\n`).join("");
}
const OK_STREAM = sseBody([
{ choices: [{ delta: { content: "Hello" }, finish_reason: null }] },
{ choices: [{ delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 3, completion_tokens: 2 } },
"[DONE]",
]);
afterEach(() => {
vi.unstubAllGlobals();
});
describe("ChatSessionStore.truncateTranscript", () => {
it("rolls an appended delta back and no-ops on stale targets", () => {
const store = new ChatSessionStore();
store.create({
escrowRef: ESCROW_REF,
claimedRef: { txHash: "c".repeat(64), index: 0 },
advert: { model: "kimi" } as AdvertDatum,
escrowDatum: {} as EscrowDatum,
});
store.appendMessages(ESCROW_REF, [{ role: "user", content: "hi" }]);
store.truncateTranscript(ESCROW_REF, 0);
expect(store.get(ESCROW_REF)!.transcript).toEqual([]);
store.appendMessages(ESCROW_REF, [{ role: "user", content: "hi" }]);
store.truncateTranscript(ESCROW_REF, 5); // target beyond current length
store.truncateTranscript(ESCROW_REF, -1); // nonsense target
store.truncateTranscript("missing", 0); // unknown session
expect(store.get(ESCROW_REF)!.transcript).toHaveLength(1);
});
});
describe("POST /v1/chat/message — failed-turn rollback", () => {
it("truncates the delta on backend failure, then a retry appends exactly once", async () => {
const { app, record } = makeAppWithSession();
const delta = [{ role: "user", content: "hi" }];
vi.stubGlobal("fetch", vi.fn().mockRejectedValue(new Error("boom")));
const failed = await request(app)
.post("/v1/chat/message")
.set("X-Escrow-Ref", ESCROW_REF)
.send({ messages: delta });
expect(failed.text).toContain('"type":"error"');
// The failed delta was rolled back — nothing to duplicate on retry.
expect(record.transcript).toEqual([]);
vi.stubGlobal(
"fetch",
vi.fn().mockResolvedValue(new Response(OK_STREAM, { status: 200, headers: { "content-type": "text/event-stream" } })),
);
const ok = await request(app)
.post("/v1/chat/message")
.set("X-Escrow-Ref", ESCROW_REF)
.send({ messages: delta });
expect(ok.text).toContain('"type":"done"');
expect(record.transcript).toEqual([
{ role: "user", content: "hi" },
{ role: "assistant", content: "Hello" },
]);
});
});

View File

@@ -0,0 +1,277 @@
/**
* supplier-chat-session-ticket.test.ts — entry-ticket settle mode + N slots.
*
* Ticket mode (CHAT_SETTLE_MODE=ticket): /v1/chat/start verifies the Open
* escrow but never Claims (zero chain txs), arms no hard-cap timer, and
* /v1/chat/end returns a receipt-less {status:"ended"}. A used escrow ref can
* never open a second session (records are retained). MAX_CHAT_SESSIONS
* admits N concurrent sessions and the N+1th start gets 409 supplier_busy.
*/
import { describe, it, expect, vi, afterEach } from "vitest";
import request from "supertest";
import { createHash } from "crypto";
import type { Application } from "express";
import { MockChainProvider } from "../../packages/shared/src/chain/MockChainProvider.js";
import { canonicalize } from "../../packages/shared/src/cbor/canonical.js";
import { encodeAdvertDatum } from "../../packages/shared/src/cbor/AdvertDatum.js";
import { encodeEscrowDatum } from "../../packages/shared/src/cbor/EscrowDatum.js";
import type { AdvertDatum, EscrowDatum } from "../../packages/shared/src/cbor/types.js";
import { chatSessionPromptHash } from "../../packages/shared/src/tx/escrow/postChatEscrow.js";
import type { SupplierConfig } from "../../supplier/src/config.js";
import { SupplierState } from "../../supplier/src/state.js";
import { JobStore } from "../../supplier/src/jobs.js";
import { ChatSessionStore } from "../../supplier/src/chatSession.js";
import { createApp } from "../../supplier/src/server.js";
import { buildSampleConfig, SAMPLE_ADVERT_TX_HASH, SAMPLE_ADVERT_INDEX } from "../fixtures/supplier-side/sample-config.js";
import { buildSupplierWalletKey, SUPPLIER_PKH } from "../fixtures/supplier-side/wallet-keys.js";
const MODEL = "kimi";
const MAX_TOKENS = 512;
const CAPABILITY = "llm.chat.v1";
const NONCE = "nonce-abc";
const POSTED_AT = 1_745_500_000_000;
const DELIVER_BY = POSTED_AT + 90_000;
function sha256Hex(s: string): string {
return createHash("sha256").update(s, "utf8").digest("hex");
}
function advertDatum(): AdvertDatum {
return {
supplier_pkh: SUPPLIER_PKH,
capability_id: CAPABILITY,
model: MODEL,
max_output_tokens: MAX_TOKENS,
max_processing_ms: 1_800_000,
price_lovelace: 200_000n,
supplier_bond_lovelace: 1_000_000n,
buyer_bond_lovelace: 1_000_000n,
endpoint_url: "http://supplier.example:8080",
detail_uri: "ipfs://QmTest",
detail_hash: "a".repeat(64),
advertised_at: POSTED_AT,
status: "Active",
};
}
function escrowDatum(nonce: string): EscrowDatum {
return {
buyer_pkh: "1234567890abcdef1234567890abcdef1234567890abcdef12345678",
supplier_pkh: SUPPLIER_PKH,
advert_ref: { txHash: SAMPLE_ADVERT_TX_HASH, index: SAMPLE_ADVERT_INDEX },
capability_id: CAPABILITY,
request_spec_hash: sha256Hex(canonicalize({ capability_id: CAPABILITY, max_output_tokens: MAX_TOKENS, model: MODEL })),
prompt_hash: chatSessionPromptHash({ session_nonce: nonce }),
payment_lovelace: 200_000n,
buyer_bond_lovelace: 1_000_000n,
supplier_bond_lovelace: 1_000_000n,
deliver_by: DELIVER_BY,
posted_at: POSTED_AT,
submitted_at: null,
result_receipt_hash: null,
state: "Open",
};
}
interface Harness {
app: Application;
chain: MockChainProvider;
state: SupplierState;
chatSessions: ChatSessionStore;
seedEscrow: (txHashChar: string, nonce?: string) => string;
}
function makeHarness(overrides?: Partial<SupplierConfig>): Harness {
const chain = new MockChainProvider();
chain.seed({
ref: { txHash: SAMPLE_ADVERT_TX_HASH, index: SAMPLE_ADVERT_INDEX },
address: "addr_test1wfakeadvertaddress",
lovelace: 2_000_000n,
assets: {},
datumHex: encodeAdvertDatum(advertDatum()),
scriptRef: null,
});
const config: SupplierConfig = {
...buildSampleConfig(),
capabilityKind: "chat-session",
llmBackend: "openai",
ollamaUrl: "",
openaiBaseUrl: "http://up",
openaiApiKey: "",
openaiTimeoutMs: 5_000,
openaiMaxTokens: 0,
openaiReasoningDisabled: false,
chatIdleTimeoutMs: 60_000,
chatSettleMode: "ticket",
maxChatSessions: 1,
...overrides,
} as SupplierConfig;
const state = new SupplierState(config.maxChatSessions);
const chatSessions = new ChatSessionStore();
const app = createApp({
chain,
state,
config,
supplierKey: buildSupplierWalletKey(),
jobs: new JobStore(),
chatSessions,
});
const seedEscrow = (txHashChar: string, nonce = NONCE): string => {
const txHash = txHashChar.repeat(64);
chain.seed({
ref: { txHash, index: 0 },
address: "addr_test1wfakeescrowaddress",
lovelace: 2_200_000n,
assets: {},
datumHex: encodeEscrowDatum(escrowDatum(nonce)),
scriptRef: null,
});
return `${txHash}#0`;
};
return { app, chain, state, chatSessions, seedEscrow };
}
afterEach(() => {
vi.restoreAllMocks();
});
describe("chat-session ticket mode — /v1/chat/start", () => {
it("verifies the escrow, skips Claim, and arms no hard-cap timer", async () => {
const h = makeHarness();
const submitSpy = vi.spyOn(h.chain, "submitTx");
const ref = h.seedEscrow("e");
const res = await request(h.app)
.post("/v1/chat/start")
.set("X-Escrow-Ref", ref)
.send({ session_nonce: NONCE });
expect(res.status).toBe(200);
expect(res.body).toEqual({ status: "ticket", escrow_ref: ref, settle_mode: "ticket" });
expect(submitSpy).not.toHaveBeenCalled(); // zero chain txs
const record = h.chatSessions.get(ref)!;
expect(record.settleMode).toBe("ticket");
expect(record.claimedRef).toBeUndefined();
expect(record.idleTimer).toBeDefined();
expect(record.hardCapTimer).toBeUndefined();
expect(h.state.snapshot().activeSessions).toBe(1);
});
it("still rejects a foreign-supplier escrow", async () => {
const h = makeHarness();
const txHash = "c".repeat(64);
h.chain.seed({
ref: { txHash, index: 0 },
address: "addr_test1wfakeescrowaddress",
lovelace: 2_200_000n,
assets: {},
datumHex: encodeEscrowDatum({ ...escrowDatum(NONCE), supplier_pkh: "9".repeat(56) }),
scriptRef: null,
});
const res = await request(h.app)
.post("/v1/chat/start")
.set("X-Escrow-Ref", `${txHash}#0`)
.send({ session_nonce: NONCE });
expect(res.status).toBe(403);
expect(res.body.reason).toBe("wrong_supplier");
});
it("rejects reusing an escrow ref while the session is active AND after it ended", async () => {
const h = makeHarness();
const ref = h.seedEscrow("e");
const first = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", ref).send({ session_nonce: NONCE });
expect(first.status).toBe(200);
const whileActive = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", ref).send({ session_nonce: NONCE });
expect(whileActive.status).toBe(409);
expect(whileActive.body.reason).toBe("session_exists");
await request(h.app).post("/v1/chat/end").set("X-Escrow-Ref", ref).send({});
// The escrow is still Open on-chain (never Claimed) — the retained record
// is the only thing standing between a used ticket and a free session.
const afterEnd = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", ref).send({ session_nonce: NONCE });
expect(afterEnd.status).toBe(409);
expect(afterEnd.body.reason).toBe("session_exists");
});
});
describe("chat-session ticket mode — /v1/chat/end", () => {
it("ends without a receipt, frees the slot, and is idempotent", async () => {
const h = makeHarness();
const submitSpy = vi.spyOn(h.chain, "submitTx");
const ref = h.seedEscrow("e");
await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", ref).send({ session_nonce: NONCE });
const end = await request(h.app).post("/v1/chat/end").set("X-Escrow-Ref", ref).send({});
expect(end.status).toBe(200);
expect(end.body).toEqual({ status: "ended", escrow_ref: ref, settle_mode: "ticket" });
expect(end.body.receipt).toBeUndefined();
expect(submitSpy).not.toHaveBeenCalled();
expect(h.state.snapshot().activeSessions).toBe(0);
const again = await request(h.app).post("/v1/chat/end").set("X-Escrow-Ref", ref).send({});
expect(again.status).toBe(200);
expect(again.body.status).toBe("ended");
});
});
describe("chat-session slots — MAX_CHAT_SESSIONS", () => {
it("admits N concurrent sessions and 409s the N+1th", async () => {
const h = makeHarness({ maxChatSessions: 2 });
// NB: "b".repeat(64) is the advert tx hash — do not reuse it for escrows.
const refA = h.seedEscrow("d");
const refB = h.seedEscrow("e");
const refC = h.seedEscrow("f");
const a = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refA).send({ session_nonce: NONCE });
const b = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refB).send({ session_nonce: NONCE });
expect(a.status).toBe(200);
expect(b.status).toBe(200);
const status = await request(h.app).get("/status");
expect(status.body.status).toBe("working");
expect(status.body.active_sessions).toBe(2);
expect(status.body.max_sessions).toBe(2);
const c = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refC).send({ session_nonce: NONCE });
expect(c.status).toBe(409);
expect(c.body.reason).toBe("supplier_busy");
// Ending one session readmits the third.
await request(h.app).post("/v1/chat/end").set("X-Escrow-Ref", refA).send({});
const retry = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refC).send({ session_nonce: NONCE });
expect(retry.status).toBe(200);
});
it("serves concurrent message turns on distinct sessions", async () => {
const h = makeHarness({ maxChatSessions: 2 });
const refA = h.seedEscrow("d");
const refB = h.seedEscrow("e");
await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refA).send({ session_nonce: NONCE });
await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refB).send({ session_nonce: NONCE });
const stream = [
`data: ${JSON.stringify({ choices: [{ delta: { content: "Hello" }, finish_reason: null }] })}\n\n`,
`data: ${JSON.stringify({ choices: [{ delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 3, completion_tokens: 2 } })}\n\n`,
"data: [DONE]\n\n",
].join("");
vi.stubGlobal(
"fetch",
vi.fn().mockImplementation(async () =>
new Response(stream, { status: 200, headers: { "content-type": "text/event-stream" } })),
);
const [ta, tb] = await Promise.all([
request(h.app).post("/v1/chat/message").set("X-Escrow-Ref", refA).send({ messages: [{ role: "user", content: "hi" }] }),
request(h.app).post("/v1/chat/message").set("X-Escrow-Ref", refB).send({ messages: [{ role: "user", content: "yo" }] }),
]);
expect(ta.text).toContain('"type":"done"');
expect(tb.text).toContain('"type":"done"');
expect(h.chatSessions.get(refA)!.transcript).toHaveLength(2);
expect(h.chatSessions.get(refB)!.transcript).toHaveLength(2);
vi.unstubAllGlobals();
});
});

View File

@@ -184,6 +184,46 @@ describe("loadConfig() — LIVE_CHAIN parsing", () => {
});
});
// ─── loadConfig — HuggingFace router preset (W1) ────────────────────────────
// The HF Inference Providers router (router.huggingface.co) is OpenAI-compatible
// and reuses LLM_BACKEND=openai, but rejects the OpenRouter-specific
// reasoning:{enabled:false} param with HTTP 400 — so OPENAI_REASONING must stay
// unset for it. loadConfig fails fast at boot to catch the copy-from-OpenRouter
// footgun. See docs/HUGGINGFACE_ROUTER_SETUP.md.
describe("loadConfig() — HuggingFace router guard", () => {
const hfEnv = (over: Record<string, string> = {}) => ({
...buildSampleEnv(),
LLM_BACKEND: "openai",
OPENAI_BASE_URL: "https://router.huggingface.co",
OPENAI_API_KEY: "hf_sampletoken",
...over,
});
it("accepts the HF router with OPENAI_REASONING unset", () => {
const cfg = loadConfig(hfEnv());
expect(cfg.openaiBaseUrl).toBe("https://router.huggingface.co");
expect(cfg.llmBackend).toBe("openai");
expect(cfg.openaiReasoningDisabled).toBe(false);
});
it("throws when OPENAI_REASONING=off is set against the HF router", () => {
expect(() => loadConfig(hfEnv({ OPENAI_REASONING: "off" }))).toThrow(
/OPENAI_REASONING must be unset for the HuggingFace router/i,
);
});
it("still allows OPENAI_REASONING=off for non-HF openai backends (e.g. OpenRouter)", () => {
const cfg = loadConfig({
...buildSampleEnv(),
LLM_BACKEND: "openai",
OPENAI_BASE_URL: "https://openrouter.ai/api",
OPENAI_REASONING: "off",
});
expect(cfg.openaiReasoningDisabled).toBe(true);
});
});
// ─── parseAdvertRef ──────────────────────────────────────────────────────────
describe("parseAdvertRef()", () => {
@@ -218,3 +258,39 @@ describe("parseAdvertRef()", () => {
expect(() => parseAdvertRef(`${"a".repeat(64)}#-1`)).toThrow();
});
});
// ─── loadConfig — chat-session concurrency + settle mode ─────────────────────
describe("loadConfig() — MAX_CHAT_SESSIONS / CHAT_SETTLE_MODE", () => {
const chatSessionEnv = () => ({
...buildSampleEnv(),
CAPABILITY_KIND: "chat-session",
LLM_BACKEND: "openai",
OPENAI_BASE_URL: "http://up",
});
it("defaults to 1 slot and full settle mode", () => {
const cfg = loadConfig(buildSampleEnv());
expect(cfg.maxChatSessions).toBe(1);
expect(cfg.chatSettleMode).toBe("full");
});
it("parses MAX_CHAT_SESSIONS and CHAT_SETTLE_MODE for chat-session kind", () => {
const cfg = loadConfig({ ...chatSessionEnv(), MAX_CHAT_SESSIONS: "8", CHAT_SETTLE_MODE: "ticket" });
expect(cfg.maxChatSessions).toBe(8);
expect(cfg.chatSettleMode).toBe("ticket");
});
it("rejects MAX_CHAT_SESSIONS > 1 for non-chat-session kinds", () => {
expect(() => loadConfig({ ...buildSampleEnv(), MAX_CHAT_SESSIONS: "4" })).toThrow();
});
it("rejects CHAT_SETTLE_MODE=ticket for non-chat-session kinds", () => {
expect(() => loadConfig({ ...buildSampleEnv(), CHAT_SETTLE_MODE: "ticket" })).toThrow();
});
it("rejects invalid values", () => {
expect(() => loadConfig({ ...chatSessionEnv(), MAX_CHAT_SESSIONS: "0" })).toThrow();
expect(() => loadConfig({ ...chatSessionEnv(), CHAT_SETTLE_MODE: "nope" })).toThrow();
});
});

View File

@@ -0,0 +1,103 @@
/**
* supplier-openai-stream.test.ts — callOpenAiStream tools passthrough.
*
* Mocks global.fetch with SSE-shaped Response bodies. Covers:
* - tools/tool_choice included in the upstream payload only when provided
* - fragmented delta.tool_calls accumulation across frames
* - pure tool-call turns (empty content) no longer throw openai_malformed
* - empty content AND no tool calls still throws
* - finish_reason capture
*/
import { describe, it, expect, vi, afterEach } from "vitest";
import { callOpenAiStream, OpenAiError } from "../../supplier/src/openai.js";
const BASE = { baseUrl: "http://up", model: "kimi", messages: [{ role: "user" as const, content: "hi" }], timeoutMs: 5_000 };
const TOOLS = [{ type: "function", function: { name: "get_time", parameters: {} } }];
function sse(frames: unknown[]): string {
return frames.map((f) => `data: ${typeof f === "string" ? f : JSON.stringify(f)}\n\n`).join("");
}
function makeStreamFetch(body: string) {
return vi.fn().mockResolvedValue(new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } }));
}
afterEach(() => {
vi.restoreAllMocks();
});
describe("callOpenAiStream — tools passthrough", () => {
const CONTENT_FRAMES = sse([
{ choices: [{ delta: { content: "Hello" }, finish_reason: null }] },
{ choices: [{ delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 3, completion_tokens: 2 } },
"[DONE]",
]);
it("includes tools + tool_choice in the payload when provided", async () => {
const fetchFn = makeStreamFetch(CONTENT_FRAMES);
vi.stubGlobal("fetch", fetchFn);
await callOpenAiStream({ ...BASE, tools: TOOLS, toolChoice: "auto" }, () => {});
const body = JSON.parse(fetchFn.mock.calls[0][1].body as string);
expect(body.tools).toEqual(TOOLS);
expect(body.tool_choice).toBe("auto");
});
it("omits tools from the payload when not provided", async () => {
const fetchFn = makeStreamFetch(CONTENT_FRAMES);
vi.stubGlobal("fetch", fetchFn);
await callOpenAiStream(BASE, () => {});
const body = JSON.parse(fetchFn.mock.calls[0][1].body as string);
expect("tools" in body).toBe(false);
expect("tool_choice" in body).toBe(false);
});
it("accumulates fragmented delta.tool_calls across frames", async () => {
const frames = sse([
{ choices: [{ delta: { tool_calls: [{ index: 0, id: "call_1", type: "function", function: { name: "get_time", arguments: "" } }] }, finish_reason: null }] },
{ choices: [{ delta: { tool_calls: [{ index: 0, function: { arguments: '{"tz":' } }] }, finish_reason: null }] },
{ choices: [{ delta: { tool_calls: [{ index: 0, function: { arguments: '"utc"}' } }] }, finish_reason: null }] },
{ choices: [{ delta: {}, finish_reason: "tool_calls" }], usage: { prompt_tokens: 7, completion_tokens: 5 } },
"[DONE]",
]);
vi.stubGlobal("fetch", makeStreamFetch(frames));
const result = await callOpenAiStream({ ...BASE, tools: TOOLS }, () => {});
expect(result.tool_calls).toEqual([
{ id: "call_1", type: "function", function: { name: "get_time", arguments: '{"tz":"utc"}' } },
]);
expect(result.finish_reason).toBe("tool_calls");
expect(result.content).toBe("");
expect(result.prompt_tokens).toBe(7);
});
it("orders multiple tool calls by index", async () => {
const frames = sse([
{ choices: [{ delta: { tool_calls: [
{ index: 1, id: "call_b", type: "function", function: { name: "b", arguments: "{}" } },
{ index: 0, id: "call_a", type: "function", function: { name: "a", arguments: "{}" } },
] }, finish_reason: null }] },
{ choices: [{ delta: {}, finish_reason: "tool_calls" }] },
"[DONE]",
]);
vi.stubGlobal("fetch", makeStreamFetch(frames));
const result = await callOpenAiStream({ ...BASE, tools: TOOLS }, () => {});
expect(result.tool_calls?.map((t) => t.id)).toEqual(["call_a", "call_b"]);
});
it("still throws openai_malformed when content AND tool calls are both empty", async () => {
const frames = sse([{ choices: [{ delta: {}, finish_reason: "stop" }] }, "[DONE]"]);
vi.stubGlobal("fetch", makeStreamFetch(frames));
await expect(callOpenAiStream(BASE, () => {})).rejects.toMatchObject({ reason: "openai_malformed" });
await expect(callOpenAiStream(BASE, () => {})).rejects.toBeInstanceOf(OpenAiError);
});
it("still streams plain content unchanged (no tools)", async () => {
vi.stubGlobal("fetch", makeStreamFetch(CONTENT_FRAMES));
const tokens: string[] = [];
const result = await callOpenAiStream(BASE, (t) => tokens.push(t));
expect(tokens).toEqual(["Hello"]);
expect(result.content).toBe("Hello");
expect(result.tool_calls).toBeUndefined();
expect(result.finish_reason).toBe("stop");
});
});

View File

@@ -117,6 +117,25 @@ describe("GET /status — response body when offline", () => {
});
});
describe("GET /status — session slot fields", () => {
it("reports active_sessions/max_sessions when free", async () => {
const state = new SupplierState();
const res = await request(makeApp(state)).get("/status");
expect(res.body.active_sessions).toBe(0);
expect(res.body.max_sessions).toBe(1);
});
it("stays free with slots remaining and reports the active count", async () => {
const state = new SupplierState(3);
state.tryAcquire(ESCROW_REF);
const res = await request(makeApp(state)).get("/status");
expect(res.body.status).toBe("free");
expect(res.body.active_sessions).toBe(1);
expect(res.body.max_sessions).toBe(3);
expect(res.body.current_escrow_ref).toBeUndefined();
});
});
describe("GET /status — does not touch chain", () => {
it("returns 200 even when chain has no UTxOs seeded", async () => {
// If the handler incorrectly touches chain, it would throw because

View File

@@ -1,11 +1,13 @@
/**
* supplier-state.test.ts — RED phase tests for supplier/src/state.ts
* supplier-state.test.ts — tests for supplier/src/state.ts
*
* Tests the SupplierState in-process slot lock:
* - tryAcquire(escrowRef): boolean — atomic test-and-set
* - release()
* Tests the SupplierState session-slot admission (capacity-aware) + the
* wallet mutex:
* - tryAcquire(escrowRef): boolean — slot admission (default capacity 1)
* - release(escrowRef) — frees exactly that ref's slot
* - markOffline()
* - snapshot(): { status, currentEscrowRef?, lastSeenIso }
* - snapshot(): { status, currentEscrowRef?, activeSessions, maxSessions, ... }
* - walletMutex.run — strict serialization of wallet chain ops
*/
import { describe, it, expect, beforeEach } from "vitest";
@@ -13,6 +15,7 @@ import { SupplierState } from "../../supplier/src/state.js";
const ESCROW_REF_A = `${"f".repeat(64)}#0`;
const ESCROW_REF_B = `${"e".repeat(64)}#1`;
const ESCROW_REF_C = `${"d".repeat(64)}#2`;
describe("SupplierState — initial state", () => {
it("starts in free state", () => {
@@ -102,25 +105,25 @@ describe("SupplierState.release()", () => {
it("transitions working → free", () => {
state.tryAcquire(ESCROW_REF_A);
state.release();
state.release(ESCROW_REF_A);
expect(state.snapshot().status).toBe("free");
});
it("clears currentEscrowRef on release", () => {
state.tryAcquire(ESCROW_REF_A);
state.release();
state.release(ESCROW_REF_A);
expect(state.snapshot().currentEscrowRef).toBeUndefined();
});
it("allows re-acquire after release", () => {
state.tryAcquire(ESCROW_REF_A);
state.release();
state.release(ESCROW_REF_A);
expect(state.tryAcquire(ESCROW_REF_B)).toBe(true);
});
it("re-acquire after release sets new currentEscrowRef", () => {
state.tryAcquire(ESCROW_REF_A);
state.release();
state.release(ESCROW_REF_A);
state.tryAcquire(ESCROW_REF_B);
expect(state.snapshot().currentEscrowRef).toBe(ESCROW_REF_B);
});
@@ -128,7 +131,7 @@ describe("SupplierState.release()", () => {
it("updates lastSeenIso on release", () => {
state.tryAcquire(ESCROW_REF_A);
const beforeRelease = state.snapshot().lastSeenIso;
state.release();
state.release(ESCROW_REF_A);
const afterRelease = state.snapshot().lastSeenIso;
expect(new Date(afterRelease).getTime()).not.toBeNaN();
expect(afterRelease >= beforeRelease).toBe(true);
@@ -199,4 +202,94 @@ describe("SupplierState.snapshot()", () => {
// State must remain working
expect(state.snapshot().status).toBe("working");
});
it("reports active/max session counts", () => {
const state = new SupplierState();
expect(state.snapshot().activeSessions).toBe(0);
expect(state.snapshot().maxSessions).toBe(1);
state.tryAcquire(ESCROW_REF_A);
expect(state.snapshot().activeSessions).toBe(1);
expect(state.snapshot().activeEscrowRefs).toEqual([ESCROW_REF_A]);
});
});
describe("SupplierState — capacity > 1", () => {
let state: SupplierState;
beforeEach(() => {
state = new SupplierState(2);
});
it("stays free until every slot is taken (free-until-full)", () => {
expect(state.tryAcquire(ESCROW_REF_A)).toBe(true);
expect(state.snapshot().status).toBe("free");
expect(state.snapshot().activeSessions).toBe(1);
expect("currentEscrowRef" in state.snapshot()).toBe(false);
expect(state.tryAcquire(ESCROW_REF_B)).toBe(true);
expect(state.snapshot().status).toBe("working");
expect(state.snapshot().currentEscrowRef).toBe(ESCROW_REF_A); // oldest
});
it("rejects the N+1th acquire", () => {
state.tryAcquire(ESCROW_REF_A);
state.tryAcquire(ESCROW_REF_B);
expect(state.tryAcquire(ESCROW_REF_C)).toBe(false);
});
it("rejects a duplicate ref while held", () => {
state.tryAcquire(ESCROW_REF_A);
expect(state.tryAcquire(ESCROW_REF_A)).toBe(false);
});
it("release frees exactly that ref's slot", () => {
state.tryAcquire(ESCROW_REF_A);
state.tryAcquire(ESCROW_REF_B);
state.release(ESCROW_REF_A);
expect(state.snapshot().status).toBe("free");
expect(state.snapshot().activeEscrowRefs).toEqual([ESCROW_REF_B]);
expect(state.tryAcquire(ESCROW_REF_C)).toBe(true);
});
it("release of an unheld ref is a no-op", () => {
state.tryAcquire(ESCROW_REF_A);
state.release(ESCROW_REF_B);
expect(state.snapshot().activeSessions).toBe(1);
});
it("markOffline clears all slots and refuses new acquires", () => {
state.tryAcquire(ESCROW_REF_A);
state.tryAcquire(ESCROW_REF_B);
state.markOffline();
expect(state.snapshot().status).toBe("offline");
expect(state.snapshot().activeSessions).toBe(0);
expect(state.tryAcquire(ESCROW_REF_C)).toBe(false);
});
it("rejects a non-positive capacity", () => {
expect(() => new SupplierState(0)).toThrow();
});
});
describe("SupplierState.walletMutex", () => {
it("serializes runs in submission order and survives rejection", async () => {
const state = new SupplierState(4);
const order: number[] = [];
const p1 = state.walletMutex.run(async () => {
await new Promise((r) => setTimeout(r, 20));
order.push(1);
});
const p2 = state.walletMutex.run(async () => {
order.push(2);
});
await Promise.all([p1, p2]);
expect(order).toEqual([1, 2]);
await state.walletMutex
.run(async () => {
throw new Error("boom");
})
.catch(() => undefined);
expect(await state.walletMutex.run(async () => 42)).toBe(42);
});
});

View File

@@ -4,11 +4,11 @@
*
* Coverage:
* - Ticker calls consolidate when idle, logs already-healthy
* - Ticker skips when the supplier state is busy
* - Ticker releases the lock around runConsolidateWallet (success + failure)
* - Consolidation runs even while session slots are held (wallet ops don't
* consume session slots) but queues behind a held walletMutex
* - Failure paths never leave the walletMutex poisoned
* - On-failure trigger fires consolidate the first time
* - On-failure trigger debounces back-to-back invocations
* - On-failure trigger respects the state lock
*
* The consolidate function is injected via deps so the test never touches
* a real Ogmios.
@@ -91,7 +91,7 @@ describe("walletHealth — ticker", () => {
ticker.stop();
});
it("skips when supplier state is busy", async () => {
it("runs even while a session slot is held (wallet ops don't consume slots)", async () => {
state.tryAcquire(ESCROW_REF);
consolidate.mockResolvedValue(healthyResult);
@@ -101,16 +101,39 @@ describe("walletHealth — ticker", () => {
);
await vi.advanceTimersByTimeAsync(1_000);
await vi.advanceTimersByTimeAsync(0);
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
expect(consolidate).not.toHaveBeenCalled();
expect(logLines).toContain("skip: supplier busy");
// The held session slot is untouched.
expect(state.snapshot().status).toBe("working");
expect(state.snapshot().activeEscrowRefs).toEqual([ESCROW_REF]);
ticker.stop();
});
it("releases the lock after a successful consolidate", async () => {
it("queues behind a held walletMutex instead of racing it", async () => {
consolidate.mockResolvedValue(healthyResult);
let releaseChainOp: () => void = () => undefined;
const chainOp = state.walletMutex.run(
() => new Promise<void>((resolve) => { releaseChainOp = resolve; }),
);
const ticker = startWalletHealthTicker(
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
{ intervalMs: 1_000, log },
);
await vi.advanceTimersByTimeAsync(1_000);
// The mutex is held by the fake chain op — consolidate must wait.
expect(consolidate).not.toHaveBeenCalled();
releaseChainOp();
await chainOp;
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
ticker.stop();
});
it("logs success after a real consolidate", async () => {
consolidate.mockResolvedValue(consolidatedResult);
const ticker = startWalletHealthTicker(
@@ -127,7 +150,7 @@ describe("walletHealth — ticker", () => {
ticker.stop();
});
it("releases the lock when consolidate throws", async () => {
it("does not poison the walletMutex when consolidate throws", async () => {
consolidate.mockRejectedValue(new Error("ogmios unreachable"));
const ticker = startWalletHealthTicker(
@@ -138,8 +161,9 @@ describe("walletHealth — ticker", () => {
await vi.advanceTimersByTimeAsync(1_000);
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
expect(state.snapshot().status).toBe("free");
expect(logLines.some((l) => l.includes("consolidate failed"))).toBe(true);
// The mutex still serves the next wallet op.
expect(await state.walletMutex.run(async () => "ok")).toBe("ok");
ticker.stop();
});
@@ -202,7 +226,7 @@ describe("walletHealth — triggerOnFailureConsolidate", () => {
expect(consolidate).toHaveBeenCalledTimes(1);
});
it("does not throw when supplier is busy — just skips", async () => {
it("runs even when session slots are held (queues on the wallet mutex)", async () => {
state.tryAcquire(ESCROW_REF);
consolidate.mockResolvedValue(healthyResult);
@@ -213,9 +237,8 @@ describe("walletHealth — triggerOnFailureConsolidate", () => {
),
).not.toThrow();
await Promise.resolve();
expect(consolidate).not.toHaveBeenCalled();
expect(logLines).toContain("skip: supplier busy");
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
// The held session slot is untouched.
expect(state.snapshot().status).toBe("working");
});
});