feat(gateway,supplier,shared): shared demo key with session-backed OpenAI endpoint + tool-calling
MVP demo: one operator-funded API key anyone can plug into an IDE/harness,
running real escrow-per-session marketplace conversations (llm.chat.v1)
behind the standard OpenAI /chat/completions wire shape.
- shared: ChatMessage gains tool role + tool_calls; normalizeChatMessage/
chatMessagesEquivalent keep supplier transcript and gateway mirror
hash-identical for receipts (field presence is part of the identity)
- supplier: /v1/chat/message accepts {messages,tools,tool_choice} alongside
legacy {content}; callOpenAiStream forwards tools, accumulates fragmented
delta.tool_calls, allows content-less tool turns; done frame carries the
assistant message verbatim + finish_reason + usage
- gateway: demo flag on api_keys (additive migration), create-demo-key CLI,
withdraw 403 + hidden deposit address for demo keys, per-IP demo rate
limit (demo keys skip the shared per-key limit), models filtered to chat
capability
- gateway executor (demoChat.ts): transcript-prefix affinity maps stateless
calls onto open sessions, forwards only unseen deltas, transparently
reopens on supplier idle-close; never falls back to one-shot (its settle
path assumes <=1 in-flight escrow per wallet)
- mutex scope: key mutex held only for chain work (open/close); off-chain
turns serialize on per-session locks so replicas serve users concurrently
- sweeper: chat-aware reclaim guard (35min vs 11min) and abandoned-session
cleanup; terminal state confirmed via on-chain queryUtxo (indexer keeps
lineage rows at their old state forever)
- deploy: kimi-chat replicas 2+3 compose files; chat idle timeout 2.5min
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -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",
|
||||
);
|
||||
}
|
||||
|
||||
76
deploy/mainnet/docker-compose.supplier-kimi-chat-2.yml
Normal file
76
deploy/mainnet/docker-compose.supplier-kimi-chat-2.yml
Normal file
@@ -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
|
||||
76
deploy/mainnet/docker-compose.supplier-kimi-chat-3.yml
Normal file
76
deploy/mainnet/docker-compose.supplier-kimi-chat-3.yml
Normal file
@@ -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
|
||||
@@ -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"
|
||||
|
||||
@@ -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:*",
|
||||
|
||||
@@ -15,7 +15,7 @@ import { seal } from "../crypto/seal.js";
|
||||
import { hashApiKey, requireKey } from "../middleware/apiKeyAuth.js";
|
||||
import { asyncHandler } from "../middleware/http.js";
|
||||
import { totalLovelace, hasCollateral } from "../onchain/preflight.js";
|
||||
import { badRequest } from "../openai/errors.js";
|
||||
import { badRequest, forbidden } from "../openai/errors.js";
|
||||
|
||||
const ACTIVE_ESCROW_STATES = new Set(["Open", "Claimed", "Submitted"]);
|
||||
|
||||
@@ -89,7 +89,10 @@ export function makeAccountHandler(deps: GatewayDeps) {
|
||||
|
||||
res.status(200).json({
|
||||
key_prefix: keyRow.key_prefix,
|
||||
deposit_address: keyRow.deposit_address,
|
||||
// The shared demo key's wallet is operator-funded; hiding the deposit
|
||||
// address keeps strangers from confusing it with their own wallet.
|
||||
deposit_address: keyRow.demo ? null : keyRow.deposit_address,
|
||||
...(keyRow.demo ? { demo: true } : {}),
|
||||
balance: {
|
||||
available_lovelace: available.toString(),
|
||||
locked_in_escrow_lovelace: lockedInEscrow.toString(),
|
||||
@@ -118,6 +121,11 @@ export function makeAccountHandler(deps: GatewayDeps) {
|
||||
export function makeWithdrawHandler(deps: GatewayDeps) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const keyRow = requireKey(req);
|
||||
if (keyRow.demo) {
|
||||
// The demo key is shared publicly; without this, anyone holding it
|
||||
// could drain the operator-funded wallet to their own address.
|
||||
throw forbidden("demo_key_restricted", "withdrawals are disabled for the shared demo key");
|
||||
}
|
||||
const body = (req.body ?? {}) as Record<string, unknown>;
|
||||
|
||||
const toAddress = body.to_address;
|
||||
|
||||
76
gateway/src/cli/create-demo-key.ts
Normal file
76
gateway/src/cli/create-demo-key.ts
Normal file
@@ -0,0 +1,76 @@
|
||||
/**
|
||||
* gateway/src/cli/create-demo-key.ts — mint the shared demo API key.
|
||||
*
|
||||
* Same custodial machinery as POST /signup, but sets demo=1: withdraw is
|
||||
* blocked, /account hides the deposit address, chat completions run through
|
||||
* the session-backed demo executor, and per-IP rate limits apply. Public
|
||||
* signup can never mint a demo key — this CLI is the only path.
|
||||
*
|
||||
* Run inside the gateway container (hits the volume-mounted DB):
|
||||
* docker exec <gateway> pnpm --filter @marketplace/gateway create-demo-key
|
||||
* Or locally:
|
||||
* GATEWAY_MASTER_KEY=<64hex> GATEWAY_DB_DIR=./data/gateway NETWORK_ID=1 \
|
||||
* pnpm --filter @marketplace/gateway create-demo-key [--label shared-demo]
|
||||
*
|
||||
* The raw key is printed ONCE and never stored (only its sha256).
|
||||
*/
|
||||
|
||||
import { randomBytes, randomUUID } from "crypto";
|
||||
import { GatewayStore } from "../db/store.js";
|
||||
import { genPrivKeyHex, deriveWalletKey } from "../wallet.js";
|
||||
import { seal } from "../crypto/seal.js";
|
||||
import { hashApiKey } from "../middleware/apiKeyAuth.js";
|
||||
|
||||
const HEX64 = /^[0-9a-fA-F]{64}$/;
|
||||
|
||||
function arg(name: string): string | undefined {
|
||||
const i = process.argv.indexOf(`--${name}`);
|
||||
return i >= 0 ? process.argv[i + 1] : undefined;
|
||||
}
|
||||
|
||||
function main(): void {
|
||||
const masterKey = process.env.GATEWAY_MASTER_KEY;
|
||||
if (!masterKey || !HEX64.test(masterKey)) {
|
||||
throw new Error("create-demo-key: GATEWAY_MASTER_KEY must be 64 hex chars");
|
||||
}
|
||||
const dbDir = arg("db") ?? process.env.GATEWAY_DB_DIR ?? "./data/gateway";
|
||||
const networkId: 0 | 1 = process.env.NETWORK_ID === "1" ? 1 : 0;
|
||||
const label = arg("label") ?? "shared-demo";
|
||||
|
||||
const privHex = genPrivKeyHex();
|
||||
const walletKey = deriveWalletKey(privHex, networkId);
|
||||
const sealed = seal(privHex, masterKey);
|
||||
|
||||
const rawKey = `vk_${networkId === 1 ? "live" : "test"}_${randomBytes(24).toString("hex")}`;
|
||||
|
||||
const store = new GatewayStore(dbDir);
|
||||
store.insertKey({
|
||||
id: randomUUID(),
|
||||
key_hash: hashApiKey(rawKey),
|
||||
key_prefix: rawKey.slice(0, 12),
|
||||
label,
|
||||
wallet_pkh: walletKey.pubKeyHash,
|
||||
deposit_address: walletKey.address,
|
||||
enc_priv_nonce: sealed.nonce,
|
||||
enc_priv_ct: sealed.ct,
|
||||
enc_priv_tag: sealed.tag,
|
||||
master_key_version: 1,
|
||||
created_at: Date.now(),
|
||||
demo: 1,
|
||||
});
|
||||
|
||||
// eslint-disable-next-line no-console
|
||||
console.log(
|
||||
[
|
||||
`[gateway] demo key created (label=${label}, network=${networkId === 1 ? "mainnet" : "testnet"})`,
|
||||
``,
|
||||
` api_key: ${rawKey}`,
|
||||
` deposit_address: ${walletKey.address}`,
|
||||
``,
|
||||
`Save the api_key now — it is shown only once. Fund the deposit address`,
|
||||
`with the demo float (withdrawals are disabled for this key).`,
|
||||
].join("\n"),
|
||||
);
|
||||
}
|
||||
|
||||
main();
|
||||
@@ -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<string, string | undefined>): GatewayConf
|
||||
max: posInt(env, "KEY_RATE_MAX", 60),
|
||||
windowMs: posInt(env, "KEY_RATE_WINDOW_MS", 60 * 1000),
|
||||
},
|
||||
demoIpRate: {
|
||||
max: posInt(env, "DEMO_IP_RATE_MAX", 20),
|
||||
windowMs: posInt(env, "DEMO_IP_RATE_WINDOW_MS", 60 * 1000),
|
||||
},
|
||||
sweeperIntervalMs: posInt(env, "SWEEPER_INTERVAL_MS", 60 * 1000),
|
||||
walletHealthIntervalMs: posInt(env, "WALLET_HEALTH_INTERVAL_MS", 10 * 60 * 1000),
|
||||
sdkRegistryMax: posInt(env, "SDK_REGISTRY_MAX", 500),
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -24,6 +24,8 @@ export interface ApiKeyRow {
|
||||
master_key_version: number;
|
||||
created_at: number;
|
||||
disabled: number;
|
||||
/** 1 = shared demo key: withdraw blocked, session-backed completions, per-IP limits. */
|
||||
demo: number;
|
||||
}
|
||||
|
||||
export interface UsageRow {
|
||||
@@ -81,13 +83,20 @@ export class GatewayStore {
|
||||
this.db.pragma("journal_mode = WAL");
|
||||
this.db.exec(CREATE_TABLES_SQL);
|
||||
|
||||
// Additive migration for DBs created before the demo flag existed
|
||||
// (CREATE TABLE IF NOT EXISTS won't alter an existing table).
|
||||
const keyCols = this.db.pragma("table_info(api_keys)") as Array<{ name: string }>;
|
||||
if (!keyCols.some((c) => c.name === "demo")) {
|
||||
this.db.exec("ALTER TABLE api_keys ADD COLUMN demo INTEGER NOT NULL DEFAULT 0");
|
||||
}
|
||||
|
||||
this.sInsertKey = this.db.prepare(`
|
||||
INSERT INTO api_keys (
|
||||
id, key_hash, key_prefix, label, wallet_pkh, deposit_address,
|
||||
enc_priv_nonce, enc_priv_ct, enc_priv_tag, master_key_version, created_at, disabled
|
||||
enc_priv_nonce, enc_priv_ct, enc_priv_tag, master_key_version, created_at, disabled, demo
|
||||
) VALUES (
|
||||
@id, @key_hash, @key_prefix, @label, @wallet_pkh, @deposit_address,
|
||||
@enc_priv_nonce, @enc_priv_ct, @enc_priv_tag, @master_key_version, @created_at, 0
|
||||
@enc_priv_nonce, @enc_priv_ct, @enc_priv_tag, @master_key_version, @created_at, 0, @demo
|
||||
)
|
||||
`);
|
||||
this.sGetKeyByHash = this.db.prepare(`SELECT * FROM api_keys WHERE key_hash = @key_hash`);
|
||||
@@ -133,8 +142,8 @@ export class GatewayStore {
|
||||
this.sListOpenSessions = this.db.prepare(`SELECT * FROM sessions WHERE state = 'open'`);
|
||||
}
|
||||
|
||||
insertKey(row: Omit<ApiKeyRow, "disabled">): void {
|
||||
this.sInsertKey.run({ ...row, label: row.label ?? null });
|
||||
insertKey(row: Omit<ApiKeyRow, "disabled" | "demo"> & { demo?: number }): void {
|
||||
this.sInsertKey.run({ ...row, label: row.label ?? null, demo: row.demo ?? 0 });
|
||||
}
|
||||
|
||||
getKeyByHash(keyHash: string): ApiKeyRow | undefined {
|
||||
@@ -193,4 +202,8 @@ export class GatewayStore {
|
||||
listOpenSessions(): SessionRow[] {
|
||||
return this.sListOpenSessions.all() as SessionRow[];
|
||||
}
|
||||
|
||||
listOpenSessionsByKey(keyId: string): SessionRow[] {
|
||||
return this.listOpenSessions().filter((s) => s.key_id === keyId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
};
|
||||
}
|
||||
|
||||
67
gateway/src/openai/affinity.ts
Normal file
67
gateway/src/openai/affinity.ts
Normal file
@@ -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) };
|
||||
}
|
||||
@@ -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));
|
||||
});
|
||||
}
|
||||
|
||||
279
gateway/src/openai/demoChat.ts
Normal file
279
gateway/src/openai/demoChat.ts
Normal file
@@ -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<string, DemoSessionMeta>();
|
||||
|
||||
/** Forget a session (closed/settled elsewhere, e.g. by the sweeper). */
|
||||
export function dropDemoSession(sessionId: string): void {
|
||||
demoSessions.delete(sessionId);
|
||||
}
|
||||
|
||||
function candidatesFor(keyId: string, model: string): AffinityCandidate[] {
|
||||
const out: AffinityCandidate[] = [];
|
||||
for (const [sessionId, meta] of demoSessions) {
|
||||
if (meta.keyId !== keyId || meta.model !== model) continue;
|
||||
const mirror = transcripts.get(sessionId);
|
||||
if (!mirror) continue;
|
||||
out.push({ sessionId, model: meta.model, mirror, lastUsedAt: meta.lastUsedAt });
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Record the session's 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<void> {
|
||||
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<typeof setInterval> | undefined;
|
||||
let roleSent = false;
|
||||
if (parsed.stream) {
|
||||
res.status(200);
|
||||
res.setHeader("Content-Type", "text/event-stream");
|
||||
res.setHeader("Cache-Control", "no-cache, no-transform");
|
||||
res.setHeader("Connection", "keep-alive");
|
||||
res.setHeader("X-Accel-Buffering", "no");
|
||||
res.write(": connected\n\n");
|
||||
keepalive = setInterval(() => {
|
||||
try {
|
||||
res.write(": keepalive\n\n");
|
||||
} catch {
|
||||
/* client gone */
|
||||
}
|
||||
}, 10_000);
|
||||
}
|
||||
|
||||
const ensureRole = (): void => {
|
||||
if (roleSent) return;
|
||||
roleSent = true;
|
||||
res.write(sseData(buildChunk({ id, model: parsed.model, delta: { role: "assistant" }, finishReason: null })));
|
||||
};
|
||||
const sink: StreamSink = {
|
||||
onToken: (token) => {
|
||||
if (!parsed.stream) return;
|
||||
ensureRole();
|
||||
res.write(sseData(buildChunk({ id, model: parsed.model, delta: { content: token }, finishReason: null })));
|
||||
},
|
||||
emitToolCalls: (toolCalls) => {
|
||||
if (!parsed.stream) return;
|
||||
ensureRole();
|
||||
res.write(
|
||||
sseData(
|
||||
buildChunk({
|
||||
id,
|
||||
model: parsed.model,
|
||||
delta: { tool_calls: toolCalls.map((tc, index) => ({ index, ...tc })) },
|
||||
finishReason: null,
|
||||
}),
|
||||
),
|
||||
);
|
||||
},
|
||||
};
|
||||
|
||||
try {
|
||||
const result = await runWithAffinity(deps, keyRow, ctx, 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<TurnResult> {
|
||||
// Affinity: try matching sessions until one serves the turn. The match runs
|
||||
// synchronously (no await between match and lock acquire, so the candidate
|
||||
// set can't shift underneath us); the prefix is re-checked inside the lock
|
||||
// because a concurrent request may have advanced the session first.
|
||||
for (let attempt = 0; attempt < 3; attempt++) {
|
||||
const match = matchSessionPrefix(candidatesFor(keyRow.id, parsed.model), parsed.messages, parsed.model);
|
||||
if (!match) break;
|
||||
|
||||
const outcome: LockedTurnOutcome = await getSessionLock(match.sessionId).run(async () => {
|
||||
const meta = demoSessions.get(match.sessionId);
|
||||
const mirror = transcripts.get(match.sessionId);
|
||||
const session = deps.store.getSession(match.sessionId);
|
||||
if (!meta || !mirror || !session || session.state !== "open") {
|
||||
dropDemoSession(match.sessionId);
|
||||
return { kind: "retry" };
|
||||
}
|
||||
const recheck = matchSessionPrefix(
|
||||
[{ sessionId: session.id, model: meta.model, mirror, lastUsedAt: meta.lastUsedAt }],
|
||||
parsed.messages,
|
||||
parsed.model,
|
||||
);
|
||||
if (!recheck) return { kind: "retry" };
|
||||
|
||||
try {
|
||||
const result = await streamSupplierTurn(
|
||||
deps,
|
||||
session,
|
||||
{ messages: recheck.delta, tools: parsed.tools, tool_choice: parsed.toolChoice },
|
||||
sink.onToken,
|
||||
);
|
||||
meta.lastUsedAt = Date.now();
|
||||
return { kind: "done", result };
|
||||
} catch (err) {
|
||||
if (err instanceof SessionGoneError) {
|
||||
// Supplier idle-closed (or restarted). Close our row, record the
|
||||
// fee (the escrow settles via the sweeper) and reopen fresh.
|
||||
markSessionGone(deps, session);
|
||||
return { kind: "gone" };
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
|
||||
if (outcome.kind === "done") return outcome.result;
|
||||
if (outcome.kind === "gone") break;
|
||||
// retry: candidate vanished/advanced — re-match against current state.
|
||||
}
|
||||
|
||||
// Miss: open a new session (chain work → key mutex) 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;
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -36,6 +36,8 @@ export const unauthorized = (message = "invalid API key"): GatewayError =>
|
||||
new GatewayError(401, "authentication_error", "invalid_api_key", message);
|
||||
export const paymentRequired = (message: string, extra?: Record<string, unknown>): GatewayError =>
|
||||
new GatewayError(402, "invalid_request_error", "insufficient_funds", message, extra);
|
||||
export const forbidden = (code: string, message: string): GatewayError =>
|
||||
new GatewayError(403, "invalid_request_error", code, message);
|
||||
export const notFound = (code: string, message: string): GatewayError =>
|
||||
new GatewayError(404, "invalid_request_error", code, message);
|
||||
export const rateLimited = (message: string): GatewayError =>
|
||||
|
||||
@@ -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));
|
||||
});
|
||||
}
|
||||
|
||||
@@ -15,8 +15,9 @@
|
||||
|
||||
import { randomUUID } from "crypto";
|
||||
import type { Request, Response } from "express";
|
||||
import { TxConstructionError, type ChatMessage } from "@marketplace/shared/tx";
|
||||
import { TxConstructionError, normalizeChatMessage, type ChatMessage } from "@marketplace/shared/tx";
|
||||
import { SupplierError } from "@marketplace/buyer/sdk";
|
||||
import { transcripts, getSessionLock, dropSessionState } from "./transcripts.js";
|
||||
import type { GatewayDeps } from "../deps.js";
|
||||
import type { ApiKeyRow, GatewayStore, SessionRow } from "../db/store.js";
|
||||
import type { KeyContext } from "../sdk/registry.js";
|
||||
@@ -36,10 +37,15 @@ import {
|
||||
type Usage,
|
||||
} from "./shapes.js";
|
||||
|
||||
const CAPABILITY = "llm.chat.v1";
|
||||
export const CAPABILITY = "llm.chat.v1";
|
||||
|
||||
/** In-memory per-session transcript mirror (sessionId → messages). */
|
||||
const transcripts = new Map<string, ChatMessage[]>();
|
||||
/** Thrown when the supplier no longer knows the session (idle-closed/restart). */
|
||||
export class SessionGoneError extends Error {
|
||||
constructor(sessionId: string) {
|
||||
super(`supplier no longer has session ${sessionId}`);
|
||||
this.name = "SessionGoneError";
|
||||
}
|
||||
}
|
||||
|
||||
function refStr(ref: { txHash: string; index: number }): string {
|
||||
return `${ref.txHash}#${ref.index}`;
|
||||
@@ -56,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<void> {
|
||||
): Promise<SessionRow> {
|
||||
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<string, unk
|
||||
return content;
|
||||
}
|
||||
|
||||
/** One turn's payload to the supplier: legacy single user string, or a
|
||||
* multi-message delta (e.g. tool results) with optional tools passthrough. */
|
||||
export type TurnPayload =
|
||||
| { content: string }
|
||||
| { messages: ChatMessage[]; tools?: unknown[]; tool_choice?: unknown };
|
||||
|
||||
export interface TurnResult {
|
||||
assistantMessage: ChatMessage;
|
||||
finishReason: "stop" | "tool_calls";
|
||||
usage: { prompt_tokens: number; completion_tokens: number };
|
||||
}
|
||||
|
||||
/**
|
||||
* Drive one off-chain turn against the session's supplier and mirror the
|
||||
* result into the session transcript. Caller must hold the session lock.
|
||||
* Throws SessionGoneError when the supplier no longer knows the session
|
||||
* (idle-closed or restarted) so callers can transparently reopen.
|
||||
*/
|
||||
export async function streamSupplierTurn(
|
||||
deps: GatewayDeps,
|
||||
session: SessionRow,
|
||||
payload: TurnPayload,
|
||||
onToken?: (token: string) => void,
|
||||
): Promise<TurnResult> {
|
||||
const upstream = await deps.fetchFn(`${session.supplier_base_url.replace(/\/+$/, "")}/v1/chat/message`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json", accept: "text/event-stream", "X-Escrow-Ref": session.escrow_ref },
|
||||
body: JSON.stringify(payload),
|
||||
});
|
||||
if (upstream.status === 404) {
|
||||
await upstream.text().catch(() => "");
|
||||
throw new SessionGoneError(session.id);
|
||||
}
|
||||
if (!upstream.ok || !upstream.body) {
|
||||
const detail = await upstream.text().catch(() => "");
|
||||
throw toGatewayError(new SupplierError("supplier_http_error", { status: upstream.status, message: detail.slice(0, 200) }));
|
||||
}
|
||||
|
||||
let assistant = "";
|
||||
let errored: string | undefined;
|
||||
let doneMessage: ChatMessage | null = null;
|
||||
let doneFinish: string | undefined;
|
||||
let usage = { prompt_tokens: 0, completion_tokens: 0 };
|
||||
await readSupplierSse(upstream.body as ReadableStream<Uint8Array>, (frame) => {
|
||||
if (frame.type === "token" && typeof frame.value === "string") {
|
||||
assistant += frame.value;
|
||||
onToken?.(frame.value);
|
||||
} else if (frame.type === "done") {
|
||||
// Upgraded suppliers echo the final assistant message verbatim; mirror
|
||||
// that exact object so both transcripts stay hash-identical.
|
||||
doneMessage = normalizeChatMessage(frame.message);
|
||||
if (typeof frame.finish_reason === "string") doneFinish = frame.finish_reason;
|
||||
if (frame.usage && typeof frame.usage === "object") {
|
||||
const u = frame.usage as Record<string, unknown>;
|
||||
usage = {
|
||||
prompt_tokens: typeof u.prompt_tokens === "number" ? u.prompt_tokens : 0,
|
||||
completion_tokens: typeof u.completion_tokens === "number" ? u.completion_tokens : 0,
|
||||
};
|
||||
}
|
||||
} else if (frame.type === "error") {
|
||||
errored = typeof frame.message === "string" ? frame.message : "supplier stream error";
|
||||
}
|
||||
});
|
||||
|
||||
if (errored !== undefined) {
|
||||
throw toGatewayError(new SupplierError("upstream_error", { message: errored }));
|
||||
}
|
||||
|
||||
const assistantMessage: ChatMessage = doneMessage ?? { role: "assistant", content: assistant };
|
||||
const finishReason: TurnResult["finishReason"] =
|
||||
doneFinish === "tool_calls" || (assistantMessage.tool_calls?.length ?? 0) > 0 ? "tool_calls" : "stop";
|
||||
|
||||
// Mirror the turn for endChat's response_hash verification / affinity matching.
|
||||
const mirror = transcripts.get(session.id) ?? [];
|
||||
if ("content" in payload) mirror.push({ role: "user", content: payload.content });
|
||||
else mirror.push(...payload.messages);
|
||||
mirror.push(assistantMessage);
|
||||
transcripts.set(session.id, mirror);
|
||||
|
||||
return { assistantMessage, finishReason, usage };
|
||||
}
|
||||
|
||||
async function streamTurn(
|
||||
deps: GatewayDeps,
|
||||
session: SessionRow,
|
||||
@@ -184,15 +283,6 @@ async function streamTurn(
|
||||
res: Response,
|
||||
): Promise<void> {
|
||||
const id = genId();
|
||||
const upstream = await deps.fetchFn(`${session.supplier_base_url.replace(/\/+$/, "")}/v1/chat/message`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json", accept: "text/event-stream", "X-Escrow-Ref": session.escrow_ref },
|
||||
body: JSON.stringify({ content }),
|
||||
});
|
||||
if (!upstream.ok || !upstream.body) {
|
||||
const detail = await upstream.text().catch(() => "");
|
||||
throw toGatewayError(new SupplierError("supplier_http_error", { status: upstream.status, message: detail.slice(0, 200) }));
|
||||
}
|
||||
|
||||
if (stream) {
|
||||
res.status(200);
|
||||
@@ -203,46 +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<Uint8Array>, (frame) => {
|
||||
if (frame.type === "token" && typeof frame.value === "string") {
|
||||
assistant += frame.value;
|
||||
if (stream) res.write(sseData(buildChunk({ id, model: session.model, delta: { content: frame.value }, finishReason: null })));
|
||||
} else if (frame.type === "error") {
|
||||
errored = typeof frame.message === "string" ? frame.message : "supplier stream error";
|
||||
}
|
||||
let result: TurnResult;
|
||||
try {
|
||||
result = await streamSupplierTurn(deps, session, { content }, (token) => {
|
||||
if (stream) res.write(sseData(buildChunk({ id, model: session.model, delta: { content: token }, finishReason: null })));
|
||||
});
|
||||
|
||||
if (errored !== undefined) {
|
||||
} catch (err) {
|
||||
if (err instanceof SessionGoneError) {
|
||||
err = toGatewayError(new SupplierError("supplier_http_error", { status: 404, message: err.message }));
|
||||
}
|
||||
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;
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
|
||||
import { randomBytes } from "crypto";
|
||||
import type { Receipt } from "@marketplace/shared/receipt";
|
||||
import type { ToolCall } from "@marketplace/shared/tx";
|
||||
|
||||
export function genId(): string {
|
||||
return `chatcmpl-${randomBytes(16).toString("hex")}`;
|
||||
@@ -40,6 +41,8 @@ export function buildChatCompletion(args: {
|
||||
model: string;
|
||||
content: string;
|
||||
usage: Usage;
|
||||
toolCalls?: ToolCall[];
|
||||
finishReason?: "stop" | "tool_calls";
|
||||
vector?: VectorReceipt;
|
||||
}): Record<string, unknown> {
|
||||
return {
|
||||
@@ -50,8 +53,12 @@ export function buildChatCompletion(args: {
|
||||
choices: [
|
||||
{
|
||||
index: 0,
|
||||
message: { role: "assistant", content: args.content },
|
||||
finish_reason: "stop",
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: args.content,
|
||||
...(args.toolCalls?.length ? { tool_calls: args.toolCalls } : {}),
|
||||
},
|
||||
finish_reason: args.finishReason ?? "stop",
|
||||
},
|
||||
],
|
||||
usage: args.usage,
|
||||
@@ -63,8 +70,10 @@ export function buildChatCompletion(args: {
|
||||
export function buildChunk(args: {
|
||||
id: string;
|
||||
model: string;
|
||||
delta: { role?: string; content?: string } | Record<string, never>;
|
||||
finishReason: "stop" | null;
|
||||
delta:
|
||||
| { role?: string; content?: string; tool_calls?: Array<{ index: number } & ToolCall> }
|
||||
| Record<string, never>;
|
||||
finishReason: "stop" | "tool_calls" | null;
|
||||
usage?: Usage;
|
||||
vector?: VectorReceipt;
|
||||
}): Record<string, unknown> {
|
||||
|
||||
35
gateway/src/openai/transcripts.ts
Normal file
35
gateway/src/openai/transcripts.ts
Normal file
@@ -0,0 +1,35 @@
|
||||
/**
|
||||
* gateway/src/openai/transcripts.ts — in-memory per-session state shared by
|
||||
* the session routes (sessions.ts), the demo affinity executor (demoChat.ts)
|
||||
* and the sweeper's housekeeping.
|
||||
*
|
||||
* transcripts: mirror of each open session's transcript (sessionId → messages).
|
||||
* Kept hash-identical to the supplier's authoritative transcript (endChat's
|
||||
* response_hash check). Lost on restart — abandoned sessions are recovered
|
||||
* by the supplier idle watchdog + gateway sweeper.
|
||||
* sessionLocks: per-session Mutex serializing turns on ONE session (concurrent
|
||||
* turns would interleave supplier transcript appends). Distinct from the
|
||||
* per-key Mutex, which only serializes on-chain wallet work.
|
||||
*/
|
||||
|
||||
import type { ChatMessage } from "@marketplace/shared/tx";
|
||||
import { Mutex } from "../sdk/registry.js";
|
||||
|
||||
export const transcripts = new Map<string, ChatMessage[]>();
|
||||
|
||||
const sessionLocks = new Map<string, Mutex>();
|
||||
|
||||
export function getSessionLock(sessionId: string): Mutex {
|
||||
let lock = sessionLocks.get(sessionId);
|
||||
if (!lock) {
|
||||
lock = new Mutex();
|
||||
sessionLocks.set(sessionId, lock);
|
||||
}
|
||||
return lock;
|
||||
}
|
||||
|
||||
/** Forget all in-memory state for a session (closed/gone). */
|
||||
export function dropSessionState(sessionId: string): void {
|
||||
transcripts.delete(sessionId);
|
||||
sessionLocks.delete(sessionId);
|
||||
}
|
||||
@@ -2,7 +2,7 @@
|
||||
* gateway/src/openai/validate.ts — request body validation for chat endpoints.
|
||||
*/
|
||||
|
||||
import type { ChatMessage } from "@marketplace/shared/tx";
|
||||
import { normalizeChatMessage, type ChatMessage } from "@marketplace/shared/tx";
|
||||
import { badRequest } from "./errors.js";
|
||||
|
||||
export interface ParsedChatRequest {
|
||||
@@ -10,21 +10,30 @@ export interface ParsedChatRequest {
|
||||
messages: ChatMessage[];
|
||||
maxTokens?: number;
|
||||
stream: boolean;
|
||||
tools?: unknown[];
|
||||
toolChoice?: unknown;
|
||||
}
|
||||
|
||||
const ROLES = new Set(["system", "user", "assistant"]);
|
||||
|
||||
/** Validate an OpenAI chat.completions-style body. Rejects tools/functions;
|
||||
* silently ignores temperature/n/stop/top_p/etc (documented). */
|
||||
export function parseChatRequest(body: unknown): ParsedChatRequest {
|
||||
/** Validate an OpenAI chat.completions-style body. Default path rejects
|
||||
* tools/functions; `allowTools` (demo keys) accepts tools/tool_choice, the
|
||||
* `tool` role, null content and assistant tool_calls, normalizing every
|
||||
* message into canonical mirror shape. Silently ignores temperature/n/stop/
|
||||
* top_p/etc (documented). */
|
||||
export function parseChatRequest(body: unknown, opts?: { allowTools?: boolean }): ParsedChatRequest {
|
||||
const allowTools = opts?.allowTools === true;
|
||||
if (typeof body !== "object" || body === null) {
|
||||
throw badRequest("invalid_body", "request body must be a JSON object");
|
||||
}
|
||||
const b = body as Record<string, unknown>;
|
||||
|
||||
if ("tools" in b || "tool_choice" in b || "functions" in b || "function_call" in b) {
|
||||
if (!allowTools && ("tools" in b || "tool_choice" in b || "functions" in b || "function_call" in b)) {
|
||||
throw badRequest("unsupported_parameter", "tools and function-calling are not supported by this gateway");
|
||||
}
|
||||
if (allowTools && ("functions" in b || "function_call" in b)) {
|
||||
throw badRequest("unsupported_parameter", "legacy functions/function_call are not supported; use tools");
|
||||
}
|
||||
|
||||
const model = b.model;
|
||||
if (typeof model !== "string" || model === "") {
|
||||
@@ -37,6 +46,14 @@ export function parseChatRequest(body: unknown): ParsedChatRequest {
|
||||
}
|
||||
const messages: ChatMessage[] = [];
|
||||
for (const m of rawMessages) {
|
||||
if (allowTools) {
|
||||
const msg = normalizeChatMessage(m);
|
||||
if (!msg) {
|
||||
throw badRequest("invalid_messages", "each message must be OpenAI-shaped {role, content, tool_calls?, tool_call_id?}");
|
||||
}
|
||||
messages.push(msg);
|
||||
continue;
|
||||
}
|
||||
if (
|
||||
typeof m !== "object" ||
|
||||
m === null ||
|
||||
@@ -50,6 +67,13 @@ export function parseChatRequest(body: unknown): ParsedChatRequest {
|
||||
messages.push({ role: mm.role, content: mm.content });
|
||||
}
|
||||
|
||||
let tools: unknown[] | undefined;
|
||||
let toolChoice: unknown;
|
||||
if (allowTools && Array.isArray(b.tools) && b.tools.length > 0) {
|
||||
tools = b.tools;
|
||||
toolChoice = b.tool_choice;
|
||||
}
|
||||
|
||||
let maxTokens: number | undefined;
|
||||
const mt = b.max_tokens ?? b.max_completion_tokens;
|
||||
if (mt !== undefined && mt !== null) {
|
||||
@@ -60,5 +84,5 @@ export function parseChatRequest(body: unknown): ParsedChatRequest {
|
||||
}
|
||||
|
||||
const stream = b.stream === true;
|
||||
return { model, messages, maxTokens, stream };
|
||||
return { model, messages, maxTokens, stream, tools, toolChoice };
|
||||
}
|
||||
|
||||
@@ -101,19 +101,27 @@ export async function selectCandidates(opts: SelectSupplierOpts): Promise<Suppli
|
||||
return candidates;
|
||||
}
|
||||
|
||||
/** Distinct models across Active suppliers (for GET /openai/v1/models). */
|
||||
/** Distinct models across Active suppliers (for GET /openai/v1/models).
|
||||
* Pass capabilityId to list only models under one capability (demo keys see
|
||||
* only llm.chat.v1 models — the session-backed executor's routing target). */
|
||||
export async function listModels(opts: {
|
||||
indexerUrl: string;
|
||||
capabilityId?: string;
|
||||
fetchFn?: typeof globalThis.fetch;
|
||||
}): Promise<string[]> {
|
||||
const fetchFn = opts.fetchFn ?? globalThis.fetch;
|
||||
const res = await fetchFn(`${opts.indexerUrl}/suppliers`);
|
||||
const url = opts.capabilityId
|
||||
? `${opts.indexerUrl}/suppliers?capability_id=${encodeURIComponent(opts.capabilityId)}`
|
||||
: `${opts.indexerUrl}/suppliers`;
|
||||
const res = await fetchFn(url);
|
||||
if (!res.ok) throw new Error(`indexer /suppliers returned ${res.status}`);
|
||||
const body = (await res.json()) as unknown;
|
||||
if (!Array.isArray(body)) throw new Error("indexer /suppliers did not return an array");
|
||||
const models = new Set<string>();
|
||||
for (const raw of body as SupplierView[]) {
|
||||
if (raw.advert_status === "Active" && typeof raw.model === "string") models.add(raw.model);
|
||||
if (raw.advert_status !== "Active" || typeof raw.model !== "string") continue;
|
||||
if (opts.capabilityId !== undefined && raw.capability_id !== opts.capabilityId) continue;
|
||||
models.add(raw.model);
|
||||
}
|
||||
return [...models].sort();
|
||||
}
|
||||
|
||||
@@ -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));
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Lineage-match open sessions: the session stores its ORIGINAL Open ref;
|
||||
// rows sharing that row's posted_at are the same escrow in later states.
|
||||
const sessionByPostedAt = new Map<number, SessionRow>();
|
||||
for (const session of deps.store.listOpenSessionsByKey(keyRow.id)) {
|
||||
const origin = rows.find((r) => r.utxo_ref === session.escrow_ref);
|
||||
if (origin) sessionByPostedAt.set(origin.posted_at, session);
|
||||
}
|
||||
|
||||
const ctx = deps.registry.getContext(keyRow);
|
||||
for (const row of rows) {
|
||||
const ref = parseRef(row.utxo_ref);
|
||||
if (!ref) continue;
|
||||
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") {
|
||||
try {
|
||||
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) {
|
||||
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<void> {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Mark closed/settled chat sessions whose escrow is no longer active.
|
||||
// (Best-effort housekeeping; the escrow recovery above is the load-bearing part.)
|
||||
}
|
||||
|
||||
/** Start the periodic sweeper. Returns a stop() function. */
|
||||
|
||||
81
gateway/test/affinity.test.ts
Normal file
81
gateway/test/affinity.test.ts
Normal file
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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<string, unknown>, finishReason: string, usage = { prompt_tokens: 3, completion_tokens: 2 }) {
|
||||
return { type: "done", message, finish_reason: finishReason, usage };
|
||||
}
|
||||
|
||||
/** fetchFn serving the indexer supplier list and scripted /v1/chat/message responses. */
|
||||
function demoFetch(messageResponses: Array<() => Response>) {
|
||||
const messageCalls: Array<Record<string, unknown>> = [];
|
||||
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<string, unknown>) : {});
|
||||
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<Record<string, unknown>> = [{ role: "user", content: "what time is it?" }];
|
||||
const res1 = await request(app)
|
||||
.post("/openai/v1/chat/completions")
|
||||
.set("authorization", `Bearer ${rawKey}`)
|
||||
.send({ model: "kimi", messages: history, tools: TOOLS });
|
||||
expect(res1.status).toBe(200);
|
||||
expect(res1.body.choices[0].message.tool_calls).toEqual([TOOL_CALL]);
|
||||
expect(res1.body.choices[0].finish_reason).toBe("tool_calls");
|
||||
expect(startChat).toHaveBeenCalledTimes(1);
|
||||
// First turn replays the full history and forwards tools.
|
||||
expect((messageCalls[0].messages as unknown[]).length).toBe(1);
|
||||
expect(messageCalls[0].tools).toEqual(TOOLS);
|
||||
|
||||
// Turn 2: client echoes the assistant tool_calls + tool result.
|
||||
history.push({ role: "assistant", content: "", tool_calls: [TOOL_CALL] });
|
||||
history.push({ role: "tool", content: "3pm", tool_call_id: "call_1" });
|
||||
const res2 = await request(app)
|
||||
.post("/openai/v1/chat/completions")
|
||||
.set("authorization", `Bearer ${rawKey}`)
|
||||
.send({ model: "kimi", messages: history, tools: TOOLS });
|
||||
expect(res2.status).toBe(200);
|
||||
expect(res2.body.choices[0].message.content).toBe("It is 3pm");
|
||||
expect(res2.body.choices[0].finish_reason).toBe("stop");
|
||||
// Affinity hit: no new session, only the unseen delta is forwarded.
|
||||
expect(startChat).toHaveBeenCalledTimes(1);
|
||||
const delta = messageCalls[1].messages as Array<{ role: string }>;
|
||||
expect(delta).toHaveLength(1);
|
||||
expect(delta[0].role).toBe("tool");
|
||||
});
|
||||
|
||||
it("streams tool_calls chunks with finish_reason tool_calls", async () => {
|
||||
const { fetchFn } = demoFetch([
|
||||
() => supplierSse([doneFrame({ role: "assistant", content: "", tool_calls: [TOOL_CALL] }, "tool_calls")]),
|
||||
]);
|
||||
const deps = makeDeps(fetchFn, { chain: fundedChain });
|
||||
const { rawKey } = setupDemoKey(deps);
|
||||
const res = await request(createApp(deps))
|
||||
.post("/openai/v1/chat/completions")
|
||||
.set("authorization", `Bearer ${rawKey}`)
|
||||
.send({ model: "kimi", messages: [{ role: "user", content: "time?" }], tools: TOOLS, stream: true });
|
||||
expect(res.status).toBe(200);
|
||||
const chunks = res.text
|
||||
.split("\n\n")
|
||||
.filter((l) => l.startsWith("data: ") && !l.includes("[DONE]"))
|
||||
.map((l) => JSON.parse(l.slice(6)));
|
||||
const toolChunk = chunks.find((c) => c.choices?.[0]?.delta?.tool_calls);
|
||||
expect(toolChunk.choices[0].delta.tool_calls[0].function.name).toBe("get_time");
|
||||
expect(chunks.at(-1).choices[0].finish_reason).toBe("tool_calls");
|
||||
expect(res.text).toContain("data: [DONE]");
|
||||
});
|
||||
|
||||
it("transparently reopens when the supplier lost the session (idle-close)", async () => {
|
||||
const { fetchFn, messageCalls } = demoFetch([
|
||||
() => supplierSse([{ type: "token", value: "Hi" }, doneFrame({ role: "assistant", content: "Hi" }, "stop")]),
|
||||
() => new Response("not found", { status: 404 }),
|
||||
() => supplierSse([{ type: "token", value: "Sure" }, doneFrame({ role: "assistant", content: "Sure" }, "stop")]),
|
||||
]);
|
||||
const deps = makeDeps(fetchFn, { chain: fundedChain });
|
||||
const { rawKey, startChat } = setupDemoKey(deps);
|
||||
const app = createApp(deps);
|
||||
|
||||
const res1 = await request(app)
|
||||
.post("/openai/v1/chat/completions")
|
||||
.set("authorization", `Bearer ${rawKey}`)
|
||||
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
|
||||
expect(res1.status).toBe(200);
|
||||
|
||||
// Extended history matches the session, but the supplier 404s the turn →
|
||||
// the executor closes the session and reopens, replaying full history.
|
||||
const history = [
|
||||
{ role: "user", content: "hi" },
|
||||
{ role: "assistant", content: "Hi" },
|
||||
{ role: "user", content: "more please" },
|
||||
];
|
||||
const res2 = await request(app)
|
||||
.post("/openai/v1/chat/completions")
|
||||
.set("authorization", `Bearer ${rawKey}`)
|
||||
.send({ model: "kimi", messages: history });
|
||||
expect(res2.status).toBe(200);
|
||||
expect(res2.body.choices[0].message.content).toBe("Sure");
|
||||
expect(startChat).toHaveBeenCalledTimes(2);
|
||||
expect((messageCalls[2].messages as unknown[]).length).toBe(3);
|
||||
// The dead session's fee was recorded for spend audit.
|
||||
const usage = deps.store.listUsage(deps.store.listAllKeys().find((k) => k.demo === 1)!.id, 10);
|
||||
expect(usage.some((u) => u.kind === "chat_session" && u.cost_lovelace === "1000000")).toBe(true);
|
||||
});
|
||||
|
||||
it("rejects histories 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);
|
||||
});
|
||||
});
|
||||
|
||||
79
gateway/test/migration.test.ts
Normal file
79
gateway/test/migration.test.ts
Normal file
@@ -0,0 +1,79 @@
|
||||
/**
|
||||
* migration.test.ts — the demo-column migration on pre-existing databases.
|
||||
*
|
||||
* GatewayStore has no migration framework (pure CREATE TABLE IF NOT EXISTS),
|
||||
* so opening a DB created before the `demo` column must ALTER it in place.
|
||||
*/
|
||||
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { tmpdir } from "os";
|
||||
import { join } from "path";
|
||||
import { mkdirSync } from "fs";
|
||||
import { randomUUID } from "crypto";
|
||||
import Database from "better-sqlite3";
|
||||
import { GatewayStore } from "../src/db/store.js";
|
||||
|
||||
const LEGACY_API_KEYS_SQL = `
|
||||
CREATE TABLE IF NOT EXISTS api_keys (
|
||||
id TEXT PRIMARY KEY,
|
||||
key_hash TEXT NOT NULL UNIQUE,
|
||||
key_prefix TEXT NOT NULL,
|
||||
label TEXT,
|
||||
wallet_pkh TEXT NOT NULL,
|
||||
deposit_address TEXT NOT NULL,
|
||||
enc_priv_nonce TEXT NOT NULL,
|
||||
enc_priv_ct TEXT NOT NULL,
|
||||
enc_priv_tag TEXT NOT NULL,
|
||||
master_key_version INTEGER NOT NULL DEFAULT 1,
|
||||
created_at INTEGER NOT NULL,
|
||||
disabled INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
`;
|
||||
|
||||
function keyRow(id: string) {
|
||||
return {
|
||||
id,
|
||||
key_hash: `hash-${id}`,
|
||||
key_prefix: "vk_test_xxxx",
|
||||
label: null,
|
||||
wallet_pkh: "pkh",
|
||||
deposit_address: "addr_test1x",
|
||||
enc_priv_nonce: "n",
|
||||
enc_priv_ct: "c",
|
||||
enc_priv_tag: "t",
|
||||
master_key_version: 1,
|
||||
created_at: Date.now(),
|
||||
};
|
||||
}
|
||||
|
||||
describe("GatewayStore demo-column migration", () => {
|
||||
it("adds the demo column to a legacy DB and defaults existing keys to 0", () => {
|
||||
const dir = join(tmpdir(), `gw-mig-${randomUUID()}`);
|
||||
mkdirSync(dir, { recursive: true });
|
||||
const raw = new Database(join(dir, "gateway.db"));
|
||||
raw.exec(LEGACY_API_KEYS_SQL);
|
||||
raw
|
||||
.prepare(
|
||||
`INSERT INTO api_keys (id, key_hash, key_prefix, label, wallet_pkh, deposit_address,
|
||||
enc_priv_nonce, enc_priv_ct, enc_priv_tag, master_key_version, created_at, disabled)
|
||||
VALUES (@id, @key_hash, @key_prefix, @label, @wallet_pkh, @deposit_address,
|
||||
@enc_priv_nonce, @enc_priv_ct, @enc_priv_tag, @master_key_version, @created_at, 0)`,
|
||||
)
|
||||
.run(keyRow("legacy"));
|
||||
raw.close();
|
||||
|
||||
const store = new GatewayStore(dir);
|
||||
const legacy = store.getKeyByHash("hash-legacy");
|
||||
expect(legacy?.demo).toBe(0);
|
||||
|
||||
store.insertKey({ ...keyRow("demo-key"), demo: 1 });
|
||||
expect(store.getKeyByHash("hash-demo-key")?.demo).toBe(1);
|
||||
});
|
||||
|
||||
it("persists the demo flag on fresh databases (default 0)", () => {
|
||||
const dir = join(tmpdir(), `gw-fresh-${randomUUID()}`);
|
||||
const store = new GatewayStore(dir);
|
||||
store.insertKey(keyRow("plain"));
|
||||
expect(store.getKeyByHash("hash-plain")?.demo).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -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");
|
||||
});
|
||||
|
||||
88
packages/shared/src/tx/chatMessage.ts
Normal file
88
packages/shared/src/tx/chatMessage.ts
Normal file
@@ -0,0 +1,88 @@
|
||||
/**
|
||||
* tx/chatMessage.ts — canonical ChatMessage normalization + equivalence.
|
||||
*
|
||||
* The supplier's session transcript and the gateway's in-memory mirror must be
|
||||
* hash-identical (receipt response_hash = sha256(canonicalize(transcript))),
|
||||
* so both sides build messages through normalizeChatMessage: optional fields
|
||||
* are truly absent when empty, content is never null, and vendor extras are
|
||||
* stripped. chatMessagesEquivalent is the tolerant comparison used by the
|
||||
* demo-endpoint transcript-prefix affinity matcher.
|
||||
*/
|
||||
|
||||
import type { ChatMessage, ToolCall } from "./types.js";
|
||||
|
||||
const ROLES = new Set(["system", "user", "assistant", "tool"]);
|
||||
|
||||
function normalizeToolCall(raw: unknown): ToolCall | null {
|
||||
if (typeof raw !== "object" || raw === null) return null;
|
||||
const tc = raw as Record<string, unknown>;
|
||||
const fn = tc.function as Record<string, unknown> | undefined;
|
||||
if (typeof tc.id !== "string" || tc.id === "") return null;
|
||||
if (tc.type !== "function") return null;
|
||||
if (typeof fn !== "object" || fn === null) return null;
|
||||
if (typeof fn.name !== "string" || fn.name === "") return null;
|
||||
if (typeof fn.arguments !== "string") return null;
|
||||
return { id: tc.id, type: "function", function: { name: fn.name, arguments: fn.arguments } };
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate and canonicalize an untrusted message-shaped value.
|
||||
* Returns null when the value is not a usable ChatMessage.
|
||||
*/
|
||||
export function normalizeChatMessage(raw: unknown): ChatMessage | null {
|
||||
if (typeof raw !== "object" || raw === null) return null;
|
||||
const m = raw as Record<string, unknown>;
|
||||
if (typeof m.role !== "string" || !ROLES.has(m.role)) return null;
|
||||
const role = m.role as ChatMessage["role"];
|
||||
|
||||
let content: string;
|
||||
if (typeof m.content === "string") content = m.content;
|
||||
else if (m.content === null || m.content === undefined) content = "";
|
||||
else return null;
|
||||
|
||||
const msg: ChatMessage = { role, content };
|
||||
|
||||
if (m.tool_calls !== undefined && m.tool_calls !== null) {
|
||||
if (role !== "assistant" || !Array.isArray(m.tool_calls)) return null;
|
||||
const calls: ToolCall[] = [];
|
||||
for (const raw of m.tool_calls) {
|
||||
const tc = normalizeToolCall(raw);
|
||||
if (!tc) return null;
|
||||
calls.push(tc);
|
||||
}
|
||||
if (calls.length > 0) msg.tool_calls = calls;
|
||||
}
|
||||
|
||||
if (m.tool_call_id !== undefined && m.tool_call_id !== null) {
|
||||
if (role !== "tool" || typeof m.tool_call_id !== "string" || m.tool_call_id === "") return null;
|
||||
msg.tool_call_id = m.tool_call_id;
|
||||
}
|
||||
if (role === "tool" && msg.tool_call_id === undefined) return null;
|
||||
|
||||
return msg;
|
||||
}
|
||||
|
||||
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)
|
||||
);
|
||||
}
|
||||
@@ -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";
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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 <apiKey>`
|
||||
* 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<OpenAiResult> {
|
||||
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<string, unknown> = {
|
||||
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<string, string> = {
|
||||
@@ -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<number, { id: string; name: string; arguments: string }>();
|
||||
|
||||
const handleToolCallDeltas = (raw: unknown): void => {
|
||||
if (!Array.isArray(raw)) return;
|
||||
for (const item of raw) {
|
||||
if (!item || typeof item !== "object") continue;
|
||||
const frag = item as Record<string, unknown>;
|
||||
const index = typeof frag.index === "number" ? frag.index : 0;
|
||||
const part = toolCallParts.get(index) ?? { id: "", name: "", arguments: "" };
|
||||
if (typeof frag.id === "string" && frag.id.length > 0) part.id = frag.id;
|
||||
const fn = frag.function as Record<string, unknown> | undefined;
|
||||
if (fn && typeof fn === "object") {
|
||||
if (typeof fn.name === "string" && fn.name.length > 0) part.name = fn.name;
|
||||
if (typeof fn.arguments === "string") part.arguments += fn.arguments;
|
||||
}
|
||||
toolCallParts.set(index, part);
|
||||
}
|
||||
};
|
||||
|
||||
const handleData = (payload: string): void => {
|
||||
const trimmed = payload.trim();
|
||||
@@ -284,14 +320,15 @@ export async function callOpenAiStream(
|
||||
const obj = parsed as Record<string, unknown>;
|
||||
const choices = obj.choices;
|
||||
if (Array.isArray(choices) && choices.length > 0) {
|
||||
const delta = (choices[0] as Record<string, unknown>)?.delta as
|
||||
| Record<string, unknown>
|
||||
| undefined;
|
||||
const choice = choices[0] as Record<string, unknown>;
|
||||
const delta = choice?.delta as Record<string, unknown> | undefined;
|
||||
const content = delta?.content;
|
||||
if (typeof content === "string" && content.length > 0) {
|
||||
accumulated += content;
|
||||
onToken(content);
|
||||
}
|
||||
handleToolCallDeltas(delta?.tool_calls);
|
||||
if (typeof choice?.finish_reason === "string") finishReason = choice.finish_reason;
|
||||
}
|
||||
const usageRaw = obj.usage;
|
||||
if (usageRaw && typeof usageRaw === "object") {
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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 };
|
||||
// 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 });
|
||||
|
||||
96
tests/unit/shared-chat-message.test.ts
Normal file
96
tests/unit/shared-chat-message.test.ts
Normal file
@@ -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);
|
||||
});
|
||||
});
|
||||
103
tests/unit/supplier-openai-stream.test.ts
Normal file
103
tests/unit/supplier-openai-stream.test.ts
Normal file
@@ -0,0 +1,103 @@
|
||||
/**
|
||||
* supplier-openai-stream.test.ts — callOpenAiStream tools passthrough.
|
||||
*
|
||||
* Mocks global.fetch with SSE-shaped Response bodies. Covers:
|
||||
* - tools/tool_choice included in the upstream payload only when provided
|
||||
* - fragmented delta.tool_calls accumulation across frames
|
||||
* - pure tool-call turns (empty content) no longer throw openai_malformed
|
||||
* - empty content AND no tool calls still throws
|
||||
* - finish_reason capture
|
||||
*/
|
||||
|
||||
import { describe, it, expect, vi, afterEach } from "vitest";
|
||||
import { callOpenAiStream, OpenAiError } from "../../supplier/src/openai.js";
|
||||
|
||||
const BASE = { baseUrl: "http://up", model: "kimi", messages: [{ role: "user" as const, content: "hi" }], timeoutMs: 5_000 };
|
||||
const TOOLS = [{ type: "function", function: { name: "get_time", parameters: {} } }];
|
||||
|
||||
function sse(frames: unknown[]): string {
|
||||
return frames.map((f) => `data: ${typeof f === "string" ? f : JSON.stringify(f)}\n\n`).join("");
|
||||
}
|
||||
|
||||
function makeStreamFetch(body: string) {
|
||||
return vi.fn().mockResolvedValue(new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } }));
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
describe("callOpenAiStream — tools passthrough", () => {
|
||||
const CONTENT_FRAMES = sse([
|
||||
{ choices: [{ delta: { content: "Hello" }, finish_reason: null }] },
|
||||
{ choices: [{ delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 3, completion_tokens: 2 } },
|
||||
"[DONE]",
|
||||
]);
|
||||
|
||||
it("includes tools + tool_choice in the payload when provided", async () => {
|
||||
const fetchFn = makeStreamFetch(CONTENT_FRAMES);
|
||||
vi.stubGlobal("fetch", fetchFn);
|
||||
await callOpenAiStream({ ...BASE, tools: TOOLS, toolChoice: "auto" }, () => {});
|
||||
const body = JSON.parse(fetchFn.mock.calls[0][1].body as string);
|
||||
expect(body.tools).toEqual(TOOLS);
|
||||
expect(body.tool_choice).toBe("auto");
|
||||
});
|
||||
|
||||
it("omits tools from the payload when not provided", async () => {
|
||||
const fetchFn = makeStreamFetch(CONTENT_FRAMES);
|
||||
vi.stubGlobal("fetch", fetchFn);
|
||||
await callOpenAiStream(BASE, () => {});
|
||||
const body = JSON.parse(fetchFn.mock.calls[0][1].body as string);
|
||||
expect("tools" in body).toBe(false);
|
||||
expect("tool_choice" in body).toBe(false);
|
||||
});
|
||||
|
||||
it("accumulates fragmented delta.tool_calls across frames", async () => {
|
||||
const frames = sse([
|
||||
{ choices: [{ delta: { tool_calls: [{ index: 0, id: "call_1", type: "function", function: { name: "get_time", arguments: "" } }] }, finish_reason: null }] },
|
||||
{ choices: [{ delta: { tool_calls: [{ index: 0, function: { arguments: '{"tz":' } }] }, finish_reason: null }] },
|
||||
{ choices: [{ delta: { tool_calls: [{ index: 0, function: { arguments: '"utc"}' } }] }, finish_reason: null }] },
|
||||
{ choices: [{ delta: {}, finish_reason: "tool_calls" }], usage: { prompt_tokens: 7, completion_tokens: 5 } },
|
||||
"[DONE]",
|
||||
]);
|
||||
vi.stubGlobal("fetch", makeStreamFetch(frames));
|
||||
const result = await callOpenAiStream({ ...BASE, tools: TOOLS }, () => {});
|
||||
expect(result.tool_calls).toEqual([
|
||||
{ id: "call_1", type: "function", function: { name: "get_time", arguments: '{"tz":"utc"}' } },
|
||||
]);
|
||||
expect(result.finish_reason).toBe("tool_calls");
|
||||
expect(result.content).toBe("");
|
||||
expect(result.prompt_tokens).toBe(7);
|
||||
});
|
||||
|
||||
it("orders multiple tool calls by index", async () => {
|
||||
const frames = sse([
|
||||
{ choices: [{ delta: { tool_calls: [
|
||||
{ index: 1, id: "call_b", type: "function", function: { name: "b", arguments: "{}" } },
|
||||
{ index: 0, id: "call_a", type: "function", function: { name: "a", arguments: "{}" } },
|
||||
] }, finish_reason: null }] },
|
||||
{ choices: [{ delta: {}, finish_reason: "tool_calls" }] },
|
||||
"[DONE]",
|
||||
]);
|
||||
vi.stubGlobal("fetch", makeStreamFetch(frames));
|
||||
const result = await callOpenAiStream({ ...BASE, tools: TOOLS }, () => {});
|
||||
expect(result.tool_calls?.map((t) => t.id)).toEqual(["call_a", "call_b"]);
|
||||
});
|
||||
|
||||
it("still throws openai_malformed when content AND tool calls are both empty", async () => {
|
||||
const frames = sse([{ choices: [{ delta: {}, finish_reason: "stop" }] }, "[DONE]"]);
|
||||
vi.stubGlobal("fetch", makeStreamFetch(frames));
|
||||
await expect(callOpenAiStream(BASE, () => {})).rejects.toMatchObject({ reason: "openai_malformed" });
|
||||
await expect(callOpenAiStream(BASE, () => {})).rejects.toBeInstanceOf(OpenAiError);
|
||||
});
|
||||
|
||||
it("still streams plain content unchanged (no tools)", async () => {
|
||||
vi.stubGlobal("fetch", makeStreamFetch(CONTENT_FRAMES));
|
||||
const tokens: string[] = [];
|
||||
const result = await callOpenAiStream(BASE, (t) => tokens.push(t));
|
||||
expect(tokens).toEqual(["Hello"]);
|
||||
expect(result.content).toBe("Hello");
|
||||
expect(result.tool_calls).toBeUndefined();
|
||||
expect(result.finish_reason).toBe("stop");
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user