feat: llm.chat.v1 multi-turn supplier type + free Kimi K2.6 chat demo
Some checks failed
CI / typecheck + vitest (Node 20) (push) Has been cancelled

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) <noreply@anthropic.com>
This commit is contained in:
david
2026-06-08 16:18:54 +02:00
parent e01a03b7f2
commit 9acad5b2b8
17 changed files with 1997 additions and 13 deletions

View File

@@ -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<string, string | undefined>): 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<string, string | undefined>): BuyerConfig
ogmiosUrl,
liveChain,
ttsPiperBaseUrl,
openrouterApiKey,
openrouterBaseUrl,
archiveDir,
password,
sessionSecret,

View File

@@ -127,6 +127,8 @@ export async function runMain(env: Record<string, string | undefined>): Promise<
indexerUrl: config.indexerUrl,
marketplace,
ttsPiperBaseUrl: config.ttsPiperBaseUrl,
openrouterApiKey: config.openrouterApiKey,
openrouterBaseUrl: config.openrouterBaseUrl,
archive,
password: config.password,
sessionSecret: config.sessionSecret,

View File

@@ -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<StartChatResult> {
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<EndChatResult> {
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<void> {

View File

@@ -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().
*/

View File

@@ -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":"<delta>"}\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<string, unknown>): 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<Uint8Array>).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<string, unknown>)?.choices;
if (Array.isArray(choices) && choices.length > 0) {
const delta = (choices[0] as Record<string, unknown>)?.delta as
| Record<string, unknown>
| 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<string, { supplierBaseUrl: string }>();
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>#<index>" }');
}
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>#<index>" }');
}
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<string, unknown>): 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<Uint8Array>).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>#<index>" }');
}
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,

View File

@@ -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":"<delta>"}\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<void> {
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<ChatTurn[]>([]);
const [streaming, setStreaming] = useState<string | null>(null); // in-progress assistant text
const [input, setInput] = useState("");
const [busy, setBusy] = useState(false);
const [error, setError] = useState<string | null>(null);
// Paid-only session + finalization state.
const [session, setSession] = useState<Session | null>(null);
const [starting, setStarting] = useState(false);
const [ending, setEnding] = useState(false);
const [charged, setCharged] = useState<string | null>(null);
const sessionRef = useRef<Session | null>(null);
sessionRef.current = session;
const active = isPaid ? session !== null && charged === null : true;
async function startChat(): Promise<void> {
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<void> {
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<void> {
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 (
<div className="space-y-4 rounded border border-gray-200 bg-white p-4" data-testid="chat-form">
{isPaid && session === null && (
<button
type="button"
onClick={startChat}
disabled={starting}
className="rounded bg-blue-600 px-4 py-2 text-sm font-medium text-white hover:bg-blue-700 disabled:bg-gray-400"
data-testid="chat-start"
>
{starting ? "Opening escrow…" : "Start chat"}
</button>
)}
{(active || streaming !== null || turns.length > 0) && (
<div className="space-y-3">
<div className="max-h-96 space-y-2 overflow-auto" data-testid="chat-transcript">
{turns.map((t, i) => (
<div key={i} className={t.role === "user" ? "text-right" : "text-left"}>
<span
className={
"inline-block max-w-[85%] whitespace-pre-wrap rounded px-3 py-2 text-sm " +
(t.role === "user"
? "bg-blue-600 text-white"
: "bg-gray-100 text-gray-900")
}
>
{t.content}
</span>
</div>
))}
{streaming !== null && (
<div className="text-left">
<span className="inline-block max-w-[85%] whitespace-pre-wrap rounded bg-gray-100 px-3 py-2 text-sm text-gray-900">
{streaming.length > 0 ? streaming : "…"}
</span>
</div>
)}
</div>
{active && (
<form
onSubmit={(e) => { e.preventDefault(); void sendMessage(); }}
className="flex gap-2"
>
<input
type="text"
value={input}
onChange={(e) => 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"
/>
<button
type="submit"
disabled={busy || input.trim().length === 0}
className="rounded bg-blue-600 px-4 py-2 text-sm font-medium text-white hover:bg-blue-700 disabled:bg-gray-400"
data-testid="chat-send"
>
{busy ? "…" : "Send"}
</button>
</form>
)}
{isPaid && active && (
<button
type="button"
onClick={endChat}
disabled={ending || busy}
className="rounded bg-gray-700 px-4 py-2 text-sm font-medium text-white hover:bg-gray-800 disabled:bg-gray-400"
data-testid="chat-end"
>
{ending ? "Finalizing payment…" : "End chat"}
</button>
)}
</div>
)}
{charged !== null && (
<div className="rounded border border-green-300 bg-green-50 p-3 text-sm text-green-700" data-testid="chat-charged">
Chat ended — charged {charged} AP3X, bond refunded.
</div>
)}
{error && (
<div className="rounded border border-red-300 bg-red-50 p-3 text-sm text-red-700" data-testid="chat-error">
{error}
</div>
)}
</div>
);
}

View File

@@ -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<SupplierView[]>([]);
@@ -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 <PiperTTSForm />;
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 <ChatForm advertRef={advertRef} payment_lovelace={payment_lovelace} />;
}
return <ChatForm />;
default:
return <UnsupportedCapability capabilityId={supplier.capability_id} />;
}

View File

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

View File

@@ -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<PostEscrowBuildResult> {
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 },
};
}

View File

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

View File

@@ -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 "<txHash>#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 "<txHash>#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=<txhash>#0
# ADVERT_REF_UTXO=<txhash>#1
# VECTOR_ZERO_TIME_MS=<epoch-ms>

127
supplier/src/chatSession.ts Normal file
View File

@@ -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<string, unknown>;
receipt_signature: string;
/** The Submitted escrow UTxO ("<txHash>#<index>") 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<typeof setTimeout>;
hardCapTimer?: ReturnType<typeof setTimeout>;
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<string, ChatSessionRecord>();
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;
}
}

View File

@@ -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<ChatSessionRecord | null> {
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<string, unknown>,
receipt_signature: signed.signature,
// The Submit tx output (index 0) is the new Submitted UTxO the buyer Accepts.
submitted_ref: `${buildResult.expectedTxHash}#0`,
};
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 */
}
}
}

View File

@@ -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<string, string | undefined>): 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<string, string | undefined>): 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<string, string | undefined>): 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<string, string | undefined>): SupplierCon
piperTimeoutMs,
capabilityKind,
llmBackend,
chatIdleTimeoutMs,
liveChain,
walletHealthIntervalMs,
};

View File

@@ -167,3 +167,163 @@ export async function callOpenAi(params: CallOpenAiParams): Promise<OpenAiResult
wallclock_ms: wallclockMs,
};
}
/**
* callOpenAiStream — streaming variant of callOpenAi for the chat-session
* supplier. POSTs with `stream: true` and invokes `onToken(delta)` for each
* content delta as it arrives, then returns the SAME OpenAiResult shape
* (full accumulated content + token usage) so receipt-building code is shared.
*
* Parses the OpenAI/OpenRouter SSE wire format:
* data: {"choices":[{"delta":{"content":"tok"}}]}\n\n
* ...
* data: {"choices":[],"usage":{prompt_tokens,completion_tokens,...}}\n\n (include_usage)
* data: [DONE]\n\n
*
* Error reasons match callOpenAi (openai_failure / openai_timeout / openai_malformed).
* The timeout bounds the WHOLE stream (AbortController), matching the non-stream path.
*/
export async function callOpenAiStream(
params: CallOpenAiParams,
onToken: (delta: string) => void,
): Promise<OpenAiResult> {
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<string, string> = {
"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<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 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<string, unknown>;
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,
};
}

View File

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

View File

@@ -126,17 +126,22 @@ describe("<Dashboard /> smoke", () => {
]);
render(wrap(<Dashboard />, 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);
});
});