diff --git a/buyer/src/server.ts b/buyer/src/server.ts index 1ba13d6..ae85aaf 100644 --- a/buyer/src/server.ts +++ b/buyer/src/server.ts @@ -685,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", ); } diff --git a/deploy/mainnet/docker-compose.supplier-kimi-chat-2.yml b/deploy/mainnet/docker-compose.supplier-kimi-chat-2.yml new file mode 100644 index 0000000..fac05db --- /dev/null +++ b/deploy/mainnet/docker-compose.supplier-kimi-chat-2.yml @@ -0,0 +1,76 @@ +# 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: "150000" + 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 diff --git a/deploy/mainnet/docker-compose.supplier-kimi-chat-3.yml b/deploy/mainnet/docker-compose.supplier-kimi-chat-3.yml new file mode 100644 index 0000000..260e4cf --- /dev/null +++ b/deploy/mainnet/docker-compose.supplier-kimi-chat-3.yml @@ -0,0 +1,76 @@ +# 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: "150000" + 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 diff --git a/deploy/mainnet/docker-compose.supplier-kimi-chat.yml b/deploy/mainnet/docker-compose.supplier-kimi-chat.yml index f056b4c..ae1a751 100644 --- a/deploy/mainnet/docker-compose.supplier-kimi-chat.yml +++ b/deploy/mainnet/docker-compose.supplier-kimi-chat.yml @@ -49,8 +49,10 @@ 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: "150000" PORT: "8080" # Enable LiveOgmiosProvider so Claim/Submit actually broadcast. LIVE_CHAIN: "1" diff --git a/gateway/package.json b/gateway/package.json index 660b7f8..90b56bd 100644 --- a/gateway/package.json +++ b/gateway/package.json @@ -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:*", diff --git a/gateway/src/account/routes.ts b/gateway/src/account/routes.ts index 5f9357a..885cd21 100644 --- a/gateway/src/account/routes.ts +++ b/gateway/src/account/routes.ts @@ -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; const toAddress = body.to_address; diff --git a/gateway/src/cli/create-demo-key.ts b/gateway/src/cli/create-demo-key.ts new file mode 100644 index 0000000..8605d3d --- /dev/null +++ b/gateway/src/cli/create-demo-key.ts @@ -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 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(); diff --git a/gateway/src/config.ts b/gateway/src/config.ts index bb01dbd..ddfa579 100644 --- a/gateway/src/config.ts +++ b/gateway/src/config.ts @@ -39,6 +39,7 @@ export interface GatewayConfig { dbDir: string; signupRate: { max: number; windowMs: number }; keyRate: { max: number; windowMs: number }; + demoIpRate: { max: number; windowMs: number }; sweeperIntervalMs: number; walletHealthIntervalMs: number; sdkRegistryMax: number; @@ -111,6 +112,10 @@ export function loadConfig(env: Record): 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), walletHealthIntervalMs: posInt(env, "WALLET_HEALTH_INTERVAL_MS", 10 * 60 * 1000), sdkRegistryMax: posInt(env, "SDK_REGISTRY_MAX", 500), diff --git a/gateway/src/db/schema.ts b/gateway/src/db/schema.ts index 1524b17..6a0ade0 100644 --- a/gateway/src/db/schema.ts +++ b/gateway/src/db/schema.ts @@ -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); diff --git a/gateway/src/db/store.ts b/gateway/src/db/store.ts index 3010a8e..438e7e9 100644 --- a/gateway/src/db/store.ts +++ b/gateway/src/db/store.ts @@ -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): void { - this.sInsertKey.run({ ...row, label: row.label ?? null }); + insertKey(row: Omit & { 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); + } } diff --git a/gateway/src/middleware/rateLimit.ts b/gateway/src/middleware/rateLimit.ts index 6d47cf5..c4800dd 100644 --- a/gateway/src/middleware/rateLimit.ts +++ b/gateway/src/middleware/rateLimit.ts @@ -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); + }; } diff --git a/gateway/src/openai/affinity.ts b/gateway/src/openai/affinity.ts new file mode 100644 index 0000000..c975d36 --- /dev/null +++ b/gateway/src/openai/affinity.ts @@ -0,0 +1,67 @@ +/** + * 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 strict prefix of the incoming messages — + * the suffix (delta) is the next turn. Agentic tool loops extend the + * transcript monotonically, so affinity hits nearly always; history divergence + * (IDE compaction, regenerate, retry) just misses and costs one new session. + * + * Pure module — no I/O — so the matching rules are unit-testable. + */ + +import { chatMessagesEquivalent, 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[]; +} + +function isPrefix(mirror: ChatMessage[], incoming: ChatMessage[]): boolean { + if (mirror.length >= incoming.length) return false; + return mirror.every((m, i) => chatMessagesEquivalent(m, incoming[i])); +} + +function deltaRespondable(delta: ChatMessage[]): boolean { + const last = delta[delta.length - 1]; + return last !== undefined && (last.role === "user" || last.role === "tool"); +} + +/** + * Pick the open session whose mirror best matches the incoming history. + * 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: AffinityCandidate | null = null; + for (const cand of candidates) { + if (cand.model !== model) continue; + if (!isPrefix(cand.mirror, incoming)) continue; + if (!deltaRespondable(incoming.slice(cand.mirror.length))) continue; + if ( + best === null || + cand.mirror.length > best.mirror.length || + (cand.mirror.length === best.mirror.length && cand.lastUsedAt > best.lastUsedAt) + ) { + best = cand; + } + } + if (!best) return null; + return { sessionId: best.sessionId, delta: incoming.slice(best.mirror.length) }; +} diff --git a/gateway/src/openai/chatCompletions.ts b/gateway/src/openai/chatCompletions.ts index 7a90a75..65ef92f 100644 --- a/gateway/src/openai/chatCompletions.ts +++ b/gateway/src/openai/chatCompletions.ts @@ -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)); }); } diff --git a/gateway/src/openai/demoChat.ts b/gateway/src/openai/demoChat.ts new file mode 100644 index 0000000..19e5608 --- /dev/null +++ b/gateway/src/openai/demoChat.ts @@ -0,0 +1,279 @@ +/** + * 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, 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 { 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(); + +/** 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 flat fee when the demo executor closes its row (the + * supplier already Submitted; the sweeper Accepts the escrow afterwards). */ +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: 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); +} + +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 { + const last = parsed.messages[parsed.messages.length - 1]; + if (last.role !== "user" && last.role !== "tool") { + // Reject before any chain work — opening an escrow for an unanswerable + // request would burn a session fee for a guaranteed supplier 400. + throw badRequest("invalid_messages", "the last message must have role user or tool"); + } + + const id = genId(); + let keepalive: ReturnType | 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, parsed, 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 { + // 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) and replay the full + // history as its first turn. + const session = await ctx.mutex.run(() => openSessionCore(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 { + return await streamSupplierTurn( + deps, + session, + { messages: parsed.messages, tools: parsed.tools, tool_choice: parsed.toolChoice }, + sink.onToken, + ); + } 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; + } + }); +} diff --git a/gateway/src/openai/errors.ts b/gateway/src/openai/errors.ts index 493d2da..e8687c6 100644 --- a/gateway/src/openai/errors.ts +++ b/gateway/src/openai/errors.ts @@ -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): 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 => diff --git a/gateway/src/openai/models.ts b/gateway/src/openai/models.ts index 1e58c24..dd256f5 100644 --- a/gateway/src/openai/models.ts +++ b/gateway/src/openai/models.ts @@ -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)); }); } diff --git a/gateway/src/openai/sessions.ts b/gateway/src/openai/sessions.ts index 132355f..163f4f1 100644 --- a/gateway/src/openai/sessions.ts +++ b/gateway/src/openai/sessions.ts @@ -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(); +/** 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,17 +62,22 @@ 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 { +): Promise { const candidates = await selectCandidates({ indexerUrl: deps.config.indexerUrl, model, @@ -127,7 +138,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 +157,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 +193,88 @@ function foldSystem(sessionId: string, content: string, body: Record void, +): Promise { + 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, (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; + 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 +283,6 @@ async function streamTurn( res: Response, ): Promise { 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 +293,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, (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 +395,7 @@ async function closeSession( }); deps.store.setSessionState(session.id, "closed", Date.now()); - transcripts.delete(session.id); + dropSessionState(session.id); const prompt = result.receipt.prompt_tokens ?? 0; const completion = result.receipt.completion_tokens ?? 0; diff --git a/gateway/src/openai/shapes.ts b/gateway/src/openai/shapes.ts index 89d99b4..a6d82bb 100644 --- a/gateway/src/openai/shapes.ts +++ b/gateway/src/openai/shapes.ts @@ -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 { 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; - finishReason: "stop" | null; + delta: + | { role?: string; content?: string; tool_calls?: Array<{ index: number } & ToolCall> } + | Record; + finishReason: "stop" | "tool_calls" | null; usage?: Usage; vector?: VectorReceipt; }): Record { diff --git a/gateway/src/openai/transcripts.ts b/gateway/src/openai/transcripts.ts new file mode 100644 index 0000000..37f4458 --- /dev/null +++ b/gateway/src/openai/transcripts.ts @@ -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(); + +const sessionLocks = new Map(); + +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); +} diff --git a/gateway/src/openai/validate.ts b/gateway/src/openai/validate.ts index 1616997..cf7ccdb 100644 --- a/gateway/src/openai/validate.ts +++ b/gateway/src/openai/validate.ts @@ -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; - 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 }; } diff --git a/gateway/src/routing/selectSupplier.ts b/gateway/src/routing/selectSupplier.ts index a5f61eb..7da52ed 100644 --- a/gateway/src/routing/selectSupplier.ts +++ b/gateway/src/routing/selectSupplier.ts @@ -101,19 +101,27 @@ export async function selectCandidates(opts: SelectSupplierOpts): Promise { 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(); 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(); } diff --git a/gateway/src/server.ts b/gateway/src/server.ts index a82b255..4e542fd 100644 --- a/gateway/src/server.ts +++ b/gateway/src/server.ts @@ -10,7 +10,7 @@ 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"; @@ -69,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)); diff --git a/gateway/src/sweeper.ts b/gateway/src/sweeper.ts index e3d16e5..a7933a6 100644 --- a/gateway/src/sweeper.ts +++ b/gateway/src/sweeper.ts @@ -10,26 +10,77 @@ * 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". */ +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 } 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; function log(msg: string): void { // eslint-disable-next-line no-console console.log(`[gateway:sweeper] ${msg}`); } +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})`); +} + export async function runSweepOnce(deps: GatewayDeps): Promise { 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 +88,53 @@ export async function runSweepOnce(deps: GatewayDeps): Promise { 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(); + 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; 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(() => acceptAndConfirm(deps.chain, ctx.walletKey, ref)); + log(`accepted stranded Submitted escrow ${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 (chatSession && (await deps.chain.queryUtxo(ref)) === null) { + 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; + if (ageMs > reclaimAfterMs) { + await ctx.mutex.run(() => reclaimAndConfirm(deps.chain, ctx.walletKey, ref)); + log(`reclaimed stranded ${row.state} escrow ${row.utxo_ref}`); + if (chatSession) { + // 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 +142,6 @@ export async function runSweepOnce(deps: GatewayDeps): Promise { } } } - - // 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. */ diff --git a/gateway/test/affinity.test.ts b/gateway/test/affinity.test.ts new file mode 100644 index 0000000..e09917c --- /dev/null +++ b/gateway/test/affinity.test.ts @@ -0,0 +1,81 @@ +/** + * affinity.test.ts — transcript-prefix session matching rules. + */ + +import { describe, it, expect } from "vitest"; +import { matchSessionPrefix, 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"); + }); +}); diff --git a/gateway/test/http.test.ts b/gateway/test/http.test.ts index a7eec75..f3afb46 100644 --- a/gateway/test/http.test.ts +++ b/gateway/test/http.test.ts @@ -1,23 +1,32 @@ -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 { 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"] }, +): 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 +35,13 @@ 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, 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 +114,238 @@ 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, 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. */ +function demoFetch(messageResponses: Array<() => Response>) { + const messageCalls: Array> = []; + let i = 0; + const fetchFn = (async (url: unknown, init?: { body?: string }) => { + const u = String(url); + if (u.includes("/v1/chat/message")) { + messageCalls.push(init?.body ? (JSON.parse(init.body) as Record) : {}); + const make = messageResponses[Math.min(i, messageResponses.length - 1)]; + i += 1; + return make(); + } + if (u.includes("/suppliers")) return new Response(JSON.stringify(SUPPLIERS), { status: 200 }); + return new Response("[]", { status: 200 }); + }) as unknown as typeof globalThis.fetch; + return { fetchFn, messageCalls }; +} + +/** 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> = [{ 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 that do not end with user/tool 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(startChat).not.toHaveBeenCalled(); + }); + + 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); + }); +}); diff --git a/gateway/test/migration.test.ts b/gateway/test/migration.test.ts new file mode 100644 index 0000000..7038524 --- /dev/null +++ b/gateway/test/migration.test.ts @@ -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); + }); +}); diff --git a/gateway/test/unit.test.ts b/gateway/test/unit.test.ts index 5532d0f..5ac81a9 100644 --- a/gateway/test/unit.test.ts +++ b/gateway/test/unit.test.ts @@ -84,6 +84,10 @@ 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(); @@ -104,6 +108,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 +156,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"); }); diff --git a/packages/shared/src/tx/chatMessage.ts b/packages/shared/src/tx/chatMessage.ts new file mode 100644 index 0000000..e39b549 --- /dev/null +++ b/packages/shared/src/tx/chatMessage.ts @@ -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; + const fn = tc.function as Record | 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; + 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; +} + +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) + ); +} diff --git a/packages/shared/src/tx/index.ts b/packages/shared/src/tx/index.ts index 0fc4cf7..6e8d688 100644 --- a/packages/shared/src/tx/index.ts +++ b/packages/shared/src/tx/index.ts @@ -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 } 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"; diff --git a/packages/shared/src/tx/types.ts b/packages/shared/src/tx/types.ts index b8930a1..1e45d13 100644 --- a/packages/shared/src/tx/types.ts +++ b/packages/shared/src/tx/types.ts @@ -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; } /** diff --git a/supplier/src/chatSession.ts b/supplier/src/chatSession.ts index 50173dd..8f3a67c 100644 --- a/supplier/src/chatSession.ts +++ b/supplier/src/chatSession.ts @@ -92,14 +92,22 @@ 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(); diff --git a/supplier/src/openai.ts b/supplier/src/openai.ts index 6f333be..3d2eb88 100644 --- a/supplier/src/openai.ts +++ b/supplier/src/openai.ts @@ -26,13 +26,20 @@ * - 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 ` * header is sent. Omit (or pass "") for endpoints that don't require auth, @@ -60,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"; @@ -204,7 +215,7 @@ export async function callOpenAiStream( params: CallOpenAiParams, onToken: (delta: string) => void, ): Promise { - const { baseUrl, model, messages, timeoutMs, apiKey, maxTokens, disableReasoning } = params; + const { baseUrl, model, messages, timeoutMs, apiKey, maxTokens, disableReasoning, tools, toolChoice } = params; const url = `${baseUrl}/v1/chat/completions`; const payload: Record = { model, @@ -214,6 +225,10 @@ export async function callOpenAiStream( }; 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 = { @@ -268,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(); + + 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; + 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 | 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(); @@ -284,14 +320,15 @@ export async function callOpenAiStream( const obj = parsed as Record; const choices = obj.choices; if (Array.isArray(choices) && choices.length > 0) { - const delta = (choices[0] as Record)?.delta as - | Record - | undefined; + const choice = choices[0] as Record; + const delta = choice?.delta as Record | 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") { @@ -336,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; } diff --git a/supplier/src/server.ts b/supplier/src/server.ts index d0823e6..ec9401e 100644 --- a/supplier/src/server.ts +++ b/supplier/src/server.ts @@ -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"; @@ -825,16 +831,46 @@ 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); + deps.chatSessions.appendMessages(escrowRefStr, delta); res.status(200); res.setHeader("Content-Type", "text/event-stream"); @@ -858,14 +894,30 @@ function makeChatSessionHandlers(deps: ResolvedDeps) { 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) { const errMsg = err instanceof Error ? err.message : String(err); sse({ type: "error", message: errMsg }); diff --git a/tests/unit/shared-chat-message.test.ts b/tests/unit/shared-chat-message.test.ts new file mode 100644 index 0000000..4a6b027 --- /dev/null +++ b/tests/unit/shared-chat-message.test.ts @@ -0,0 +1,96 @@ +/** + * 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 } 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); + }); +}); diff --git a/tests/unit/supplier-openai-stream.test.ts b/tests/unit/supplier-openai-stream.test.ts new file mode 100644 index 0000000..210353c --- /dev/null +++ b/tests/unit/supplier-openai-stream.test.ts @@ -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"); + }); +});