From 9acad5b2b812656744e0c368e3d029dd1e9a5fc1 Mon Sep 17 00:00:00 2001 From: david Date: Mon, 8 Jun 2026 16:18:54 +0200 Subject: [PATCH] feat: llm.chat.v1 multi-turn supplier type + free Kimi K2.6 chat demo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a conversational supplier capability and a free demo, both streaming. New paid `llm.chat.v1` type: one escrow per conversation (flat fee). "Start chat" posts the escrow with a session-init placeholder prompt_hash and the supplier Claims (reserving its single slot); turns then stream off-chain via SSE with zero chain interaction; "End chat" has the supplier Submit a receipt over the full transcript and the buyer Accept — the moment the user is charged. The on-chain validator is content-agnostic (pins but never verifies the hashes), so this reuses the existing contract unchanged. A supplier-side idle watchdog auto-ends abandoned sessions before deliver_by. Free Kimi K2.6 chat demo: a synthetic "demo:kimi-chat" tile that uses the same chat UI but proxies straight to OpenRouter (no escrow, no payment), mirroring the existing Piper demo. Shared: - postChatEscrow.ts + chatSessionPromptHash() (session-init prompt_hash helper shared by the buyer builder and the supplier start gate). Supplier: - CAPABILITY_KIND="chat-session" (+ CHAT_IDLE_TIMEOUT_MS); callOpenAiStream() streaming client; ChatSessionStore + endChatSession(); /v1/chat/{start, message,end} handlers with idle + hard-cap watchdog timers. Buyer: - Marketplace.startChat()/endChat() with relaxed (session-init) receipt verification + crypto verifyReceipt; /v1/chat/{start,message,end} routes (SSE passthrough) and a free /v1/chat-demo/message OpenRouter proxy; OPENROUTER_API_KEY/OPENROUTER_BASE_URL config; ChatForm.tsx + Dashboard KIMI_DEMO_SUPPLIER tile and llm.chat.v1 dispatch. Deploy: - docker-compose.supplier-kimi-chat.yml + supplier/.env.kimi-chat.example. Typecheck passes across all workspaces; full vitest suite green except 15 pre-existing time-bomb/RED failures (tx-claim/accept/post-advert live-CBOR fixtures with hardcoded past deadlines + the M0-C golden-datum test), confirmed identical on a clean tree. Co-Authored-By: Claude Opus 4.8 (1M context) --- buyer/src/config.ts | 17 + buyer/src/index.ts | 2 + buyer/src/sdk/Marketplace.ts | 222 ++++++++++++ buyer/src/sdk/types.ts | 41 ++- buyer/src/server.ts | 272 +++++++++++++++ buyer/src/ui/components/ChatForm.tsx | 318 ++++++++++++++++++ buyer/src/ui/pages/Dashboard.tsx | 42 ++- .../docker-compose.supplier-kimi-chat.yml | 86 +++++ .../shared/src/tx/escrow/postChatEscrow.ts | 207 ++++++++++++ packages/shared/src/tx/index.ts | 2 + supplier/.env.kimi-chat.example | 62 ++++ supplier/src/chatSession.ts | 127 +++++++ supplier/src/chatSessionRunner.ts | 120 +++++++ supplier/src/config.ts | 41 ++- supplier/src/openai.ts | 160 +++++++++ supplier/src/server.ts | 280 ++++++++++++++- tests/unit/buyer-ui-smoke.test.tsx | 11 +- 17 files changed, 1997 insertions(+), 13 deletions(-) create mode 100644 buyer/src/ui/components/ChatForm.tsx create mode 100644 deploy/mainnet/docker-compose.supplier-kimi-chat.yml create mode 100644 packages/shared/src/tx/escrow/postChatEscrow.ts create mode 100644 supplier/.env.kimi-chat.example create mode 100644 supplier/src/chatSession.ts create mode 100644 supplier/src/chatSessionRunner.ts diff --git a/buyer/src/config.ts b/buyer/src/config.ts index 8faff6c..8c166f0 100644 --- a/buyer/src/config.ts +++ b/buyer/src/config.ts @@ -24,6 +24,13 @@ * When unset, /v1/synth-speech responds 503 (the * capability stays disabled — the rest of the * buyer-app boots normally). + * OPENROUTER_API_KEY — Bearer token for the free Kimi K2.6 chat demo + * (/v1/chat-demo/message). When unset, the demo + * endpoint responds 503 (the paid chat path is + * unaffected — it talks to on-chain suppliers). + * OPENROUTER_BASE_URL — Base URL for the demo proxy (default + * "https://openrouter.ai/api"; the "/v1/chat/completions" + * suffix is appended by the proxy). * ARCHIVE_DIR — Directory where the response archive (SQLite * metadata + per-escrow artefact files) lives. * Defaults to "./data/archive" relative to cwd. @@ -43,6 +50,8 @@ export interface BuyerConfig { ogmiosUrl: string; liveChain: boolean; ttsPiperBaseUrl: string; + openrouterApiKey: string; + openrouterBaseUrl: string; archiveDir: string; password: string; sessionSecret: string; @@ -99,6 +108,12 @@ export function loadConfig(env: Record): BuyerConfig // shape here; an unreachable URL surfaces as a clear 502 from the proxy. const ttsPiperBaseUrl = env.TTS_PIPER_BASE_URL ?? ""; + // OPENROUTER_API_KEY — optional. Empty string disables the free Kimi chat + // demo (/v1/chat-demo/message responds 503). The base URL defaults to the + // public OpenRouter host; the proxy appends "/v1/chat/completions". + const openrouterApiKey = env.OPENROUTER_API_KEY ?? ""; + const openrouterBaseUrl = (env.OPENROUTER_BASE_URL ?? "https://openrouter.ai/api").replace(/\/+$/, ""); + // ARCHIVE_DIR — host path (typically a docker bind-mount) where the // response archive lives. Default is a per-process relative dir to keep // dev ergonomics; in production the compose file overrides to a @@ -134,6 +149,8 @@ export function loadConfig(env: Record): BuyerConfig ogmiosUrl, liveChain, ttsPiperBaseUrl, + openrouterApiKey, + openrouterBaseUrl, archiveDir, password, sessionSecret, diff --git a/buyer/src/index.ts b/buyer/src/index.ts index 52286b5..90890cd 100644 --- a/buyer/src/index.ts +++ b/buyer/src/index.ts @@ -127,6 +127,8 @@ export async function runMain(env: Record): Promise< indexerUrl: config.indexerUrl, marketplace, ttsPiperBaseUrl: config.ttsPiperBaseUrl, + openrouterApiKey: config.openrouterApiKey, + openrouterBaseUrl: config.openrouterBaseUrl, archive, password: config.password, sessionSecret: config.sessionSecret, diff --git a/buyer/src/sdk/Marketplace.ts b/buyer/src/sdk/Marketplace.ts index 52bd263..84dda76 100644 --- a/buyer/src/sdk/Marketplace.ts +++ b/buyer/src/sdk/Marketplace.ts @@ -72,12 +72,15 @@ import type { WalletKey } from "@marketplace/shared/tx"; import { buildPostEscrowTx, buildPostTtsEscrowTx, + buildPostChatEscrowTx, ttsPromptHash, + chatSessionPromptHash, buildAcceptTx, buildReclaimTx, TxConstructionError, } from "@marketplace/shared/tx"; import { decodeAdvertDatum, decodeEscrowDatum, canonicalize } from "@marketplace/shared/cbor"; +import { verifyReceipt } from "@marketplace/shared/receipt"; import type { AdvertDatum } from "@marketplace/shared/cbor"; import type { SupplierView, @@ -86,6 +89,10 @@ import type { SubmitPromptResult, SubmitTtsOptions, SubmitTtsResult, + StartChatOptions, + StartChatResult, + EndChatOptions, + EndChatResult, AcceptResultOptions, ReclaimOptions, TaskRecord, @@ -144,6 +151,25 @@ function refToString(ref: OutputReference): string { return `${ref.txHash}#${ref.index}`; } +function parseRef(s: string): OutputReference | null { + const m = /^([0-9a-fA-F]{64})#(0|[1-9]\d*)$/.exec(s); + if (!m) return null; + return { txHash: m[1], index: Number(m[2]) }; +} + +/** Random 32-hex session nonce. Works in both Node and the browser. */ +function randomNonce(): string { + const g = globalThis as { + crypto?: { randomUUID?: () => string; getRandomValues?: (a: Uint8Array) => Uint8Array }; + }; + if (g.crypto?.randomUUID) return g.crypto.randomUUID().replace(/-/g, ""); + const bytes = new Uint8Array(16); + g.crypto?.getRandomValues?.(bytes); + let out = ""; + for (let i = 0; i < bytes.length; i++) out += bytes[i].toString(16).padStart(2, "0"); + return out; +} + function previewMessages(messages: { role: string; content: string }[]): string { const firstUser = messages.find((m) => m.role === "user") ?? messages[0]; const text = firstUser?.content ?? ""; @@ -781,6 +807,202 @@ export class Marketplace extends EventEmitterBase { }; } + // ─── startChat / endChat — multi-turn llm.chat.v1 lifecycle ──────────── + + /** + * Open a chat session: post the escrow with a session-init prompt_hash, wait + * for it to confirm, then tell the supplier to Claim (reserving its slot). + * The conversation then runs off-chain via the buyer-app's SSE passthrough. + */ + async startChat(opts: StartChatOptions): Promise { + const { advertRef, payment_lovelace } = opts; + + const advertUtxo = await this.chain.queryUtxo(advertRef); + if (!advertUtxo || !advertUtxo.datumHex) { + throw new TxConstructionError("advert ref not on chain", `no advert UTxO at ${refToString(advertRef)}`); + } + const advertDatum = decodeAdvertDatum(advertUtxo.datumHex); + + const sessionNonce = randomNonce(); + let escrowResult; + try { + escrowResult = await buildPostChatEscrowTx({ + chain: this.chain, + buyerKey: this.walletKey, + advertRef, + session_nonce: sessionNonce, + payment_lovelace, + }); + } catch (err) { + if (!(err instanceof TxConstructionError)) { + this.emitProgress({ type: "chain_submit_failed", detail: (err as Error).message }); + } + throw err; + } + + const escrowOutputRef = escrowResult.escrowOutputRef; + const escrowRefStr = refToString(escrowOutputRef); + this.emitProgress({ type: "escrow_posted", escrow_ref: escrowRefStr }); + + // Wait for the PostEscrow tx to confirm before the supplier Claims. + try { + await this.chain.awaitTx(escrowResult.expectedTxHash, 120_000); + } catch { + /* mock providers / tests */ + } + + const supplierHttp = new HttpClient({ baseUrl: advertDatum.endpoint_url, fetch: this.fetchImpl }); + let startRes; + try { + startRes = await supplierHttp.postJson( + "/v1/chat/start", + { session_nonce: sessionNonce, model: advertDatum.model }, + { headers: { "X-Escrow-Ref": escrowRefStr } }, + ); + } catch (err) { + if (err instanceof HttpError) { + const reason = err.kind === "timeout" ? "timeout" : "network_error"; + throw new SupplierError(reason, { message: err.message }); + } + throw err; + } + if (!startRes.ok) { + const bodyReason = startRes.body && typeof startRes.body === "object" + ? ((startRes.body as { reason?: string }).reason ?? "supplier_http_error") + : "supplier_http_error"; + throw new SupplierError(bodyReason, { + status: startRes.status, + message: `supplier /v1/chat/start returned ${startRes.status}`, + }); + } + + this.emitProgress({ type: "chat_started", escrow_ref: escrowRefStr }); + return { escrowRef: escrowOutputRef, sessionNonce, supplierBaseUrl: advertDatum.endpoint_url }; + } + + /** + * End a chat session: ask the supplier to Submit a transcript receipt, verify + * it (structural + crypto + session-init prompt_hash), then Accept the + * Submitted escrow — which is when the user is actually charged. + */ + async endChat(opts: EndChatOptions): Promise { + const { escrowRef, sessionNonce, transcript } = opts; + const escrowRefStr = refToString(escrowRef); + + // Resolve escrow → advert for endpoint_url, supplier_pkh, model, pub key. + const escrowUtxo = await this.chain.queryUtxo(escrowRef); + if (!escrowUtxo || !escrowUtxo.datumHex) { + throw new TxConstructionError("escrow ref not on chain", `no escrow UTxO at ${escrowRefStr}`); + } + const escrowDatum = decodeEscrowDatum(escrowUtxo.datumHex); + const advertUtxo = await this.chain.queryUtxo(escrowDatum.advert_ref); + if (!advertUtxo || !advertUtxo.datumHex) { + throw new TxConstructionError("advert ref not on chain", "advert UTxO for this escrow is missing"); + } + const advertDatum = decodeAdvertDatum(advertUtxo.datumHex); + + const supplierHttp = new HttpClient({ baseUrl: advertDatum.endpoint_url, fetch: this.fetchImpl }); + let endRes; + try { + endRes = await supplierHttp.postJson("/v1/chat/end", {}, { headers: { "X-Escrow-Ref": escrowRefStr } }); + } catch (err) { + if (err instanceof HttpError) { + const reason = err.kind === "timeout" ? "timeout" : "network_error"; + throw new SupplierError(reason, { message: err.message }); + } + throw err; + } + if (!endRes.ok || !endRes.body || typeof endRes.body !== "object") { + const bodyReason = endRes.body && typeof endRes.body === "object" + ? ((endRes.body as { reason?: string }).reason ?? "supplier_http_error") + : "supplier_http_error"; + throw new SupplierError(bodyReason, { + status: endRes.status, + message: `supplier /v1/chat/end returned ${endRes.status}`, + }); + } + const endBody = endRes.body as { + receipt?: import("@marketplace/shared/receipt").Receipt; + receipt_signature?: string; + submitted_ref?: string; + }; + if (!endBody.receipt || !endBody.receipt_signature) { + throw new SupplierError("malformed_response", { message: "supplier end response missing receipt" }); + } + const receipt = endBody.receipt; + const receiptSignature = endBody.receipt_signature; + + // ── Verify receipt ─────────────────────────────────────────────────── + if (receipt.supplier_pkh !== escrowDatum.supplier_pkh) { + throw new ReceiptVerificationError("wrong_supplier"); + } + if (receipt.escrow_ref !== escrowRefStr) { + throw new ReceiptVerificationError("wrong_escrow_ref"); + } + // prompt_hash is the session-init placeholder, NOT sha256(messages). + if (receipt.prompt_hash !== chatSessionPromptHash({ session_nonce: sessionNonce })) { + throw new ReceiptVerificationError("prompt_hash_mismatch"); + } + if (receipt.model !== advertDatum.model) { + throw new ReceiptVerificationError("request_spec_hash_mismatch"); + } + if (typeof receiptSignature !== "string" || !SIG_RE.test(receiptSignature) || receiptSignature === ZERO_SIGNATURE) { + throw new ReceiptVerificationError("invalid_signature"); + } + if (!HEX64_RE.test(receipt.prompt_hash) || !HEX64_RE.test(receipt.response_hash)) { + throw new ReceiptVerificationError("malformed_receipt"); + } + // Optional: recompute the transcript hash from the browser's mirror. + if (transcript && transcript.length > 0) { + const localHash = sha256Hex(canonicalize(transcript)); + if (localHash !== receipt.response_hash) { + console.warn( + `[marketplace] endChat transcript hash mismatch for ${escrowRefStr}: local ${localHash} != receipt ${receipt.response_hash}`, + ); + } + } + // Cryptographic verification via the supplier's published pub key (best-effort). + try { + const capRes = await supplierHttp.getJson("/capability"); + const pubKeyHex = capRes.body && typeof capRes.body === "object" + ? (capRes.body as { pub_key_hex?: string }).pub_key_hex + : undefined; + if (typeof pubKeyHex === "string" && /^[0-9a-fA-F]{64}$/.test(pubKeyHex)) { + if (!verifyReceipt({ receipt, signature: receiptSignature }, pubKeyHex)) { + throw new ReceiptVerificationError("invalid_signature"); + } + } + } catch (err) { + if (err instanceof ReceiptVerificationError) throw err; + /* /capability unreachable — structural verification stands */ + } + + this.emitProgress({ type: "receipt_verified", escrow_ref: escrowRefStr }); + + // ── Accept the Submitted escrow — the user is charged here ──────────── + const acceptedRef = parseRef(typeof endBody.submitted_ref === "string" ? endBody.submitted_ref : ""); + if (!acceptedRef) { + throw new SupplierError("malformed_response", { message: "supplier end response missing/invalid submitted_ref" }); + } + let acceptBuilt; + try { + acceptBuilt = await buildAcceptTx({ chain: this.chain, buyerKey: this.walletKey, escrowRef: acceptedRef }); + } catch (err) { + if (err instanceof TxConstructionError) throw err; + this.emitProgress({ type: "chain_submit_failed", detail: (err as Error).message }); + throw err; + } + try { + await this.chain.awaitTx(acceptBuilt.expectedTxHash, 120_000); + } catch { + /* mock providers / tests — Accept tx already submitted */ + } + this.emitProgress({ type: "accept_submitted", escrow_ref: refToString(acceptedRef) }); + this.emitProgress({ type: "chat_ended", escrow_ref: escrowRefStr }); + + return { receipt, receiptSignature, escrowRef, acceptedRef }; + } + // ─── acceptResult / reclaim ─────────────────────────────────────────── async acceptResult(opts: AcceptResultOptions): Promise { diff --git a/buyer/src/sdk/types.ts b/buyer/src/sdk/types.ts index f598fe5..4a0e149 100644 --- a/buyer/src/sdk/types.ts +++ b/buyer/src/sdk/types.ts @@ -55,7 +55,9 @@ export type ProgressEventType = | "receipt_verified" | "accept_submitted" | "reclaim_submitted" - | "chain_submit_failed"; + | "chain_submit_failed" + | "chat_started" + | "chat_ended"; /** * ProgressEvent — payload emitted via EventEmitter on each step. @@ -185,6 +187,43 @@ export interface SubmitTtsResult { escrowRef: OutputReference; } +/** + * StartChatOptions / StartChatResult — Marketplace.startChat() for the + * `llm.chat.v1` capability. Opens the escrow (session-init prompt_hash) and + * tells the supplier to Claim, reserving its single slot. The conversation + * then runs off-chain via the buyer-app's /v1/chat/message SSE passthrough. + */ +export interface StartChatOptions { + advertRef: OutputReference; + payment_lovelace: bigint; +} +export interface StartChatResult { + escrowRef: OutputReference; + /** Random nonce committed into the escrow's session-init prompt_hash. The + * buyer-app must hold this to verify the receipt at End. */ + sessionNonce: string; + /** Supplier endpoint_url, cached by the buyer-app for message routing. */ + supplierBaseUrl: string; +} + +/** + * EndChatOptions / EndChatResult — Marketplace.endChat(). Tells the supplier to + * Submit a transcript receipt, verifies it, then Accepts (charging the user). + */ +export interface EndChatOptions { + escrowRef: OutputReference; + sessionNonce: string; + /** The browser's local transcript mirror, used to verify response_hash. */ + transcript?: import("@marketplace/shared/tx").ChatMessage[]; +} +export interface EndChatResult { + receipt: Receipt; + receiptSignature: string; + escrowRef: OutputReference; + /** The Submitted escrow UTxO that was Accepted. */ + acceptedRef: OutputReference; +} + /** * AcceptResultOptions — arguments to Marketplace.acceptResult(). */ diff --git a/buyer/src/server.ts b/buyer/src/server.ts index 18ae9c3..8953750 100644 --- a/buyer/src/server.ts +++ b/buyer/src/server.ts @@ -82,6 +82,12 @@ export interface AppDeps { * `/v1/synth-speech` proxies to. Falls back to the public Vector-testnet * host when not set. Endpoint stays disabled (503) when this is empty. */ ttsPiperBaseUrl?: string; + /** Bearer token + base URL for the free Kimi K2.6 chat demo. When + * `openrouterApiKey` is empty, `/v1/chat-demo/message` responds 503. The + * demo proxies straight to OpenRouter (no escrow, no payment), mirroring + * the `/v1/synth-speech` Piper demo. */ + openrouterApiKey?: string; + openrouterBaseUrl?: string; /** Persistent off-chain audit trail. When provided, every successful * submitPrompt / submitTts call writes a row + on-disk artefacts and * /v1/responses* exposes the archive read-only. When undefined the @@ -626,6 +632,272 @@ export function createApp(deps: AppDeps): Express { } }); + // ── POST /v1/chat-demo/message — free Kimi K2.6 chat demo (streaming). + // Body: { messages: ChatMessage[] } (full running transcript) + // Returns: an SSE stream of normalized frames the ChatForm understands: + // data: {"type":"token","value":""}\n\n (repeated) + // data: {"type":"done"}\n\n + // data: {"type":"error","message":"…"}\n\n + // + // Mirrors /v1/synth-speech (Path B of the demo story): a direct proxy that + // bypasses escrow entirely so anyone can try the new chat capability for + // free. We re-shape OpenRouter's raw OpenAI SSE deltas into the uniform + // frame format so the browser parser is identical for demo and paid chat. + const openrouterApiKey = deps.openrouterApiKey ?? ""; + const openrouterBaseUrl = (deps.openrouterBaseUrl ?? "https://openrouter.ai/api").replace(/\/+$/, ""); + const demoFetch = deps.fetchImpl ?? globalThis.fetch; + const DEMO_CHAT_MODEL = "moonshotai/kimi-k2.6"; + + function isChatMessageArray(v: unknown): v is ChatMessage[] { + return Array.isArray(v) && v.length > 0 && v.every( + (m) => m && typeof m === "object" + && (["system", "user", "assistant"] as const).includes((m as ChatMessage).role) + && typeof (m as ChatMessage).content === "string", + ); + } + + app.post("/v1/chat-demo/message", async (req: Request, res: Response) => { + if (!openrouterApiKey) { + return jsonError( + res, + 503, + "service_unavailable", + "buyer-app booted without OPENROUTER_API_KEY; /v1/chat-demo/message disabled", + ); + } + const body = readBody(req.body); + if (!isChatMessageArray(body.messages)) { + return jsonError( + res, + 400, + "messages_required", + "body must include a non-empty messages[] array of {role,content}", + ); + } + const messages = body.messages as ChatMessage[]; + + // Open the SSE response to the browser up front so errors after the first + // byte are delivered as in-band `error` frames (the status line is already + // committed once we start streaming). + 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"); + if (typeof (res as Response & { flushHeaders?: () => void }).flushHeaders === "function") { + (res as Response & { flushHeaders: () => void }).flushHeaders(); + } + const sse = (frame: Record): void => { + res.write(`data: ${JSON.stringify(frame)}\n\n`); + }; + + try { + const upstream = await demoFetch(`${openrouterBaseUrl}/v1/chat/completions`, { + method: "POST", + headers: { + "content-type": "application/json", + accept: "text/event-stream", + authorization: `Bearer ${openrouterApiKey}`, + }, + body: JSON.stringify({ model: DEMO_CHAT_MODEL, messages, stream: true }), + }); + if (!upstream.ok || !upstream.body) { + const detail = await upstream.text().catch(() => ""); + sse({ type: "error", message: `OpenRouter ${upstream.status}: ${detail.slice(0, 200)}` }); + return res.end(); + } + + const reader = (upstream.body as ReadableStream).getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + const handleData = (payload: string): void => { + const trimmed = payload.trim(); + if (trimmed.length === 0 || trimmed === "[DONE]") return; + let parsed: unknown; + try { + parsed = JSON.parse(trimmed); + } catch { + return; + } + const choices = (parsed as Record)?.choices; + if (Array.isArray(choices) && choices.length > 0) { + const delta = (choices[0] as Record)?.delta as + | Record + | undefined; + const content = delta?.content; + if (typeof content === "string" && content.length > 0) { + sse({ type: "token", value: content }); + } + } + }; + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + let sepIdx: number; + while ((sepIdx = buffer.indexOf("\n\n")) >= 0) { + const frame = buffer.slice(0, sepIdx); + buffer = buffer.slice(sepIdx + 2); + for (const line of frame.split("\n")) { + if (line.startsWith("data:")) handleData(line.slice(5)); + } + } + } + sse({ type: "done" }); + return res.end(); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + console.error(`[buyer] /v1/chat-demo/message failed: ${message}`); + sse({ type: "error", message }); + return res.end(); + } + }); + + // ── Paid multi-turn chat (llm.chat.v1) ────────────────────────────── + // start → marketplace.startChat (PostEscrow + supplier Claim); message → + // SSE passthrough to the supplier (zero chain per turn); end → + // marketplace.endChat (supplier Submit + buyer Accept = charged). + // + // The supplier base URL discovered at start is cached so the per-turn + // message route doesn't re-query the chain. In-memory is fine: single + // operator, single active chat. + const chatRouteState = new Map(); + const chatFetch = deps.fetchImpl ?? globalThis.fetch; + + function chatErrorResponse(res: Response, err: unknown): Response { + const message = err instanceof Error ? err.message : String(err); + const reason = err instanceof Error && "reason" in err + ? String((err as { reason: unknown }).reason) + : "chat_failed"; + const status = err instanceof Error && "status" in err && typeof (err as { status?: unknown }).status === "number" + ? (err as { status: number }).status + : 502; + console.error(`[buyer] chat route failed reason=${reason} status=${status} message=${message}`); + return res.status(status).json({ error: reason, message }); + } + + app.post("/v1/chat/start", async (req: Request, res: Response) => { + if (!deps.marketplace) { + return jsonError(res, 503, "service_unavailable", + "buyer-app booted without Marketplace SDK; /v1/chat/start disabled"); + } + const body = readBody(req.body); + const rawRef = body.advert_ref; + if (typeof rawRef !== "string" || !ESCROW_REF_RE.test(rawRef)) { + return jsonError(res, 400, "advert_ref_invalid", + 'body must include { "advert_ref": "<64-hex-txhash>#" }'); + } + let payment_lovelace: bigint; + try { + payment_lovelace = BigInt(body.payment_lovelace as string | number); + } catch { + return jsonError(res, 400, "payment_lovelace_invalid", "body must include `payment_lovelace`"); + } + const m = ESCROW_REF_RE.exec(rawRef)!; + const advertRef = { txHash: m[1], index: Number(m[2]) }; + try { + const result = await deps.marketplace.startChat({ advertRef, payment_lovelace }); + const escrowRefStr = `${result.escrowRef.txHash}#${result.escrowRef.index}`; + chatRouteState.set(escrowRefStr, { supplierBaseUrl: result.supplierBaseUrl }); + return res.status(200).json({ escrow_ref: escrowRefStr, session_nonce: result.sessionNonce }); + } catch (err) { + return chatErrorResponse(res, err); + } + }); + + app.post("/v1/chat/message", async (req: Request, res: Response) => { + const body = readBody(req.body); + const rawRef = body.escrow_ref; + if (typeof rawRef !== "string" || !ESCROW_REF_RE.test(rawRef)) { + return jsonError(res, 400, "escrow_ref_invalid", + 'body must include { "escrow_ref": "<64-hex-txhash>#" }'); + } + 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"); + } + const state = chatRouteState.get(rawRef); + if (!state) { + return jsonError(res, 404, "chat_session_not_found", `no active chat session for ${rawRef}`); + } + + 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"); + if (typeof (res as Response & { flushHeaders?: () => void }).flushHeaders === "function") { + (res as Response & { flushHeaders: () => void }).flushHeaders(); + } + const sse = (frame: Record): void => { + res.write(`data: ${JSON.stringify(frame)}\n\n`); + }; + + try { + const upstream = await chatFetch(`${state.supplierBaseUrl.replace(/\/+$/, "")}/v1/chat/message`, { + method: "POST", + headers: { + "content-type": "application/json", + accept: "text/event-stream", + "X-Escrow-Ref": rawRef, + }, + body: JSON.stringify({ content }), + }); + if (!upstream.ok || !upstream.body) { + const detail = await upstream.text().catch(() => ""); + sse({ type: "error", message: `supplier ${upstream.status}: ${detail.slice(0, 200)}` }); + return res.end(); + } + // Pipe the supplier's SSE frames (already in {type:token|done|error} + // form) through verbatim. + const reader = (upstream.body as ReadableStream).getReader(); + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + res.write(Buffer.from(value)); + } + return res.end(); + } catch (err) { + sse({ type: "error", message: err instanceof Error ? err.message : String(err) }); + return res.end(); + } + }); + + app.post("/v1/chat/end", async (req: Request, res: Response) => { + if (!deps.marketplace) { + return jsonError(res, 503, "service_unavailable", + "buyer-app booted without Marketplace SDK; /v1/chat/end disabled"); + } + const body = readBody(req.body); + const rawRef = body.escrow_ref; + if (typeof rawRef !== "string" || !ESCROW_REF_RE.test(rawRef)) { + return jsonError(res, 400, "escrow_ref_invalid", + 'body must include { "escrow_ref": "<64-hex-txhash>#" }'); + } + const sessionNonce = typeof body.session_nonce === "string" ? body.session_nonce : ""; + if (sessionNonce.length === 0) { + return jsonError(res, 400, "session_nonce_required", "body.session_nonce is required"); + } + const transcript = Array.isArray(body.transcript) + ? (body.transcript as ChatMessage[]) + : undefined; + const m = ESCROW_REF_RE.exec(rawRef)!; + const escrowRef = { txHash: m[1], index: Number(m[2]) }; + try { + const result = await deps.marketplace.endChat({ escrowRef, sessionNonce, transcript }); + chatRouteState.delete(rawRef); + return res.status(200).json({ + status: "accepted", + escrow_ref: rawRef, + accepted_ref: `${result.acceptedRef.txHash}#${result.acceptedRef.index}`, + receipt: result.receipt, + receipt_signature: result.receiptSignature, + }); + } catch (err) { + return chatErrorResponse(res, err); + } + }); + // ── POST /v1/submit-tts — full server-side buyer lifecycle for TTS. // Body: { advert_ref, text, voice, format, speed, payment_lovelace } // Returns: { audio_b64, format, content_type, byte_length, receipt, diff --git a/buyer/src/ui/components/ChatForm.tsx b/buyer/src/ui/components/ChatForm.tsx new file mode 100644 index 0000000..08bfec9 --- /dev/null +++ b/buyer/src/ui/components/ChatForm.tsx @@ -0,0 +1,318 @@ +/** + * buyer/src/ui/components/ChatForm.tsx — capability form for `llm.chat.v1` + * (multi-turn chat sessions). + * + * Two modes, controlled by whether `advertRef` is provided (mirrors + * PiperTTSForm's demo-vs-marketplace split): + * + * - **Demo mode** (no advertRef): the synthetic "Kimi K2.6 (demo)" tile. + * No escrow, no payment, no Start/End. Every turn streams straight from + * OpenRouter via /v1/chat-demo/message. Fully concurrent + free. + * + * - **Marketplace mode** (advertRef + payment_lovelace): the paid chat type. + * "Start chat" opens the escrow (POST /v1/chat/start → PostEscrow + supplier + * Claim). Turns stream off-chain via /v1/chat/message (zero chain per turn). + * "End chat" settles (POST /v1/chat/end → supplier Submit + buyer Accept), + * which is when the user is actually charged. + * + * Both paths stream a uniform SSE wire format over fetch+ReadableStream: + * data: {"type":"token","value":""}\n\n + * data: {"type":"done"}\n\n + * data: {"type":"error","message":"…"}\n\n + */ + +import { useRef, useState } from "react"; +import type { OutputReference } from "@marketplace/shared/chain"; + +export interface ChatFormProps { + /** When set, runs the paid marketplace lifecycle; when undefined, demo mode. */ + advertRef?: OutputReference; + /** Required iff advertRef is set; matches the supplier's advertised price. */ + payment_lovelace?: bigint; +} + +interface ChatTurn { + role: "user" | "assistant"; + content: string; +} + +interface Session { + escrowRef: string; + sessionNonce: string; +} + +function ap3x(lovelace: bigint | string): string { + return (Number(lovelace) / 1e6).toFixed(2); +} + +/** Stream a uniform SSE response, invoking onToken per content delta. + * Resolves when the `done` frame arrives; rejects on `error` frame or + * transport failure. */ +async function streamChat( + url: string, + body: unknown, + onToken: (delta: string) => void, +): Promise { + const resp = await fetch(url, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }); + if (!resp.ok || !resp.body) { + let msg = `${resp.status} ${resp.statusText}`; + try { + const j = (await resp.json()) as { error?: string; message?: string }; + if (j.error || j.message) msg = `${j.error ?? "error"}: ${j.message ?? ""}`; + } catch { /* keep status fallback */ } + throw new Error(msg); + } + const reader = resp.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + const handleFrame = (payload: string): boolean => { + const trimmed = payload.trim(); + if (trimmed.length === 0) return false; + let frame: { type?: string; value?: string; message?: string }; + try { + frame = JSON.parse(trimmed); + } catch { + return false; + } + if (frame.type === "token" && typeof frame.value === "string") { + onToken(frame.value); + } else if (frame.type === "done") { + return true; + } else if (frame.type === "error") { + throw new Error(frame.message ?? "stream error"); + } + return false; + }; + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + let sepIdx: number; + while ((sepIdx = buffer.indexOf("\n\n")) >= 0) { + const block = buffer.slice(0, sepIdx); + buffer = buffer.slice(sepIdx + 2); + for (const line of block.split("\n")) { + if (line.startsWith("data:") && handleFrame(line.slice(5))) return; + } + } + } +} + +export default function ChatForm({ advertRef, payment_lovelace }: ChatFormProps = {}): JSX.Element { + const isPaid = advertRef !== undefined && payment_lovelace !== undefined; + + const [turns, setTurns] = useState([]); + const [streaming, setStreaming] = useState(null); // in-progress assistant text + const [input, setInput] = useState(""); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(null); + + // Paid-only session + finalization state. + const [session, setSession] = useState(null); + const [starting, setStarting] = useState(false); + const [ending, setEnding] = useState(false); + const [charged, setCharged] = useState(null); + const sessionRef = useRef(null); + sessionRef.current = session; + + const active = isPaid ? session !== null && charged === null : true; + + async function startChat(): Promise { + if (!advertRef || payment_lovelace === undefined) return; + setStarting(true); + setError(null); + try { + const resp = await fetch("/v1/chat/start", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + advert_ref: `${advertRef.txHash}#${advertRef.index}`, + payment_lovelace: payment_lovelace.toString(), + }), + }); + if (!resp.ok) { + let msg = `${resp.status} ${resp.statusText}`; + try { + const j = (await resp.json()) as { error?: string; message?: string }; + if (resp.status === 409) msg = "Supplier is busy with another chat — try again shortly."; + else if (j.error || j.message) msg = `${j.error ?? "error"}: ${j.message ?? ""}`; + } catch { /* keep fallback */ } + throw new Error(msg); + } + const j = (await resp.json()) as { escrow_ref: string; session_nonce: string }; + setSession({ escrowRef: j.escrow_ref, sessionNonce: j.session_nonce }); + setTurns([]); + setCharged(null); + } catch (err) { + setError(err instanceof Error ? err.message : String(err)); + } finally { + setStarting(false); + } + } + + async function sendMessage(): Promise { + const content = input.trim(); + if (content.length === 0 || busy) return; + setBusy(true); + setError(null); + setInput(""); + const nextTurns: ChatTurn[] = [...turns, { role: "user", content }]; + setTurns(nextTurns); + setStreaming(""); + let acc = ""; + try { + if (isPaid) { + const s = sessionRef.current; + if (!s) throw new Error("no active chat session"); + await streamChat( + "/v1/chat/message", + { escrow_ref: s.escrowRef, content }, + (delta) => { acc += delta; setStreaming(acc); }, + ); + } else { + // Demo: send the full running transcript (no server-side state). + const demoMessages = nextTurns.map((t) => ({ role: t.role, content: t.content })); + await streamChat( + "/v1/chat-demo/message", + { messages: demoMessages }, + (delta) => { acc += delta; setStreaming(acc); }, + ); + } + setTurns([...nextTurns, { role: "assistant", content: acc }]); + } catch (err) { + setError(err instanceof Error ? err.message : String(err)); + // Keep the user's turn but drop the half-streamed assistant bubble. + } finally { + setStreaming(null); + setBusy(false); + } + } + + async function endChat(): Promise { + const s = sessionRef.current; + if (!s) return; + setEnding(true); + setError(null); + try { + const resp = await fetch("/v1/chat/end", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + escrow_ref: s.escrowRef, + session_nonce: s.sessionNonce, + transcript: turns, + }), + }); + if (!resp.ok) { + let msg = `${resp.status} ${resp.statusText}`; + try { + const j = (await resp.json()) as { error?: string; message?: string }; + if (j.error || j.message) msg = `${j.error ?? "error"}: ${j.message ?? ""}`; + } catch { /* keep fallback */ } + throw new Error(msg); + } + setCharged(payment_lovelace !== undefined ? ap3x(payment_lovelace) : "0"); + } catch (err) { + setError(err instanceof Error ? err.message : String(err)); + } finally { + setEnding(false); + } + } + + return ( +
+ {isPaid && session === null && ( + + )} + + {(active || streaming !== null || turns.length > 0) && ( +
+
+ {turns.map((t, i) => ( +
+ + {t.content} + +
+ ))} + {streaming !== null && ( +
+ + {streaming.length > 0 ? streaming : "…"} + +
+ )} +
+ + {active && ( +
{ e.preventDefault(); void sendMessage(); }} + className="flex gap-2" + > + setInput(e.target.value)} + placeholder="Type a message…" + disabled={busy} + className="flex-1 rounded border border-gray-300 p-2 text-sm focus:border-blue-500 focus:outline-none" + data-testid="chat-input" + /> + +
+ )} + + {isPaid && active && ( + + )} +
+ )} + + {charged !== null && ( +
+ Chat ended — charged {charged} AP3X, bond refunded. +
+ )} + + {error && ( +
+ {error} +
+ )} +
+ ); +} diff --git a/buyer/src/ui/pages/Dashboard.tsx b/buyer/src/ui/pages/Dashboard.tsx index 8549033..c805ec7 100644 --- a/buyer/src/ui/pages/Dashboard.tsx +++ b/buyer/src/ui/pages/Dashboard.tsx @@ -23,6 +23,7 @@ import { useMarketplace } from "../state/MarketplaceContext.js"; import SupplierCard from "../components/SupplierCard.js"; import PromptForm from "../components/PromptForm.js"; import PiperTTSForm from "../components/PiperTTSForm.js"; +import ChatForm from "../components/ChatForm.js"; import type { SupplierView } from "../../sdk/types.js"; function parseUtxoRef(ref: string): { txHash: string; index: number } | null { @@ -61,6 +62,32 @@ const PIPER_DEMO_SUPPLIER: SupplierView = { created_slot: 0, }; +// Synthetic free chat demo — same pattern as PIPER_DEMO_SUPPLIER. The +// "demo:kimi-chat" ref fails parseUtxoRef → advertRef null → ChatForm runs in +// demo mode (streams from OpenRouter via /v1/chat-demo/message, no escrow). +// Delete this constant once a real `llm.chat.v1` supplier with price=0 posts +// an on-chain advert; the discovered entry then wins selection. +const KIMI_DEMO_SUPPLIER: SupplierView = { + utxo_ref: "demo:kimi-chat", + supplier_pkh: "", + capability_id: "llm.chat.v1", + model: "Kimi K2.6 (demo)", + max_output_tokens: 0, + max_processing_ms: 0, + price_lovelace: "0", + supplier_bond_lovelace: "0", + buyer_bond_lovelace: "0", + endpoint_url: "https://openrouter.ai/api", + detail_uri: "", + detail_hash: "", + advertised_at: 0, + status: "demo", + advert_status: "demo", + current_escrow_ref: null, + last_seen_iso: null, + created_slot: 0, +}; + export default function Dashboard() { const marketplace = useMarketplace(); const [suppliers, setSuppliers] = useState([]); @@ -77,15 +104,15 @@ export default function Dashboard() { // If a real audio.synthesize.piper.v1 supplier appears in `list`, // the synthetic entry still shows but the real one wins selection // (cards are deduped by utxo_ref, and "demo:..." can't collide). - setSuppliers([...list, PIPER_DEMO_SUPPLIER]); + setSuppliers([...list, PIPER_DEMO_SUPPLIER, KIMI_DEMO_SUPPLIER]); } }) .catch((err: Error) => { if (!cancelled) { setError(err.message); - // Even when the indexer is unreachable, expose the demo so the + // Even when the indexer is unreachable, expose the demos so the // capability path is testable in isolation. - setSuppliers([PIPER_DEMO_SUPPLIER]); + setSuppliers([PIPER_DEMO_SUPPLIER, KIMI_DEMO_SUPPLIER]); } }); return () => { @@ -161,6 +188,15 @@ function renderCapabilityForm( } return ; + case "llm.chat.v1": + // Real on-chain chat supplier → paid mode (escrow bookends). The + // synthetic demo supplier has utxo_ref="demo:kimi-chat" which fails + // parseUtxoRef → advertRef is null → ChatForm runs in free demo mode. + if (advertRef) { + return ; + } + return ; + default: return ; } diff --git a/deploy/mainnet/docker-compose.supplier-kimi-chat.yml b/deploy/mainnet/docker-compose.supplier-kimi-chat.yml new file mode 100644 index 0000000..f056b4c --- /dev/null +++ b/deploy/mainnet/docker-compose.supplier-kimi-chat.yml @@ -0,0 +1,86 @@ +# deploy/mainnet/docker-compose.supplier-kimi-chat.yml — Supplier — MAINNET +# (OpenRouter / Kimi K2.6 — MULTI-TURN CHAT capability `llm.chat.v1`) +# +# The chat-session sibling of docker-compose.supplier-kimi.yml. Same image, +# same OpenRouter backend (moonshotai/kimi-k2.6), but CAPABILITY_KIND=chat-session +# so it mounts /v1/chat/{start,message,end} (escrow bookends, off-chain +# streaming turns) instead of the one-off /v1/chat/completions route. +# +# A SEPARATE process/advert from the one-off Kimi supplier on purpose: the +# chat session holds the supplier's single-slot lock for the whole conversation +# (minutes), which would starve one-off traffic if co-hosted. +# +# Coexists with: +# - marketplace-mainnet-supplier-kimi (one-off llm.text.generate.v1, mp-suppliers-kimi.vector.apexfusion.org) +# +# Pre-flight: +# 1. DNS A-record: mp-suppliers-kimi-chat.vector.apexfusion.org → host IP +# 2. supplier/.env.kimi-chat 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 +# The advert must be posted with --capability-id llm.chat.v1 and a large +# --max-processing-ms (~1800000) so Claim+Submit both land before deliver_by. +# 3. Networks already exist (created by the canonical supplier compose): +# docker network create marketplace-mainnet-net # one-shot +# # traefik-net is created by the Traefik project + +name: marketplace-mainnet-supplier-kimi-chat + +services: + supplier: + build: + context: ../.. + dockerfile: supplier/Dockerfile + container_name: marketplace-mainnet-supplier-kimi-chat + restart: unless-stopped + env_file: + # Self-contained env for the Kimi-chat variant — separate from + # ../../supplier/.env.kimi (the one-off Kimi supplier). Holds this + # supplier's distinct SUPPLIER_PRIV_KEY_HEX, ADVERT_REF (an llm.chat.v1 + # advert), OPENAI_API_KEY, and the shared mainnet plumbing. + - path: ../../supplier/.env.kimi-chat + 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" + # Auto-end an idle session after 5 min (supplier Submits the transcript). + CHAT_IDLE_TIMEOUT_MS: "300000" + 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 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-mainnet.entrypoints=web" + - "traefik.http.routers.mp-suppliers-kimi-chat-mainnet.rule=Host(`mp-suppliers-kimi-chat.vector.apexfusion.org`)" + - "traefik.http.services.mp-suppliers-kimi-chat-mainnet.loadbalancer.server.port=8080" + logging: + driver: json-file + options: + max-size: "10m" + max-file: "3" + +networks: + marketplace-mainnet-net: + external: true + name: marketplace-mainnet-net + traefik-net: + external: true + name: traefik-net diff --git a/packages/shared/src/tx/escrow/postChatEscrow.ts b/packages/shared/src/tx/escrow/postChatEscrow.ts new file mode 100644 index 0000000..fba3f53 --- /dev/null +++ b/packages/shared/src/tx/escrow/postChatEscrow.ts @@ -0,0 +1,207 @@ +/** + * tx/escrow/postChatEscrow.ts — PostEscrow tx builder for the + * `llm.chat.v1` (multi-turn chat session) capability. + * + * Unlike the one-off `postEscrow.ts` (chat) builder, no conversation exists + * yet at Start-chat time — the buyer opens the escrow BEFORE exchanging any + * messages and the whole conversation then runs off-chain. So we cannot hash + * the messages into `prompt_hash` here. Instead we commit a deterministic + * SESSION-INIT placeholder: + * + * prompt_hash = sha256(canonical({ kind: "llm.chat.v1", session_nonce })) + * + * The on-chain validator never reads `prompt_hash` (it only pins it immutable + * through the state machine — see contracts/marketplace/validators/escrow.ak), + * so the placeholder is sound. At End-chat the supplier commits the real + * transcript hash as `result_receipt_hash` via the normal Submit tx, and the + * buyer verifies the signed receipt off-chain (recomputing this same + * session-init hash from the nonce it remembers). + * + * `chatSessionPromptHash` is exported and shared between the buyer (here) and + * the supplier (server.ts /v1/chat/start), exactly like `ttsPromptHash` — both + * sides must agree on the canonicalisation or the supplier's prompt-mismatch + * gate fails. + * + * Off-chain invariants (mirror chat/TTS): + * 1. advert UTxO exists at advertRef + * 2. advert datum.status === "Active" + * 3. payment_lovelace === advert.price_lovelace + * 4. buyerKey.pubKeyHash !== advert.supplier_pkh + * 5. session_nonce is a non-empty string + */ + +import * as nodeCrypto from "crypto"; +import type { ChainProvider, OutputReference } from "../../chain/ChainProvider.js"; +import type { EscrowDatum } from "../../cbor/types.js"; +import { decodeAdvertDatum } from "../../cbor/AdvertDatum.js"; +import { encodeEscrowDatum } from "../../cbor/EscrowDatum.js"; +import { canonicalize } from "../../cbor/canonical.js"; +import type { WalletKey, PostEscrowBuildResult } from "../types.js"; +import { TxConstructionError } from "../types.js"; +import { loadBlueprint } from "../blueprint.js"; +import { encodeTxBody, sha256Hex } from "../internal/testTxBody.js"; +import { mockSlotToWallclockMs, NETWORK_BUFFER_MS } from "../internal/constants.js"; +import { detectCborBackend } from "../internal/cborBackend.js"; +import type { LiveOgmiosProvider } from "../../chain/LiveOgmiosProvider.js"; + +export interface PostChatEscrowParams { + chain: ChainProvider; + buyerKey: WalletKey; + advertRef: OutputReference; + /** Random hex the buyer generates and remembers for End-chat verification. */ + session_nonce: string; + payment_lovelace: bigint; +} + +function sha256Utf8Hex(s: string): string { + return nodeCrypto.createHash("sha256").update(s, "utf8").digest("hex"); +} + +/** Session-init prompt_hash, shared between buyer (postChatEscrow) and supplier + * (/v1/chat/start). Both sides MUST agree on this exact JCS-canonical encoding. */ +export function chatSessionPromptHash(args: { session_nonce: string }): string { + return sha256Utf8Hex(canonicalize({ + kind: "llm.chat.v1", + session_nonce: args.session_nonce, + })); +} + +export async function buildPostChatEscrowTx( + params: PostChatEscrowParams, +): Promise { + const { chain, buyerKey, advertRef, session_nonce, payment_lovelace } = params; + + // 5. session_nonce must be a non-empty string. + if (typeof session_nonce !== "string" || session_nonce.length === 0) { + throw new TxConstructionError("session_nonce required", + "session_nonce must be a non-empty string"); + } + + // 1. Advert UTxO must exist. + const advertUtxo = await chain.queryUtxo(advertRef); + if (advertUtxo === null) { + throw new TxConstructionError( + "advert ref not on chain", + `no UTxO at ${advertRef.txHash}#${advertRef.index}`, + ); + } + if (!advertUtxo.datumHex) { + throw new TxConstructionError( + "advert datum missing", + `UTxO ${advertRef.txHash}#${advertRef.index} has no inline datum`, + ); + } + + const advertDatum = decodeAdvertDatum(advertUtxo.datumHex); + + // 2. Advert must be Active. + if (advertDatum.status !== "Active") { + throw new TxConstructionError( + "advert is retired", + `advert.status is ${advertDatum.status}, expected Active`, + ); + } + + // 3. Payment must equal advertised price. + if (payment_lovelace !== advertDatum.price_lovelace) { + throw new TxConstructionError( + "payment must equal advertised price", + `payment ${payment_lovelace} != advert.price ${advertDatum.price_lovelace}`, + ); + } + + // 4. Buyer cannot be supplier. + if (buyerKey.pubKeyHash === advertDatum.supplier_pkh) { + throw new TxConstructionError( + "buyer cannot be supplier", + `buyerKey pkh ${buyerKey.pubKeyHash} equals advert.supplier_pkh`, + ); + } + + // Time source: same convention as chat/TTS. + const tipSlot = await chain.tip(); + const isLive = detectCborBackend(chain) === "live"; + const postedAt = isLive ? Date.now() : mockSlotToWallclockMs(tipSlot); + const deliverBy = postedAt + advertDatum.max_processing_ms + NETWORK_BUFFER_MS; + + const requestSpecCanonical = canonicalize({ + capability_id: advertDatum.capability_id, + max_output_tokens: advertDatum.max_output_tokens, + model: advertDatum.model, + }); + const requestSpecHash = sha256Utf8Hex(requestSpecCanonical); + const promptHash = chatSessionPromptHash({ session_nonce }); + + const totalLocked = + advertDatum.price_lovelace + + advertDatum.buyer_bond_lovelace + + advertDatum.supplier_bond_lovelace; + + const escrowDatum: EscrowDatum = { + buyer_pkh: buyerKey.pubKeyHash, + supplier_pkh: advertDatum.supplier_pkh, + advert_ref: advertRef, + capability_id: advertDatum.capability_id, + request_spec_hash: requestSpecHash, + prompt_hash: promptHash, + payment_lovelace: advertDatum.price_lovelace, + buyer_bond_lovelace: advertDatum.buyer_bond_lovelace, + supplier_bond_lovelace: advertDatum.supplier_bond_lovelace, + deliver_by: deliverBy, + posted_at: postedAt, + submitted_at: null, + result_receipt_hash: null, + state: "Open", + }; + + // Live path: reuse the capability-agnostic chat builder (messages = []). + // The on-chain commitment is the escrowDatum (carrying our session-init + // prompt_hash); messages are only used for the mock cbor body, which we omit. + if (isLive) { + const liveCborPath = "../internal/liveCbor.js"; + const { buildLiveTxForEscrow } = await import(/* @vite-ignore */ liveCborPath); + return buildLiveTxForEscrow({ + chain: chain as LiveOgmiosProvider, + buyerKey, + advertRef, + messages: [], + escrowDatum, + totalLocked, + deliverBy, + postedAt, + }); + } + + // Mock backend path (synthetic JSON-in-hex tx body). + const blueprint = loadBlueprint(); + const escrowAddress = blueprint.escrowScriptAddress(0); + + const body = { + type: "post-escrow", + inputs: [], + outputs: [ + { + ref: { txHash: "$self", index: 0 }, + address: escrowAddress, + lovelace: totalLocked, + assets: {}, + datumHex: encodeEscrowDatum(escrowDatum), + scriptRef: null, + }, + ], + requiredSigners: [buyerKey.pubKeyHash], + validityRange: { lowerBoundMs: postedAt, upperBoundMs: deliverBy }, + meta: { script_hash: blueprint.escrowScriptHash, advert_ref: advertRef }, + }; + + const txCborHex = encodeTxBody(body); + const expectedTxHash = sha256Hex(txCborHex); + + await chain.submitTx(txCborHex); + + return { + txCborHex, + expectedTxHash, + escrowOutputRef: { txHash: expectedTxHash, index: 0 }, + }; +} diff --git a/packages/shared/src/tx/index.ts b/packages/shared/src/tx/index.ts index 2e967a1..0fc4cf7 100644 --- a/packages/shared/src/tx/index.ts +++ b/packages/shared/src/tx/index.ts @@ -34,6 +34,8 @@ export { ALLOWED_TTS_VOICES, ALLOWED_TTS_FORMATS, } from "./escrow/postTtsEscrow.js"; +export type { PostChatEscrowParams } from "./escrow/postChatEscrow.js"; +export { buildPostChatEscrowTx, chatSessionPromptHash } from "./escrow/postChatEscrow.js"; export type { ClaimParams } from "./escrow/claim.js"; export { buildClaimTx } from "./escrow/claim.js"; export type { SubmitParams } from "./escrow/submit.js"; diff --git a/supplier/.env.kimi-chat.example b/supplier/.env.kimi-chat.example new file mode 100644 index 0000000..1843ef3 --- /dev/null +++ b/supplier/.env.kimi-chat.example @@ -0,0 +1,62 @@ +# supplier/.env.kimi-chat.example +# +# Env for the MAINNET Kimi K2.6 MULTI-TURN CHAT supplier (capability +# `llm.chat.v1`, CAPABILITY_KIND=chat-session). Copy to supplier/.env.kimi-chat +# on the host (chmod 600) and fill in real values. .env.kimi-chat is gitignored. +# +# This is a SEPARATE wallet/advert from supplier/.env.kimi (the one-off Kimi +# supplier) — the chat session holds the single-slot lock for the whole +# conversation, so it runs as its own process. + +# ─── Required ───────────────────────────────────────────────────────────── + +# 64-char hex Ed25519 private key seed for THIS supplier's wallet. SECRET. +SUPPLIER_PRIV_KEY_HEX=PASTE_64_HEX_CHARS_HERE + +# OpenRouter API key (Bearer). SECRET. +OPENAI_API_KEY=PASTE_OPENROUTER_KEY_HERE + +# Ogmios WebSocket URL (mainnet). +OGMIOS_URL=wss://ogmios.vector.mainnet.apexfusion.org + +# OutputReference of this supplier's on-chain `llm.chat.v1` advert UTxO. +# Post it first (see tx:post-advert below), then paste "#0" here. +ADVERT_REF=PASTE_TXHASH_HERE#0 + +# Mainnet. +NETWORK_ID=1 + +# Enable real chain submit/await (Claim + Submit). +LIVE_CHAIN=1 + +# ─── Backend (chat-session is openai-only) ──────────────────────────────── +# CAPABILITY_KIND, LLM_BACKEND, OPENAI_BASE_URL, OPENAI_TIMEOUT_MS, and +# CHAT_IDLE_TIMEOUT_MS are pinned in docker-compose.supplier-kimi-chat.yml. +# Override CHAT_IDLE_TIMEOUT_MS here only if you want a different idle window. + +# ─── tx:post-advert — run once before first deploy ──────────────────────── +# Post an llm.chat.v1 advert with a LARGE max-processing-ms so Claim (start) +# and Submit (end) both land before deliver_by: +# pnpm --filter @marketplace/supplier tx:post-advert \ +# --capability-id llm.chat.v1 \ +# --model moonshotai/kimi-k2.6 \ +# --max-output-tokens 4096 \ +# --max-processing-ms 1800000 \ +# --price-lovelace 2000000 \ +# --endpoint-url https://mp-suppliers-kimi-chat.vector.apexfusion.org +# Copy the printed "#0" into ADVERT_REF above. +ADVERT_CAPABILITY_ID=llm.chat.v1 +ADVERT_MODEL=moonshotai/kimi-k2.6 +ADVERT_MAX_OUTPUT_TOKENS=4096 +ADVERT_MAX_PROCESSING_MS=1800000 +ADVERT_PRICE_LOVELACE=2000000 +ADVERT_SUPPLIER_BOND_LOVELACE=1000000 +ADVERT_BUYER_BOND_LOVELACE=1000000 +ADVERT_ENDPOINT_URL=https://mp-suppliers-kimi-chat.vector.apexfusion.org +ADVERT_DETAIL_URI= +ADVERT_DETAIL_HASH= + +# ─── CIP-33 reference scripts (recommended on mainnet) ──────────────────── +# ESCROW_REF_UTXO=#0 +# ADVERT_REF_UTXO=#1 +# VECTOR_ZERO_TIME_MS= diff --git a/supplier/src/chatSession.ts b/supplier/src/chatSession.ts new file mode 100644 index 0000000..50173dd --- /dev/null +++ b/supplier/src/chatSession.ts @@ -0,0 +1,127 @@ +/** + * supplier/src/chatSession.ts — in-memory store for open multi-turn chat + * sessions (capabilityKind="chat-session", capability_id="llm.chat.v1"). + * + * Unlike the one-off JobStore (one request → one terminal payload), a chat + * session is long-lived: it's created at /v1/chat/start (after the supplier + * Claims the escrow), accumulates a transcript across many /v1/chat/message + * turns that run fully OFF-CHAIN, and is settled at /v1/chat/end (or by the + * idle watchdog) when the supplier Submits a receipt over the whole transcript. + * + * Single-slot: the supplier serves one paid chat at a time (SupplierState + * mutex), so at most one record is "active". Ended records are retained so + * /v1/chat/end is idempotent (returns the same receipt). + * + * Timer handles (idle + hard-cap) live on the record; server.ts arms them and + * endChatSession clears them. + */ + +import type { OutputReference } from "@marketplace/shared/chain"; +import type { AdvertDatum, EscrowDatum } from "@marketplace/shared/cbor"; +import type { ChatMessage } from "@marketplace/shared/tx"; + +export type ChatSessionStatus = "active" | "ending" | "ended"; + +export interface ChatSessionEndResult { + receipt: Record; + receipt_signature: string; + /** The Submitted escrow UTxO ("#") the buyer must Accept. */ + submitted_ref: string; +} + +export interface ChatSessionRecord { + escrowRef: string; + claimedRef: OutputReference; + advert: AdvertDatum; + escrowDatum: EscrowDatum; + /** Full ordered conversation (user + assistant turns). Hashed at End. */ + transcript: ChatMessage[]; + promptTokens: number; + completionTokens: number; + startedAtMs: number; + lastActivityMs: number; + status: ChatSessionStatus; + idleTimer?: ReturnType; + hardCapTimer?: ReturnType; + endResult?: ChatSessionEndResult; + endFailure?: { reason: string; message: string }; +} + +export interface CreateChatSessionParams { + escrowRef: string; + claimedRef: OutputReference; + advert: AdvertDatum; + escrowDatum: EscrowDatum; +} + +export class ChatSessionStore { + private readonly records = new Map(); + + create(params: CreateChatSessionParams): ChatSessionRecord { + const now = Date.now(); + const record: ChatSessionRecord = { + escrowRef: params.escrowRef, + claimedRef: params.claimedRef, + advert: params.advert, + escrowDatum: params.escrowDatum, + transcript: [], + promptTokens: 0, + completionTokens: 0, + startedAtMs: now, + lastActivityMs: now, + status: "active", + }; + this.records.set(params.escrowRef, record); + return record; + } + + get(escrowRef: string): ChatSessionRecord | null { + return this.records.get(escrowRef) ?? null; + } + + /** True iff there is a record in "active" or "ending" state for this ref. */ + isOpen(escrowRef: string): boolean { + const r = this.records.get(escrowRef); + return r !== undefined && r.status !== "ended"; + } + + appendUser(escrowRef: string, content: string): void { + const r = this.records.get(escrowRef); + if (!r) return; + r.transcript.push({ role: "user", content }); + r.lastActivityMs = Date.now(); + } + + appendAssistant( + escrowRef: string, + content: string, + usage: { prompt_tokens: number; completion_tokens: number }, + ): void { + const r = this.records.get(escrowRef); + if (!r) return; + r.transcript.push({ role: "assistant", content }); + r.promptTokens += usage.prompt_tokens; + r.completionTokens += usage.completion_tokens; + r.lastActivityMs = Date.now(); + } + + touch(escrowRef: string): void { + const r = this.records.get(escrowRef); + if (r) r.lastActivityMs = Date.now(); + } + + clearTimers(record: ChatSessionRecord): void { + if (record.idleTimer) { + clearTimeout(record.idleTimer); + record.idleTimer = undefined; + } + if (record.hardCapTimer) { + clearTimeout(record.hardCapTimer); + record.hardCapTimer = undefined; + } + } + + count(): number { + return this.records.size; + } +} diff --git a/supplier/src/chatSessionRunner.ts b/supplier/src/chatSessionRunner.ts new file mode 100644 index 0000000..5b351be --- /dev/null +++ b/supplier/src/chatSessionRunner.ts @@ -0,0 +1,120 @@ +/** + * supplier/src/chatSessionRunner.ts — settles an open chat session. + * + * endChatSession is the chat-session analogue of runChatJob's receipt+Submit + * tail (jobRunner.ts steps 3-7). It is invoked from BOTH /v1/chat/end (user + * clicked "End chat") and the idle/hard-cap watchdog (abandoned session). It: + * 1. atomically flips the session "active" → "ending" (idempotent re-entry) + * 2. clears the idle + hard-cap timers + * 3. builds + signs a receipt over the FULL transcript + * (response_hash = sha256(canonical(transcript)); the buyer recomputes + * this from its local transcript mirror to verify) + * 4. buildSubmitTx against the Claimed UTxO → awaitTx (60s) + * 5. stores the terminal result and ALWAYS releases the supplier lock + * + * Never throws. On any failure the session is marked "ended" with endFailure + * and the lock is released (the escrow then lingers Claimed until the buyer + * Reclaims after deliver_by — same fate as a failed one-off job). + */ + +import { canonicalize } from "@marketplace/shared/cbor"; +import { buildSubmitTx } from "@marketplace/shared/tx"; +import { buildReceipt, signReceipt, receiptResultHash } from "@marketplace/shared/receipt"; +import { createHash } from "crypto"; + +import type { RunChatJobDeps } from "./jobRunner.js"; +import type { ChatSessionStore, ChatSessionRecord } from "./chatSession.js"; + +function sha256Hex(s: string): string { + return createHash("sha256").update(s, "utf8").digest("hex"); +} + +export interface EndChatSessionDeps extends RunChatJobDeps { + sessions: ChatSessionStore; +} + +export interface EndChatSessionParams { + deps: EndChatSessionDeps; + escrowRef: string; + /** Identifies what triggered the end, for logs ("end" | "idle" | "hard-cap"). */ + trigger?: string; +} + +/** + * Settle a chat session. Returns the (now terminal) record, or null if no + * session exists for the ref. Idempotent: a second call returns the same + * record without re-submitting. + */ +export async function endChatSession( + params: EndChatSessionParams, +): Promise { + const { deps, escrowRef, trigger } = params; + const record = deps.sessions.get(escrowRef); + if (!record) return null; + + // Single-threaded between awaits: this check-then-set is atomic enough to + // dedupe concurrent end/idle triggers. + if (record.status !== "active") return record; + record.status = "ending"; + deps.sessions.clearTimers(record); + + try { + // ── Receipt over the full transcript ────────────────────────────────── + const transcriptHash = sha256Hex(canonicalize(record.transcript)); + const receipt = buildReceipt({ + prompt_hash: record.escrowDatum.prompt_hash, // session-init placeholder + response_hash: transcriptHash, + model: record.advert.model, + prompt_tokens: record.promptTokens, + completion_tokens: record.completionTokens, + wallclock_ms: Math.max(0, Date.now() - record.startedAtMs), + supplier_pkh: deps.supplierKey.pubKeyHash, + escrow_ref: record.escrowRef, + }); + const signed = signReceipt(receipt, deps.supplierKey.privateKeyHex); + const resultHash = receiptResultHash(signed); + + // ── Submit (Claimed → Submitted) ────────────────────────────────────── + let buildResult: { txCborHex: string; expectedTxHash: string }; + try { + buildResult = await buildSubmitTx({ + chain: deps.chain, + supplierKey: deps.supplierKey, + escrowRef: record.claimedRef, + receiptHash: resultHash, + }); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + console.warn(`[chat_end_failed] escrow=${escrowRef} trigger=${trigger ?? "end"} reason=submit_failed msg=${message}`); + record.endFailure = { reason: "submit_failed", message: `Submit tx failed: ${message}` }; + record.status = "ended"; + return record; + } + + try { + await deps.chain.awaitTx(buildResult.expectedTxHash, 60_000); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + console.warn(`[chat_end_failed] escrow=${escrowRef} trigger=${trigger ?? "end"} reason=submit_timeout msg=${message}`); + record.endFailure = { reason: "submit_timeout", message: `Submit awaitTx failed: ${message}` }; + record.status = "ended"; + return record; + } + + record.endResult = { + receipt: signed.receipt as unknown as Record, + receipt_signature: signed.signature, + // The Submit tx output (index 0) is the new Submitted UTxO the buyer Accepts. + submitted_ref: `${buildResult.expectedTxHash}#0`, + }; + record.status = "ended"; + return record; + } finally { + // Always release the single-slot lock so the next chat can start. + try { + deps.state.release(); + } catch { + /* never let lock-release escape */ + } + } +} diff --git a/supplier/src/config.ts b/supplier/src/config.ts index 6ab45e1..e11b7b6 100644 --- a/supplier/src/config.ts +++ b/supplier/src/config.ts @@ -56,8 +56,12 @@ export interface AdvertRef { index: number; } -/** Capability the supplier is configured to serve. */ -export type CapabilityKind = "chat" | "tts"; +/** Capability the supplier is configured to serve. + * "chat" — one-off `llm.text.generate.v1` (single prompt → single completion) + * "tts" — `audio.synthesize.piper.v1` + * "chat-session" — multi-turn `llm.chat.v1` (escrow bookends, off-chain turns, + * streaming). Requires LLM_BACKEND="openai". */ +export type CapabilityKind = "chat" | "tts" | "chat-session"; /** LLM backend selected for the chat capability. */ export type LlmBackend = "ollama" | "openai"; @@ -85,6 +89,13 @@ export interface SupplierConfig { piperTimeoutMs: number; capabilityKind: CapabilityKind; llmBackend: LlmBackend; + /** + * Idle-timeout for an open chat session, ms. A session with no message + * activity for this long is auto-ended (supplier Submits the transcript + * receipt). Only meaningful for capabilityKind="chat-session". Default + * 300_000 (5 min). + */ + chatIdleTimeoutMs: number; /** * When true, the supplier boots a LiveOgmiosProvider (real submitTx/awaitTx). * Default false → ReadOnlyOgmiosProvider (safe; no chain writes). @@ -136,8 +147,8 @@ export function loadConfig(env: Record): SupplierCon // CAPABILITY_KIND drives whether OLLAMA_URL or PIPER_URL is required. // Default to "chat" so existing chat suppliers keep booting unchanged. const capKindStr = env.CAPABILITY_KIND ?? "chat"; - if (capKindStr !== "chat" && capKindStr !== "tts") { - throw new Error('loadConfig: CAPABILITY_KIND must be "chat" or "tts"'); + if (capKindStr !== "chat" && capKindStr !== "tts" && capKindStr !== "chat-session") { + throw new Error('loadConfig: CAPABILITY_KIND must be "chat", "tts", or "chat-session"'); } const capabilityKind: CapabilityKind = capKindStr; @@ -157,7 +168,16 @@ export function loadConfig(env: Record): SupplierCon let ollamaUrl = ""; let openaiBaseUrl = ""; let piperUrl = ""; - if (capabilityKind === "chat") { + if (capabilityKind === "chat-session") { + // Multi-turn chat streams via callOpenAiStream, which only speaks the + // OpenAI-compatible wire format — so chat-session is openai-only. + if (llmBackend !== "openai") { + throw new Error('loadConfig: CAPABILITY_KIND="chat-session" requires LLM_BACKEND="openai"'); + } + openaiBaseUrl = requireField(env, "OPENAI_BASE_URL"); + ollamaUrl = env.OLLAMA_URL ?? ""; + piperUrl = env.PIPER_URL ?? ""; + } else if (capabilityKind === "chat") { if (llmBackend === "openai") { openaiBaseUrl = requireField(env, "OPENAI_BASE_URL"); ollamaUrl = env.OLLAMA_URL ?? ""; @@ -210,6 +230,16 @@ export function loadConfig(env: Record): SupplierCon const openaiApiKey = env.OPENAI_API_KEY ?? ""; + // CHAT_IDLE_TIMEOUT_MS — idle auto-end window for chat sessions (default 5 min). + const chatIdleStr = env.CHAT_IDLE_TIMEOUT_MS; + let chatIdleTimeoutMs = 300_000; + if (chatIdleStr !== undefined && chatIdleStr !== "") { + if (!POS_INT_RE.test(chatIdleStr)) { + throw new Error("loadConfig: CHAT_IDLE_TIMEOUT_MS must be a positive integer"); + } + chatIdleTimeoutMs = Number(chatIdleStr); + } + const piperTimeoutStr = env.PIPER_TIMEOUT_MS; let piperTimeoutMs = 120_000; if (piperTimeoutStr !== undefined && piperTimeoutStr !== "") { @@ -249,6 +279,7 @@ export function loadConfig(env: Record): SupplierCon piperTimeoutMs, capabilityKind, llmBackend, + chatIdleTimeoutMs, liveChain, walletHealthIntervalMs, }; diff --git a/supplier/src/openai.ts b/supplier/src/openai.ts index d0b894d..e8dbdb1 100644 --- a/supplier/src/openai.ts +++ b/supplier/src/openai.ts @@ -167,3 +167,163 @@ export async function callOpenAi(params: CallOpenAiParams): Promise void, +): Promise { + const { baseUrl, model, messages, timeoutMs, apiKey } = params; + const url = `${baseUrl}/v1/chat/completions`; + const body = JSON.stringify({ + model, + messages, + stream: true, + stream_options: { include_usage: true }, + }); + + const headers: Record = { + "content-type": "application/json", + accept: "text/event-stream", + }; + if (apiKey) { + headers.authorization = `Bearer ${apiKey}`; + } + + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + + const startedAt = Date.now(); + let response: Response; + try { + response = await fetch(url, { + method: "POST", + headers, + body, + signal: controller.signal, + }); + } catch (err) { + clearTimeout(timer); + if (isAbortError(err)) { + throw new OpenAiError("openai_timeout", `OpenAI request exceeded ${timeoutMs}ms`); + } + throw new OpenAiError( + "openai_failure", + `OpenAI fetch failed: ${(err as Error)?.message ?? String(err)}`, + ); + } + + if (!response.ok) { + clearTimeout(timer); + let detail = ""; + try { + detail = await response.text(); + } catch { + // ignore — body unavailable + } + throw new OpenAiError( + "openai_failure", + `OpenAI returned HTTP ${response.status}${detail ? `: ${detail.slice(0, 200)}` : ""}`, + ); + } + if (!response.body) { + clearTimeout(timer); + throw new OpenAiError("openai_malformed", "OpenAI streaming response had no body"); + } + + let accumulated = ""; + let promptTokens = 0; + let completionTokens = 0; + let buffer = ""; + + const handleData = (payload: string): void => { + const trimmed = payload.trim(); + if (trimmed.length === 0 || trimmed === "[DONE]") return; + let parsed: unknown; + try { + parsed = JSON.parse(trimmed); + } catch { + // Skip non-JSON keepalive/comment frames. + return; + } + if (!parsed || typeof parsed !== "object") return; + const obj = parsed as Record; + const choices = obj.choices; + if (Array.isArray(choices) && choices.length > 0) { + const delta = (choices[0] as Record)?.delta as + | Record + | undefined; + const content = delta?.content; + if (typeof content === "string" && content.length > 0) { + accumulated += content; + onToken(content); + } + } + const usageRaw = obj.usage; + if (usageRaw && typeof usageRaw === "object") { + const usage = usageRaw as Record; + if (typeof usage.prompt_tokens === "number") promptTokens = usage.prompt_tokens; + if (typeof usage.completion_tokens === "number") completionTokens = usage.completion_tokens; + } + }; + + try { + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + // SSE frames are separated by a blank line. + let sepIdx: number; + while ((sepIdx = buffer.indexOf("\n\n")) >= 0) { + const frame = buffer.slice(0, sepIdx); + buffer = buffer.slice(sepIdx + 2); + for (const line of frame.split("\n")) { + if (line.startsWith("data:")) handleData(line.slice(5)); + } + } + } + // Flush any trailing frame without a final blank line. + if (buffer.length > 0) { + for (const line of buffer.split("\n")) { + if (line.startsWith("data:")) handleData(line.slice(5)); + } + } + } catch (err) { + if (isAbortError(err)) { + throw new OpenAiError("openai_timeout", `OpenAI stream exceeded ${timeoutMs}ms`); + } + throw new OpenAiError( + "openai_failure", + `OpenAI stream read failed: ${(err as Error)?.message ?? String(err)}`, + ); + } finally { + clearTimeout(timer); + } + + if (accumulated.length === 0) { + throw new OpenAiError("openai_malformed", "OpenAI stream produced no content"); + } + + return { + content: accumulated, + prompt_tokens: promptTokens, + completion_tokens: completionTokens, + wallclock_ms: Date.now() - startedAt, + }; +} diff --git a/supplier/src/server.ts b/supplier/src/server.ts index 2dd6067..836adf1 100644 --- a/supplier/src/server.ts +++ b/supplier/src/server.ts @@ -38,12 +38,15 @@ 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, mockSlotToWallclockMs, detectCborBackend } from "@marketplace/shared/tx"; +import { buildClaimTx, chatSessionPromptHash, mockSlotToWallclockMs, detectCborBackend } from "@marketplace/shared/tx"; import type { SupplierState } from "./state.js"; import type { SupplierConfig } from "./config.js"; import { JobStore } from "./jobs.js"; import { runChatJob, runTtsJob } from "./jobRunner.js"; +import { ChatSessionStore, type ChatSessionRecord } from "./chatSession.js"; +import { endChatSession, type EndChatSessionDeps } from "./chatSessionRunner.js"; +import { callOpenAiStream } from "./openai.js"; import { healthzRouter } from "./routes/healthz.js"; import { triggerOnFailureConsolidate } from "./walletHealth.js"; @@ -54,6 +57,8 @@ export interface SupplierDeps { supplierKey: WalletKey; /** Optional: defaults to a fresh JobStore. M1-F-async-chat. */ jobs?: JobStore; + /** Optional: defaults to a fresh ChatSessionStore (chat-session capability). */ + chatSessions?: ChatSessionStore; } const ESCROW_REF_RE = /^[0-9a-fA-F]{64}#(?:0|[1-9]\d*)$/; @@ -83,6 +88,7 @@ interface ResolvedDeps { config: SupplierConfig; supplierKey: WalletKey; jobs: JobStore; + chatSessions: ChatSessionStore; } function makeCapabilityHandler(deps: ResolvedDeps) { @@ -646,6 +652,272 @@ function makeGetTtsJobHandler(deps: ResolvedDeps) { }; } +// ─── /v1/chat/{start,message,end} — multi-turn chat session (llm.chat.v1) ─── +// +// Escrow bookends with off-chain turns: +// start → validate (mirror makeChatHandler 1-7, but no messages) → Claim +// (Open→Claimed) → create session + arm idle/hard-cap watchdog → 200 +// message → SSE stream a turn via callOpenAiStream; zero chain interaction +// end → endChatSession (Submit transcript receipt; Claimed→Submitted) → 200 +// +// The buyer then Accepts off this route (server-side, in the buyer-app), +// which is when the user is actually charged. Single-slot: the SupplierState +// mutex is held from Claim (start) to Submit (end), so one paid chat at a time. + +interface ChatStartBody { + model?: unknown; + session_nonce?: unknown; +} + +function makeChatSessionHandlers(deps: ResolvedDeps) { + const endDeps: EndChatSessionDeps = { + chain: deps.chain, + state: deps.state, + config: deps.config, + supplierKey: deps.supplierKey, + jobs: deps.jobs, + sessions: deps.chatSessions, + }; + + function armIdleTimer(record: ChatSessionRecord): void { + if (record.idleTimer) clearTimeout(record.idleTimer); + record.idleTimer = setTimeout(() => { + void endChatSession({ deps: endDeps, escrowRef: record.escrowRef, trigger: "idle" }); + }, deps.config.chatIdleTimeoutMs); + record.idleTimer.unref?.(); + } + + function armHardCapTimer(record: ChatSessionRecord, nowMs: number): void { + // Force-end before deliver_by so the Submit tx (60s awaitTx budget) lands + // in time. 90s margin = Submit awaitTx + buffer. + const margin = 90_000; + const capMs = Math.max(1_000, record.escrowDatum.deliver_by - nowMs - margin); + record.hardCapTimer = setTimeout(() => { + void endChatSession({ deps: endDeps, escrowRef: record.escrowRef, trigger: "hard-cap" }); + }, capMs); + record.hardCapTimer.unref?.(); + } + + const start = async (req: Request, res: Response, next: NextFunction) => { + try { + // ── 1. Header ─────────────────────────────────────────────────── + const headerVal = req.header("X-Escrow-Ref"); + if (!headerVal) { + return jsonError(res, 400, "escrow_ref_required", "X-Escrow-Ref header is required"); + } + const escrowRef = parseEscrowRef(headerVal); + if (escrowRef === null) { + return jsonError(res, 400, "escrow_ref_malformed", 'X-Escrow-Ref must match "<64-hex>#"'); + } + const escrowRefStr = `${escrowRef.txHash}#${escrowRef.index}`; + + // ── 2. Body (session_nonce; no messages) ───────────────────────── + const body = (req.body ?? {}) as ChatStartBody; + const sessionNonce = typeof body.session_nonce === "string" ? body.session_nonce : ""; + if (sessionNonce.length === 0) { + return jsonError(res, 400, "session_nonce_required", "body.session_nonce must be a non-empty string"); + } + + // ── 3. Advert ─────────────────────────────────────────────────── + const advertResult = await fetchActiveAdvert(deps); + if ("error" in advertResult) { + const e = advertResult.error; + return jsonError(res, e.status, e.reason, e.message); + } + const advert = advertResult.datum; + + // ── 4. Escrow ─────────────────────────────────────────────────── + const escrowUtxo = await deps.chain.queryUtxo(escrowRef); + if (escrowUtxo === null || !escrowUtxo.datumHex) { + return jsonError(res, 404, "escrow_not_found", `escrow UTxO ${escrowRefStr} not found on chain`); + } + let escrowDatum: EscrowDatum; + try { + escrowDatum = decodeEscrowDatum(escrowUtxo.datumHex); + } catch (err) { + return jsonError(res, 404, "escrow_decode_failed", (err as Error).message); + } + + // ── 5. State / identity / capability ───────────────────────────── + if (escrowDatum.state !== "Open") { + return jsonError(res, 409, "escrow_not_claimable", `escrow state is ${escrowDatum.state}, expected Open`); + } + if (escrowDatum.supplier_pkh !== deps.supplierKey.pubKeyHash) { + return jsonError(res, 403, "wrong_supplier", "escrow supplier_pkh does not match this node"); + } + if (escrowDatum.capability_id !== advert.capability_id) { + return jsonError(res, 409, "capability_mismatch", + `escrow capability ${escrowDatum.capability_id} != advert ${advert.capability_id}`); + } + + // ── 6. Hash checks (request spec + session-init prompt) ────────── + const expectedRequestSpecHash = sha256Hex(canonicalize({ + capability_id: advert.capability_id, + max_output_tokens: advert.max_output_tokens, + model: advert.model, + })); + if (escrowDatum.request_spec_hash !== expectedRequestSpecHash) { + return jsonError(res, 409, "request_spec_mismatch", "request_spec_hash in escrow does not match advert spec"); + } + const expectedPromptHash = chatSessionPromptHash({ session_nonce: sessionNonce }); + if (escrowDatum.prompt_hash !== expectedPromptHash) { + return jsonError(res, 409, "prompt_mismatch", "prompt_hash in escrow does not match session_nonce"); + } + + // ── 7. Deadline ────────────────────────────────────────────────── + const tipSlot = await deps.chain.tip(); + const isLive = detectCborBackend(deps.chain) === "live"; + const nowMs = isLive + ? Date.now() + : Math.max(mockSlotToWallclockMs(tipSlot), escrowDatum.posted_at); + if (nowMs >= escrowDatum.deliver_by) { + return jsonError(res, 408, "past_deliver_by", `now ${nowMs} >= deliver_by ${escrowDatum.deliver_by}`); + } + + // ── 8. Single-slot lock ────────────────────────────────────────── + if (!deps.state.tryAcquire(escrowRefStr)) { + return jsonError(res, 409, "supplier_busy", "supplier is already working another chat"); + } + + // ── 9. Claim (Open → Claimed) — hold the lock on success ───────── + let claimResult; + try { + claimResult = await buildClaimTx({ chain: deps.chain, supplierKey: deps.supplierKey, escrowRef }); + } catch (err) { + deps.state.release(); + triggerOnFailureConsolidate({ chain: deps.chain, state: deps.state, supplierKey: deps.supplierKey }); + return jsonError(res, 503, "chain_submit_failed", `Claim tx submit failed: ${(err as Error).message}`); + } + try { + await deps.chain.awaitTx(claimResult.expectedTxHash, 60_000); + } catch (err) { + deps.state.release(); + return jsonError(res, 504, "claim_timeout", `Claim awaitTx failed: ${(err as Error).message}`); + } + + // ── 10. Create session + arm watchdog ──────────────────────────── + const claimedRef: OutputReference = { txHash: claimResult.expectedTxHash, index: 0 }; + const record = deps.chatSessions.create({ escrowRef: escrowRefStr, claimedRef, advert, escrowDatum }); + armIdleTimer(record); + armHardCapTimer(record, nowMs); + + return res.status(200).json({ status: "claimed", escrow_ref: escrowRefStr }); + } catch (err) { + try { deps.state.release(); } catch { /* ignore */ } + next(err); + return; + } + }; + + const message = async (req: Request, res: Response, next: NextFunction) => { + try { + const headerVal = req.header("X-Escrow-Ref"); + if (!headerVal) { + return jsonError(res, 400, "escrow_ref_required", "X-Escrow-Ref header is required"); + } + const escrowRef = parseEscrowRef(headerVal); + if (escrowRef === null) { + return jsonError(res, 400, "escrow_ref_malformed", 'X-Escrow-Ref must match "<64-hex>#"'); + } + const escrowRefStr = `${escrowRef.txHash}#${escrowRef.index}`; + const record = deps.chatSessions.get(escrowRefStr); + if (!record || record.status !== "active") { + return jsonError(res, 404, "chat_session_not_found", + `no active chat session for ${escrowRefStr}`); + } + const body = (req.body ?? {}) as { content?: unknown }; + const content = typeof body.content === "string" ? body.content : ""; + if (content.length === 0) { + return jsonError(res, 400, "content_required", "body.content must be a non-empty string"); + } + + // 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); + + 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"); + if (typeof (res as Response & { flushHeaders?: () => void }).flushHeaders === "function") { + (res as Response & { flushHeaders: () => void }).flushHeaders(); + } + const sse = (frame: Record): void => { + res.write(`data: ${JSON.stringify(frame)}\n\n`); + }; + + try { + const result = await callOpenAiStream( + { + baseUrl: deps.config.openaiBaseUrl, + model: record.advert.model, + messages: record.transcript, + timeoutMs: deps.config.openaiTimeoutMs, + apiKey: deps.config.openaiApiKey, + }, + (delta) => sse({ type: "token", value: delta }), + ); + deps.chatSessions.appendAssistant(escrowRefStr, result.content, { + prompt_tokens: result.prompt_tokens, + completion_tokens: result.completion_tokens, + }); + sse({ type: "done" }); + } catch (err) { + const errMsg = err instanceof Error ? err.message : String(err); + sse({ type: "error", message: errMsg }); + } finally { + if (record.status === "active") armIdleTimer(record); + res.end(); + } + return; + } catch (err) { + next(err); + return; + } + }; + + const end = async (req: Request, res: Response, next: NextFunction) => { + try { + const headerVal = req.header("X-Escrow-Ref"); + if (!headerVal) { + return jsonError(res, 400, "escrow_ref_required", "X-Escrow-Ref header is required"); + } + const escrowRef = parseEscrowRef(headerVal); + if (escrowRef === null) { + return jsonError(res, 400, "escrow_ref_malformed", 'X-Escrow-Ref must match "<64-hex>#"'); + } + const escrowRefStr = `${escrowRef.txHash}#${escrowRef.index}`; + if (!deps.chatSessions.get(escrowRefStr)) { + return jsonError(res, 404, "chat_session_not_found", `no chat session for ${escrowRefStr}`); + } + const record = await endChatSession({ deps: endDeps, escrowRef: escrowRefStr, trigger: "end" }); + if (!record) { + return jsonError(res, 404, "chat_session_not_found", `no chat session for ${escrowRefStr}`); + } + if (record.endFailure) { + return jsonError(res, 502, record.endFailure.reason, record.endFailure.message); + } + if (!record.endResult) { + return jsonError(res, 500, "chat_end_incomplete", "session ended without a receipt"); + } + return res.status(200).json({ + status: "submitted", + escrow_ref: escrowRefStr, + submitted_ref: record.endResult.submitted_ref, + receipt: record.endResult.receipt, + receipt_signature: record.endResult.receipt_signature, + }); + } catch (err) { + next(err); + return; + } + }; + + return { start, message, end }; +} + // ─── App factory ─────────────────────────────────────────────────────────── export function createApp(deps: SupplierDeps): Application { @@ -655,6 +927,7 @@ export function createApp(deps: SupplierDeps): Application { config: deps.config, supplierKey: deps.supplierKey, jobs: deps.jobs ?? new JobStore(), + chatSessions: deps.chatSessions ?? new ChatSessionStore(), }; const app = express(); @@ -677,6 +950,11 @@ export function createApp(deps: SupplierDeps): Application { if (resolved.config.capabilityKind === "tts") { app.post("/v1/audio/synthesize", makeTtsHandler(resolved)); app.get("/v1/audio/synthesize/:jobId", makeGetTtsJobHandler(resolved)); + } else if (resolved.config.capabilityKind === "chat-session") { + const chat = makeChatSessionHandlers(resolved); + app.post("/v1/chat/start", chat.start); + app.post("/v1/chat/message", chat.message); + app.post("/v1/chat/end", chat.end); } else { app.post("/v1/chat/completions", makeChatHandler(resolved)); app.get("/v1/chat/completions/:jobId", makeGetJobHandler(resolved)); diff --git a/tests/unit/buyer-ui-smoke.test.tsx b/tests/unit/buyer-ui-smoke.test.tsx index ac9847e..2248b8e 100644 --- a/tests/unit/buyer-ui-smoke.test.tsx +++ b/tests/unit/buyer-ui-smoke.test.tsx @@ -126,17 +126,22 @@ describe(" smoke", () => { ]); render(wrap(, mp)); // Wait for async discover to complete and cards to render. Dashboard - // appends a synthetic Piper-TTS demo card to whatever the indexer - // returned, so the count is N+1 until the on-chain TTS supplier ships. + // appends two synthetic demo cards (Piper-TTS + Kimi K2.6 chat) to whatever + // the indexer returned, so the count is N+2 until on-chain suppliers ship. await waitFor(() => { const cards = document.querySelectorAll("[data-testid='supplier-card']"); - expect(cards.length).toBe(3); + expect(cards.length).toBe(4); }); // Sanity: one of the cards is the demo Piper supplier. const piperBadges = Array.from( document.querySelectorAll("[data-testid='supplier-card']"), ).filter((c) => c.textContent?.includes("audio.synthesize.piper.v1")); expect(piperBadges.length).toBe(1); + // Sanity: one of the cards is the demo Kimi K2.6 chat supplier. + const kimiBadges = Array.from( + document.querySelectorAll("[data-testid='supplier-card']"), + ).filter((c) => c.textContent?.includes("llm.chat.v1")); + expect(kimiBadges.length).toBe(1); }); });