diff --git a/buyer/src/sdk/Marketplace.ts b/buyer/src/sdk/Marketplace.ts index 2fb8e91..f61805e 100644 --- a/buyer/src/sdk/Marketplace.ts +++ b/buyer/src/sdk/Marketplace.ts @@ -90,6 +90,7 @@ import type { SubmitTtsResult, StartChatOptions, StartChatResult, + ChatSettleMode, EndChatOptions, EndChatResult, AcceptResultOptions, @@ -899,7 +900,13 @@ export class Marketplace extends EventEmitterBase { } this.emitProgress({ type: "chat_started", escrow_ref: escrowRefStr }); - return { escrowRef: escrowOutputRef, sessionNonce, supplierBaseUrl: advertDatum.endpoint_url }; + // Ticket-mode suppliers answer {status:"ticket"} (no Claim); anything else + // — "claimed" or an older supplier without the field — is full settlement. + const startStatus = startRes.body && typeof startRes.body === "object" + ? (startRes.body as { status?: string }).status + : undefined; + const settleMode: ChatSettleMode = startStatus === "ticket" ? "ticket" : "full"; + return { escrowRef: escrowOutputRef, sessionNonce, supplierBaseUrl: advertDatum.endpoint_url, settleMode }; } /** @@ -937,10 +944,20 @@ export class Marketplace extends EventEmitterBase { }); } const endBody = endRes.body as { + status?: string; + settle_mode?: string; receipt?: import("@marketplace/shared/receipt").Receipt; receipt_signature?: string; submitted_ref?: string; }; + // Ticket sessions produce no receipt and nothing to Accept — the Open + // escrow returns via reclaim after deliver_by. Trusting the supplier's + // response is incentive-safe: lying in either direction only forfeits the + // supplier's own payout (the buyer's sweeper reclaims unsettled escrows). + if (endBody.settle_mode === "ticket") { + this.emitProgress({ type: "chat_ended", escrow_ref: escrowRefStr }); + return { settleMode: "ticket", escrowRef }; + } if (!endBody.receipt || !endBody.receipt_signature) { throw new SupplierError("malformed_response", { message: "supplier end response missing receipt" }); } @@ -1022,7 +1039,7 @@ export class Marketplace extends EventEmitterBase { this.emitProgress({ type: "accept_submitted", escrow_ref: refToString(acceptedRef) }); this.emitProgress({ type: "chat_ended", escrow_ref: escrowRefStr }); - return { receipt, receiptSignature, escrowRef, acceptedRef }; + return { settleMode: "full", receipt, receiptSignature, escrowRef, acceptedRef }; } // ─── acceptResult / reclaim ─────────────────────────────────────────── diff --git a/buyer/src/sdk/index.ts b/buyer/src/sdk/index.ts index 56a37c6..a854482 100644 --- a/buyer/src/sdk/index.ts +++ b/buyer/src/sdk/index.ts @@ -16,6 +16,9 @@ export type { AcceptResultOptions, ReclaimOptions, GetTaskHistoryOptions, + ChatSettleMode, + StartChatResult, + EndChatResult, } from "./types.js"; export { ReceiptVerificationError, IndexerError, SupplierError } from "./types.js"; export type { TaskHistoryStore } from "./history.js"; diff --git a/buyer/src/sdk/types.ts b/buyer/src/sdk/types.ts index 17ca788..1ddad06 100644 --- a/buyer/src/sdk/types.ts +++ b/buyer/src/sdk/types.ts @@ -197,6 +197,11 @@ export interface StartChatOptions { advertRef: OutputReference; payment_lovelace: bigint; } +/** Chain settlement mode a chat session runs under (reported by the supplier + * at /v1/chat/start): "full" = Claim/Submit/Accept lifecycle; "ticket" = the + * buyer's escrow post is the only chain op (reclaimed after deliver_by). */ +export type ChatSettleMode = "full" | "ticket"; + export interface StartChatResult { escrowRef: OutputReference; /** Random nonce committed into the escrow's session-init prompt_hash. The @@ -204,6 +209,8 @@ export interface StartChatResult { sessionNonce: string; /** Supplier endpoint_url, cached by the buyer-app for message routing. */ supplierBaseUrl: string; + /** How the supplier settles this session ("full" when it Claimed). */ + settleMode: ChatSettleMode; } /** @@ -220,13 +227,21 @@ export interface EndChatOptions { /** The browser's local transcript mirror, used to verify response_hash. */ transcript?: import("@marketplace/shared/tx").ChatMessage[]; } -export interface EndChatResult { - receipt: Receipt; - receiptSignature: string; - escrowRef: OutputReference; - /** The Submitted escrow UTxO that was Accepted. */ - acceptedRef: OutputReference; -} +export type EndChatResult = + | { + settleMode: "full"; + receipt: Receipt; + receiptSignature: string; + escrowRef: OutputReference; + /** The Submitted escrow UTxO that was Accepted. */ + acceptedRef: OutputReference; + } + | { + /** Ticket session: nothing was Submitted/Accepted — no receipt exists. + * The Open escrow returns to the buyer via reclaim after deliver_by. */ + settleMode: "ticket"; + escrowRef: OutputReference; + }; /** * AcceptResultOptions — arguments to Marketplace.acceptResult(). diff --git a/buyer/src/server.ts b/buyer/src/server.ts index ae85aaf..904d1ea 100644 --- a/buyer/src/server.ts +++ b/buyer/src/server.ts @@ -934,6 +934,14 @@ export function createApp(deps: AppDeps): Express { supplierBaseUrl: state.supplierBaseUrl, }); chatRouteState.delete(rawRef); + if (result.settleMode === "ticket") { + // Ticket session: no receipt/Accept — funds return via reclaim. + return res.status(200).json({ + status: "ended", + escrow_ref: rawRef, + settle_mode: "ticket", + }); + } return res.status(200).json({ status: "accepted", escrow_ref: rawRef, diff --git a/buyer/src/ui/components/ChatForm.tsx b/buyer/src/ui/components/ChatForm.tsx index ab20996..9c08084 100644 --- a/buyer/src/ui/components/ChatForm.tsx +++ b/buyer/src/ui/components/ChatForm.tsx @@ -267,7 +267,18 @@ export default function ChatForm({ advertRef, payment_lovelace }: ChatFormProps } catch { /* keep fallback */ } throw new Error(msg); } - setCharged(payment_lovelace !== undefined ? ap3x(payment_lovelace) : "0"); + let endBody: { settle_mode?: string } = {}; + try { + endBody = (await resp.json()) as { settle_mode?: string }; + } catch { /* older buyer-app responses stay full-settle */ } + // Ticket sessions never charge — the escrow returns via reclaim. + setCharged( + endBody.settle_mode === "ticket" + ? "0 (ticket session — escrow reclaimed)" + : payment_lovelace !== undefined + ? ap3x(payment_lovelace) + : "0", + ); } catch (err) { setError(err instanceof Error ? err.message : String(err)); } finally { diff --git a/deploy/mainnet/docker-compose.gateway.yml b/deploy/mainnet/docker-compose.gateway.yml index 57dcde5..e9652a5 100644 --- a/deploy/mainnet/docker-compose.gateway.yml +++ b/deploy/mainnet/docker-compose.gateway.yml @@ -38,6 +38,13 @@ services: # frontend's "API Keys" page (served from marketplace.vector…) POSTs to # /signup here. Comma-separated; empty disables CORS entirely. GATEWAY_CORS_ORIGINS: https://marketplace.vector.apexfusion.org + # Ticket mode: the escrow post at session open is the only chain op; + # the sweeper reclaims Open escrows after deliver_by. Must match the + # kimi-chat suppliers' CHAT_SETTLE_MODE. + CHAT_SETTLE_MODE: ticket + # Demo janitor closes idle sessions BEFORE the suppliers' 15 min + # CHAT_IDLE_TIMEOUT_MS backstop. + DEMO_SESSION_IDLE_MS: "600000" volumes: - gateway-mainnet-data:/repo/data/gateway healthcheck: diff --git a/deploy/mainnet/docker-compose.supplier-kimi-chat-2.yml b/deploy/mainnet/docker-compose.supplier-kimi-chat-2.yml index fac05db..48e9aa6 100644 --- a/deploy/mainnet/docker-compose.supplier-kimi-chat-2.yml +++ b/deploy/mainnet/docker-compose.supplier-kimi-chat-2.yml @@ -40,7 +40,9 @@ services: OPENAI_TIMEOUT_MS: "300000" # 2.5 min idle end: single-slot suppliers must free up quickly for the # shared demo endpoint (the gateway transparently reopens on next turn). - CHAT_IDLE_TIMEOUT_MS: "150000" + CHAT_IDLE_TIMEOUT_MS: "900000" + CHAT_SETTLE_MODE: ticket + MAX_CHAT_SESSIONS: "8" PORT: "8080" # Enable LiveOgmiosProvider so Claim/Submit actually broadcast. LIVE_CHAIN: "1" diff --git a/deploy/mainnet/docker-compose.supplier-kimi-chat-3.yml b/deploy/mainnet/docker-compose.supplier-kimi-chat-3.yml index 260e4cf..c1f6a71 100644 --- a/deploy/mainnet/docker-compose.supplier-kimi-chat-3.yml +++ b/deploy/mainnet/docker-compose.supplier-kimi-chat-3.yml @@ -40,7 +40,9 @@ services: OPENAI_TIMEOUT_MS: "300000" # 2.5 min idle end: single-slot suppliers must free up quickly for the # shared demo endpoint (the gateway transparently reopens on next turn). - CHAT_IDLE_TIMEOUT_MS: "150000" + CHAT_IDLE_TIMEOUT_MS: "900000" + CHAT_SETTLE_MODE: ticket + MAX_CHAT_SESSIONS: "8" PORT: "8080" # Enable LiveOgmiosProvider so Claim/Submit actually broadcast. LIVE_CHAIN: "1" diff --git a/deploy/mainnet/docker-compose.supplier-kimi-chat.yml b/deploy/mainnet/docker-compose.supplier-kimi-chat.yml index ae1a751..4b97686 100644 --- a/deploy/mainnet/docker-compose.supplier-kimi-chat.yml +++ b/deploy/mainnet/docker-compose.supplier-kimi-chat.yml @@ -52,7 +52,9 @@ services: # 2.5 min idle end (supplier Submits the transcript): single-slot # suppliers must free up quickly for the shared demo endpoint (the # gateway transparently reopens a session on the next turn). - CHAT_IDLE_TIMEOUT_MS: "150000" + CHAT_IDLE_TIMEOUT_MS: "900000" + CHAT_SETTLE_MODE: ticket + MAX_CHAT_SESSIONS: "8" PORT: "8080" # Enable LiveOgmiosProvider so Claim/Submit actually broadcast. LIVE_CHAIN: "1" diff --git a/docs/gateway.md b/docs/gateway.md index de48f10..812e05b 100644 --- a/docs/gateway.md +++ b/docs/gateway.md @@ -113,6 +113,12 @@ All on-chain steps run inside the per-key `mutex.run(...)`. - **close** → `endChat` (Submit + Accept, awaits) → settle + bond refund; record usage; wallet-health. - Sessions are reclaimable by the sweeper if abandoned before close. +### Settle modes (`CHAT_SETTLE_MODE`, gateway + supplier, default `full`) +- **full** — the lifecycle above: supplier Claims at start, Submits a transcript receipt at end, gateway Accepts (buyer charged). +- **ticket** — the buyer's escrow post at open is the ONLY chain op. The supplier verifies the Open escrow as an entry ticket (`/v1/chat/start` → `{status:"ticket"}`) but never Claims/Submits; `/v1/chat/end` → `{status:"ended", settle_mode:"ticket"}` with no receipt; the sweeper Reclaims the Open escrow after `deliver_by`, returning all locked funds (net cost ≈ tx fees). Usage rows record `cost_lovelace: 0`. Sessions may outlive `deliver_by` (no hard-cap) — a live session whose ticket got reclaimed keeps running; only the idle janitor/close ends it. Close behavior is driven per-session by the supplier's `/end` response, so mixed-mode rollout windows settle correctly. +- Ticket caveats (accepted, demo-grade): a supplier restart forgets used tickets, so an unexpired un-reclaimed escrow ref could reopen a session (bounded by `deliver_by` + the sweep); `/account.locked_in_escrow_lovelace` over-reports because the indexer never updates spent rows. +- Suppliers additionally take `MAX_CHAT_SESSIONS` (default 1) — concurrent session slots; `/status` reports `active_sessions`/`max_sessions` and stays `free` until all slots are taken. + ## Sweeper & wallet-health - **Sweeper** (interval ~60s): per key, classify escrows. Open/Claimed past `deliver_by` → `reclaim`. Submitted **with a verified receipt** still in accept-window → retry `acceptResult`. Submitted past accept-window → leave for supplier Release (log). Never blind-`reclaim` Submitted. - **Wallet-health tick**: periodically + after each settle, `planConsolidate` → `consolidateWallet` so every wallet keeps a ≥5 ADA collateral UTxO. Skip if already healthy. diff --git a/gateway/.env.example b/gateway/.env.example index 6e75ec9..b0322ee 100644 --- a/gateway/.env.example +++ b/gateway/.env.example @@ -31,5 +31,6 @@ GATEWAY_DB_DIR=./data/gateway # KEY_RATE_WINDOW_MS=60000 # SWEEPER_INTERVAL_MS=60000 # DEMO_SESSION_IDLE_MS=180000 # close demo chat sessions idle this long (frees the supplier slot) +# CHAT_SETTLE_MODE=full # "full" (Claim/Submit/Accept) or "ticket" (escrow post only; sweeper reclaims) — must match suppliers # WALLET_HEALTH_INTERVAL_MS=600000 # SDK_REGISTRY_MAX=500 # max cached per-key Marketplace instances diff --git a/gateway/src/config.ts b/gateway/src/config.ts index fb77463..f67017a 100644 --- a/gateway/src/config.ts +++ b/gateway/src/config.ts @@ -41,10 +41,15 @@ export interface GatewayConfig { keyRate: { max: number; windowMs: number }; demoIpRate: { max: number; windowMs: number }; sweeperIntervalMs: number; - /** Close demo chat sessions idle this long (frees the single-slot supplier). - * Keep below the supplier's CHAT_IDLE_TIMEOUT_MS (default 5 min) so the - * gateway closes first and records usage. */ + /** Close demo chat sessions idle this long (frees the supplier slot). + * Keep below the supplier's CHAT_IDLE_TIMEOUT_MS so the gateway closes + * first and records usage. */ demoSessionIdleMs: number; + /** Chat-session settlement mode this deployment runs. "full" = suppliers + * Claim/Submit and the gateway Accepts; "ticket" = the escrow post at open + * is the only chain op and the sweeper Reclaims it after deliver_by. + * Must agree with the suppliers' CHAT_SETTLE_MODE. */ + chatSettleMode: "full" | "ticket"; walletHealthIntervalMs: number; sdkRegistryMax: number; corsOrigins: string[]; @@ -58,6 +63,14 @@ function requireField(env: Record, name: string): st return v; } +function chatSettleMode(env: Record): "full" | "ticket" { + const v = env.CHAT_SETTLE_MODE ?? "full"; + if (v !== "full" && v !== "ticket") { + throw new Error('loadConfig: CHAT_SETTLE_MODE must be "full" or "ticket"'); + } + return v; +} + function posInt( env: Record, name: string, @@ -122,6 +135,7 @@ export function loadConfig(env: Record): GatewayConf }, sweeperIntervalMs: posInt(env, "SWEEPER_INTERVAL_MS", 60 * 1000), demoSessionIdleMs: posInt(env, "DEMO_SESSION_IDLE_MS", 3 * 60 * 1000), + chatSettleMode: chatSettleMode(env), walletHealthIntervalMs: posInt(env, "WALLET_HEALTH_INTERVAL_MS", 10 * 60 * 1000), sdkRegistryMax: posInt(env, "SDK_REGISTRY_MAX", 500), corsOrigins: (env.GATEWAY_CORS_ORIGINS ?? "") diff --git a/gateway/src/onchain/settle.ts b/gateway/src/onchain/settle.ts index 630e6a9..bd4046b 100644 --- a/gateway/src/onchain/settle.ts +++ b/gateway/src/onchain/settle.ts @@ -22,6 +22,8 @@ export interface EscrowRow { utxo_ref: string; state: string; posted_at: number; + /** Validator reclaim floor (indexer serves it; absent on very old rows). */ + deliver_by?: number; } export async function fetchEscrows( diff --git a/gateway/src/openai/demoChat.ts b/gateway/src/openai/demoChat.ts index 435d0f1..2a87380 100644 --- a/gateway/src/openai/demoChat.ts +++ b/gateway/src/openai/demoChat.ts @@ -60,8 +60,10 @@ function candidatesFor(keyId: string, model: string): AffinityCandidate[] { return out; } -/** Record the session's flat fee when the demo executor closes its row (the - * supplier already Submitted; the sweeper Accepts the escrow afterwards). */ +/** Record the session's cost when the demo executor closes its row. Full + * mode: the flat session fee (supplier Submitted; the sweeper Accepts the + * escrow afterwards). Ticket mode: 0 — nothing settles, the sweeper reclaims + * the escrow and the wallet is made whole minus fees. */ function recordSessionCost(deps: GatewayDeps, session: SessionRow): void { deps.store.insertUsage({ id: randomUUID(), @@ -72,7 +74,7 @@ function recordSessionCost(deps: GatewayDeps, session: SessionRow): void { capability_id: CAPABILITY, supplier_pkh: session.supplier_pkh, escrow_ref: session.escrow_ref, - cost_lovelace: session.price_lovelace, + cost_lovelace: deps.config.chatSettleMode === "ticket" ? "0" : session.price_lovelace, prompt_tokens: 0, completion_tokens: 0, status: "completed", diff --git a/gateway/src/openai/sessions.ts b/gateway/src/openai/sessions.ts index ad5c4a6..61bbd38 100644 --- a/gateway/src/openai/sessions.ts +++ b/gateway/src/openai/sessions.ts @@ -125,6 +125,16 @@ export async function openSessionCore( } if (!started || !used) throw toGatewayError(lastErr ?? new Error("no chat supplier could be engaged")); + // Deployment sanity: supplier and gateway must agree on the settle mode. + // Close behavior is driven per-session by the supplier's /end response, so + // a mismatch (rollout window) still settles correctly — just warn. + if ((started.settleMode === "ticket") !== (deps.config.chatSettleMode === "ticket")) { + // eslint-disable-next-line no-console + console.warn( + `[gateway] settle-mode mismatch: supplier ${used.supplierPkh.slice(0, 8)} ran "${started.settleMode}" but gateway is configured "${deps.config.chatSettleMode}"`, + ); + } + const sessionId = randomUUID(); deps.store.insertSession({ id: sessionId, @@ -399,6 +409,33 @@ async function closeSession( deps.store.setSessionState(session.id, "closed", Date.now()); dropSessionState(session.id); + if (result.settleMode === "ticket") { + // Nothing was Submitted/Accepted: no receipt, no charge (the escrow + // returns via the sweeper's reclaim), and no wallet churn to tidy up. + deps.store.insertUsage({ + id: randomUUID(), + key_id: keyRow.id, + created_at: Date.now(), + kind: "chat_session", + model: session.model, + capability_id: CAPABILITY, + supplier_pkh: session.supplier_pkh, + escrow_ref: session.escrow_ref, + cost_lovelace: "0", + prompt_tokens: 0, + completion_tokens: 0, + status: "completed", + failure_reason: null, + }); + res.status(200).json({ + status: "ended", + escrow_ref: session.escrow_ref, + settle_mode: "ticket", + model: session.model, + }); + return; + } + const prompt = result.receipt.prompt_tokens ?? 0; const completion = result.receipt.completion_tokens ?? 0; deps.store.insertUsage({ diff --git a/gateway/src/sweeper.ts b/gateway/src/sweeper.ts index 119ee3a..149caaa 100644 --- a/gateway/src/sweeper.ts +++ b/gateway/src/sweeper.ts @@ -25,6 +25,17 @@ * (terminal spends update nothing), and the validator preserves posted_at * across states — so a session's lineage is "rows sharing the posted_at of the * session's original Open ref row". + * + * Because terminal spends never update indexer rows, every settled escrow + * would be retried (one doomed chain query each) on every sweep forever. The + * in-process `settledRefs` memory skips refs this process has already settled + * or found spent; a restart just causes one harmless retry burst. + * + * Ticket mode (config.chatSettleMode === "ticket"): session escrows are never + * Claimed — they sit Open until reclaimed here after deliver_by. Reclaiming + * an Open ticket does NOT close a still-open session row: nothing re-checks + * the escrow after /v1/chat/start, so the conversation (and its affinity + * mirror) lives on, bounded only by the idle janitor. */ import { randomUUID } from "crypto"; @@ -42,6 +53,23 @@ const MIN_AGE_MS = 90_000; const RECLAIM_AFTER_MS = 11 * 60 * 1000; // Chat-session escrows: deliver_by ≈ posted_at + 30 min, so wait longer. const CHAT_RECLAIM_AFTER_MS = 35 * 60 * 1000; +// Reclaim only fires once the validator allows it (validity lower bound >= +// deliver_by); attempting earlier is a guaranteed-doomed tx build. +const DELIVER_BY_BUFFER_MS = 60_000; + +/** Refs this process already settled (Accepted/Reclaimed) or found spent. + * The indexer never updates terminal rows, so without this every historical + * escrow is retried each sweep. Bounded FIFO; restart = one retry burst. */ +const settledRefs = new Set(); +const SETTLED_REFS_MAX = 5_000; + +function markSettled(ref: string): void { + settledRefs.add(ref); + if (settledRefs.size > SETTLED_REFS_MAX) { + const first = settledRefs.values().next().value; + if (first !== undefined) settledRefs.delete(first); + } +} function log(msg: string): void { // eslint-disable-next-line no-console @@ -77,7 +105,16 @@ function closeSessionRow( log(`closed abandoned session ${session.id} (${outcome.status})`); } -export async function runSweepOnce(deps: GatewayDeps): Promise { +/** Chain-settle functions, injectable for tests. */ +export interface SweepSettleFns { + accept: typeof acceptAndConfirm; + reclaim: typeof reclaimAndConfirm; +} + +export async function runSweepOnce( + deps: GatewayDeps, + settle: SweepSettleFns = { accept: acceptAndConfirm, reclaim: reclaimAndConfirm }, +): Promise { // Idle demo sessions first: closing them makes the supplier Submit, so this // sweep (or the next) can Accept the escrow right after. await sweepIdleDemoSessions(deps).catch((err) => @@ -106,6 +143,7 @@ export async function runSweepOnce(deps: GatewayDeps): Promise { for (const row of rows) { const ref = parseRef(row.utxo_ref); if (!ref) continue; + if (settledRefs.has(row.utxo_ref)) continue; const ageMs = now - (row.posted_at ?? 0); if (ageMs < MIN_AGE_MS) continue; const chatSession = sessionByPostedAt.get(row.posted_at); @@ -113,8 +151,9 @@ export async function runSweepOnce(deps: GatewayDeps): Promise { try { if (row.state === "Submitted") { try { - await ctx.mutex.run(() => acceptAndConfirm(deps.chain, ctx.walletKey, ref)); + await ctx.mutex.run(() => settle.accept(deps.chain, ctx.walletKey, ref)); log(`accepted stranded Submitted escrow ${row.utxo_ref}`); + markSettled(row.utxo_ref); if (chatSession) { closeSessionRow(deps, chatSession, { status: "completed", costLovelace: chatSession.price_lovelace }); } @@ -124,8 +163,11 @@ export async function runSweepOnce(deps: GatewayDeps): Promise { // For session lineages, confirm the spend on-chain before closing // the row — an Accept that failed for another reason (timeout) // must leave the session open for the next sweep. - if (chatSession && (await deps.chain.queryUtxo(ref)) === null) { - closeSessionRow(deps, chatSession, { status: "completed", costLovelace: chatSession.price_lovelace }); + if ((await deps.chain.queryUtxo(ref)) === null) { + markSettled(row.utxo_ref); + if (chatSession) { + closeSessionRow(deps, chatSession, { status: "completed", costLovelace: chatSession.price_lovelace }); + } } else if (!chatSession) { // Preserve the historical quiet-skip for one-shot escrows. void err; @@ -133,12 +175,25 @@ export async function runSweepOnce(deps: GatewayDeps): Promise { } } else if (row.state === "Open" || row.state === "Claimed") { const reclaimAfterMs = chatSession ? CHAT_RECLAIM_AFTER_MS : RECLAIM_AFTER_MS; - if (ageMs > reclaimAfterMs) { - await ctx.mutex.run(() => reclaimAndConfirm(deps.chain, ctx.walletKey, ref)); + // The validator only allows Reclaim at/after deliver_by — skip + // attempts that cannot succeed yet (early-closed ticket sessions). + const pastDeliverBy = + typeof row.deliver_by === "number" ? now > row.deliver_by + DELIVER_BY_BUFFER_MS : true; + if (ageMs > reclaimAfterMs && pastDeliverBy) { + await ctx.mutex.run(() => settle.reclaim(deps.chain, ctx.walletKey, ref)); log(`reclaimed stranded ${row.state} escrow ${row.utxo_ref}`); + markSettled(row.utxo_ref); if (chatSession) { - // Funds returned — the session never settled; record no cost. - closeSessionRow(deps, chatSession, { status: "reclaimed", costLovelace: null }); + const ticketOpen = deps.config.chatSettleMode === "ticket" && row.state === "Open"; + if (ticketOpen) { + // Ticket escrow reclaimed out from under a LIVE session — + // by design. Keep the session row + mirror; only the idle + // janitor / close paths end ticket sessions. + log(`ticket mode: escrow reclaimed, session ${chatSession.id} stays open`); + } else { + // Funds returned — the session never settled; record no cost. + closeSessionRow(deps, chatSession, { status: "reclaimed", costLovelace: null }); + } } } } diff --git a/gateway/test/http.test.ts b/gateway/test/http.test.ts index 11c61f9..7bb3e8f 100644 --- a/gateway/test/http.test.ts +++ b/gateway/test/http.test.ts @@ -22,7 +22,11 @@ const SUPPLIERS = [ function makeDeps( fetchFn?: typeof globalThis.fetch, - overrides?: { demoIpRate?: { max: number; windowMs: number }; chain?: GatewayDeps["chain"] }, + overrides?: { + demoIpRate?: { max: number; windowMs: number }; + chain?: GatewayDeps["chain"]; + chatSettleMode?: "full" | "ticket"; + }, ): GatewayDeps { const dbDir = join(tmpdir(), `gw-test-${randomUUID()}`); const store = new GatewayStore(dbDir); @@ -39,6 +43,7 @@ function makeDeps( demoIpRate: overrides?.demoIpRate ?? { max: 1000, windowMs: 60_000 }, sweeperIntervalMs: 60_000, demoSessionIdleMs: 180_000, + chatSettleMode: overrides?.chatSettleMode ?? "full", walletHealthIntervalMs: 600_000, sdkRegistryMax: 100, corsOrigins: [], @@ -417,6 +422,60 @@ describe("gateway demo key", () => { expect(endCalls).toHaveLength(0); }); + it("ticket mode: session close settles nothing and bills zero", async () => { + const { fetchFn } = demoFetch([ + () => supplierSse([{ type: "token", value: "Hi" }, doneFrame({ role: "assistant", content: "Hi" }, "stop")]), + ]); + const deps = makeDeps(fetchFn, { chain: fundedChain, chatSettleMode: "ticket" }); + const { rawKey, keyRow } = setupDemoKey(deps); + const ctx = deps.registry.getContext(keyRow); + const endChat = vi.fn(async (opts: { escrowRef: unknown }) => ({ + settleMode: "ticket" as const, + escrowRef: opts.escrowRef, + })); + (ctx.sdk as { endChat: unknown }).endChat = endChat; + const app = createApp(deps); + + const res1 = await request(app) + .post("/openai/v1/chat/completions") + .set("authorization", `Bearer ${rawKey}`) + .send({ model: "kimi", messages: [{ role: "user", content: "hi" }] }); + expect(res1.status).toBe(200); + + const session = deps.store.listOpenSessionsByKey(keyRow.id)[0]; + const close = await request(app) + .post(`/openai/v1/chat/sessions/${session.id}/close`) + .set("authorization", `Bearer ${rawKey}`) + .send({}); + expect(close.status).toBe(200); + expect(close.body.settle_mode).toBe("ticket"); + expect(close.body.accepted_ref).toBeUndefined(); + expect(endChat).toHaveBeenCalledTimes(1); + + const usage = deps.store.listUsage(keyRow.id, 10); + expect(usage.some((u) => u.kind === "chat_session" && u.cost_lovelace === "0")).toBe(true); + }); + + it("ticket mode: janitor-closed demo sessions bill zero", async () => { + const { fetchFn, endCalls } = demoFetch([ + () => supplierSse([{ type: "token", value: "Hi" }, doneFrame({ role: "assistant", content: "Hi" }, "stop")]), + ]); + const deps = makeDeps(fetchFn, { chain: fundedChain, chatSettleMode: "ticket" }); + const { rawKey, keyRow } = setupDemoKey(deps); + const app = createApp(deps); + + const res = await request(app) + .post("/openai/v1/chat/completions") + .set("authorization", `Bearer ${rawKey}`) + .send({ model: "kimi", messages: [{ role: "user", content: "hi" }] }); + expect(res.status).toBe(200); + + await sweepIdleDemoSessions(deps, Date.now() + deps.config.demoSessionIdleMs + 1); + expect(endCalls.length).toBeGreaterThan(0); + const usage = deps.store.listUsage(keyRow.id, 10); + expect(usage.some((u) => u.kind === "chat_session" && u.cost_lovelace === "0")).toBe(true); + }); + it("janitor closes idle demo sessions so the next request opens fresh", async () => { const { fetchFn, messageCalls, endCalls } = demoFetch([ () => supplierSse([{ type: "token", value: "Hi" }, doneFrame({ role: "assistant", content: "Hi" }, "stop")]), diff --git a/gateway/test/sweeper.test.ts b/gateway/test/sweeper.test.ts new file mode 100644 index 0000000..f47dcf9 --- /dev/null +++ b/gateway/test/sweeper.test.ts @@ -0,0 +1,186 @@ +/** + * sweeper.test.ts — reclaim/close decoupling + settled-refs memory. + * + * Ticket mode: reclaiming a live session's Open escrow must NOT close the + * session row or drop its transcript mirror (the conversation continues on a + * spent ticket). Full mode keeps close-on-reclaim. Claimed rows close even + * under ticket config (they can only come from full-mode suppliers). The + * settled-refs memory prevents re-attempting refs already settled, and the + * deliver_by gate skips reclaims the validator would reject. + */ + +import { describe, it, expect, vi } from "vitest"; +import { tmpdir } from "os"; +import { join } from "path"; +import { randomBytes, randomUUID } from "crypto"; +import { GatewayStore } from "../src/db/store.js"; +import { SdkRegistry } from "../src/sdk/registry.js"; +import { genPrivKeyHex, deriveWalletKey } from "../src/wallet.js"; +import { seal } from "../src/crypto/seal.js"; +import { hashApiKey } from "../src/middleware/apiKeyAuth.js"; +import { runSweepOnce } from "../src/sweeper.js"; +import { transcripts } from "../src/openai/transcripts.js"; +import type { GatewayConfig } from "../src/config.js"; +import type { GatewayDeps } from "../src/deps.js"; +import type { EscrowRow } from "../src/onchain/settle.js"; + +const MASTER = "ab".repeat(32); + +function makeSweepDeps(opts: { chatSettleMode: "full" | "ticket"; escrows: EscrowRow[] }) { + const dbDir = join(tmpdir(), `gw-sweep-${randomUUID()}`); + const store = new GatewayStore(dbDir); + const config: GatewayConfig = { + masterKeyHex: MASTER, + indexerUrl: "http://ix", + ogmiosUrl: "http://og", + networkId: 0, + liveChain: true, + port: 0, + dbDir, + signupRate: { max: 1000, windowMs: 60_000 }, + keyRate: { max: 1000, windowMs: 60_000 }, + demoIpRate: { max: 1000, windowMs: 60_000 }, + sweeperIntervalMs: 60_000, + demoSessionIdleMs: 180_000, + chatSettleMode: opts.chatSettleMode, + walletHealthIntervalMs: 600_000, + sdkRegistryMax: 100, + corsOrigins: [], + }; + const chain = { queryUtxo: async () => null } as unknown as GatewayDeps["chain"]; + const registry = new SdkRegistry({ + chain, + indexerUrl: config.indexerUrl, + networkId: config.networkId, + masterKeyHex: config.masterKeyHex, + max: config.sdkRegistryMax, + }); + const fetchFn = (async (url: unknown) => { + if (String(url).includes("/escrows")) return new Response(JSON.stringify(opts.escrows), { status: 200 }); + return new Response("[]", { status: 200 }); + }) as unknown as typeof globalThis.fetch; + const deps: GatewayDeps = { config, store, chain, registry, fetchFn }; + + const rawKey = `vk_test_${randomBytes(24).toString("hex")}`; + const privHex = genPrivKeyHex(); + const walletKey = deriveWalletKey(privHex, 0); + const sealed = seal(privHex, MASTER); + store.insertKey({ + id: randomUUID(), + key_hash: hashApiKey(rawKey), + key_prefix: rawKey.slice(0, 12), + label: "sweep-test", + wallet_pkh: walletKey.pubKeyHash, + deposit_address: walletKey.address, + enc_priv_nonce: sealed.nonce, + enc_priv_ct: sealed.ct, + enc_priv_tag: sealed.tag, + master_key_version: 1, + created_at: Date.now(), + demo: 1, + }); + const keyRow = store.getKeyByHash(hashApiKey(rawKey))!; + return { deps, keyRow }; +} + +function insertOpenSession(deps: GatewayDeps, keyId: string, escrowRef: string): string { + const id = randomUUID(); + deps.store.insertSession({ + id, + key_id: keyId, + escrow_ref: escrowRef, + session_nonce: "nonce", + supplier_base_url: "http://sup", + supplier_pkh: "s1", + model: "kimi", + price_lovelace: "200000", + state: "open", + opened_at: Date.now() - 40 * 60 * 1000, + }); + return id; +} + +function openRow(refChar: string, opts?: Partial): EscrowRow { + return { + utxo_ref: `${refChar.repeat(64)}#0`, + state: "Open", + posted_at: Date.now() - 40 * 60 * 1000, + deliver_by: Date.now() - 5 * 60 * 1000, + ...opts, + }; +} + +const settleOk = () => ({ + accept: vi.fn(async () => undefined) as never, + reclaim: vi.fn(async () => undefined) as never, +}); + +describe("sweeper — ticket-mode reclaim/close decoupling", () => { + it("reclaims a live session's Open ticket but keeps the session (and mirror) alive", async () => { + const row = openRow("1"); + const { deps, keyRow } = makeSweepDeps({ chatSettleMode: "ticket", escrows: [row] }); + const sessionId = insertOpenSession(deps, keyRow.id, row.utxo_ref); + transcripts.set(sessionId, [{ role: "user", content: "hi" }]); + const settle = settleOk(); + + await runSweepOnce(deps, settle); + + expect(settle.reclaim).toHaveBeenCalledTimes(1); + expect(deps.store.getSession(sessionId)!.state).toBe("open"); // NOT closed + expect(transcripts.get(sessionId)).toBeDefined(); // mirror intact + expect(deps.store.listUsage(keyRow.id, 10)).toHaveLength(0); // nothing billed + + transcripts.delete(sessionId); + }); + + it("full mode: reclaim closes the session row", async () => { + const row = openRow("2"); + const { deps, keyRow } = makeSweepDeps({ chatSettleMode: "full", escrows: [row] }); + const sessionId = insertOpenSession(deps, keyRow.id, row.utxo_ref); + const settle = settleOk(); + + await runSweepOnce(deps, settle); + + expect(settle.reclaim).toHaveBeenCalledTimes(1); + expect(deps.store.getSession(sessionId)!.state).toBe("closed"); + const usage = deps.store.listUsage(keyRow.id, 10); + expect(usage.some((u) => u.status === "reclaimed")).toBe(true); + }); + + it("ticket config still closes sessions behind Claimed rows (full-mode suppliers)", async () => { + const row = openRow("3", { state: "Claimed" }); + const { deps, keyRow } = makeSweepDeps({ chatSettleMode: "ticket", escrows: [row] }); + const sessionId = insertOpenSession(deps, keyRow.id, row.utxo_ref); + const settle = settleOk(); + + await runSweepOnce(deps, settle); + + expect(settle.reclaim).toHaveBeenCalledTimes(1); + expect(deps.store.getSession(sessionId)!.state).toBe("closed"); + }); +}); + +describe("sweeper — reclaim gating + settled-refs memory", () => { + it("skips reclaim while deliver_by has not passed", async () => { + const row = openRow("4", { deliver_by: Date.now() + 10 * 60 * 1000 }); + const { deps, keyRow } = makeSweepDeps({ chatSettleMode: "ticket", escrows: [row] }); + insertOpenSession(deps, keyRow.id, row.utxo_ref); + const settle = settleOk(); + + await runSweepOnce(deps, settle); + + expect(settle.reclaim).not.toHaveBeenCalled(); + }); + + it("does not re-attempt refs it already settled (in-process memory)", async () => { + const row = openRow("5"); + const { deps, keyRow } = makeSweepDeps({ chatSettleMode: "ticket", escrows: [row] }); + insertOpenSession(deps, keyRow.id, row.utxo_ref); + const settle = settleOk(); + + await runSweepOnce(deps, settle); + await runSweepOnce(deps, settle); + + expect(settle.reclaim).toHaveBeenCalledTimes(1); + }); +}); diff --git a/supplier/.env.example b/supplier/.env.example index 41bcb93..1f09711 100644 --- a/supplier/.env.example +++ b/supplier/.env.example @@ -126,3 +126,9 @@ ADVERT_DETAIL_HASH= # `docker compose down && up -d` to apply. # ESCROW_REF_UTXO=#0 # ADVERT_REF_UTXO=#1 + +# ─── Optional: chat-session tuning (CAPABILITY_KIND=chat-session only) ───── +# CHAT_IDLE_TIMEOUT_MS=300000 # auto-end sessions idle this long +# MAX_CHAT_SESSIONS=1 # concurrent session slots (default 1) +# CHAT_SETTLE_MODE=full # "full" (Claim/Submit) or "ticket" (no supplier +# # chain ops; buyer reclaims the Open escrow) diff --git a/supplier/src/chatSession.ts b/supplier/src/chatSession.ts index 51e45c3..442fe4e 100644 --- a/supplier/src/chatSession.ts +++ b/supplier/src/chatSession.ts @@ -8,9 +8,10 @@ * turns that run fully OFF-CHAIN, and is settled at /v1/chat/end (or by the * idle watchdog) when the supplier Submits a receipt over the whole transcript. * - * Single-slot: the supplier serves one paid chat at a time (SupplierState - * mutex), so at most one record is "active". Ended records are retained so - * /v1/chat/end is idempotent (returns the same receipt). + * Concurrency: up to MAX_CHAT_SESSIONS records are "active" at once + * (SupplierState slot admission). Ended records are retained so /v1/chat/end + * is idempotent (returns the same receipt) and so a used ticket-mode escrow + * ref can never open a second session. * * Timer handles (idle + hard-cap) live on the record; server.ts arms them and * endChatSession clears them. @@ -31,7 +32,11 @@ export interface ChatSessionEndResult { export interface ChatSessionRecord { escrowRef: string; - claimedRef: OutputReference; + /** Absent in ticket mode (the escrow is never Claimed). */ + claimedRef?: OutputReference; + /** "full" = Claim/Submit lifecycle; "ticket" = no chain ops after the + * buyer's escrow post. */ + settleMode: "full" | "ticket"; advert: AdvertDatum; escrowDatum: EscrowDatum; /** Full ordered conversation (user + assistant turns). Hashed at End. */ @@ -49,7 +54,8 @@ export interface ChatSessionRecord { export interface CreateChatSessionParams { escrowRef: string; - claimedRef: OutputReference; + claimedRef?: OutputReference; + settleMode?: "full" | "ticket"; advert: AdvertDatum; escrowDatum: EscrowDatum; } @@ -62,6 +68,7 @@ export class ChatSessionStore { const record: ChatSessionRecord = { escrowRef: params.escrowRef, claimedRef: params.claimedRef, + settleMode: params.settleMode ?? "full", advert: params.advert, escrowDatum: params.escrowDatum, transcript: [], diff --git a/supplier/src/chatSessionRunner.ts b/supplier/src/chatSessionRunner.ts index 5b351be..1471a9d 100644 --- a/supplier/src/chatSessionRunner.ts +++ b/supplier/src/chatSessionRunner.ts @@ -1,20 +1,22 @@ /** - * supplier/src/chatSessionRunner.ts — settles an open chat session. + * supplier/src/chatSessionRunner.ts — ends an open chat session. * - * endChatSession is the chat-session analogue of runChatJob's receipt+Submit - * tail (jobRunner.ts steps 3-7). It is invoked from BOTH /v1/chat/end (user - * clicked "End chat") and the idle/hard-cap watchdog (abandoned session). It: - * 1. atomically flips the session "active" → "ending" (idempotent re-entry) - * 2. clears the idle + hard-cap timers - * 3. builds + signs a receipt over the FULL transcript - * (response_hash = sha256(canonical(transcript)); the buyer recomputes - * this from its local transcript mirror to verify) - * 4. buildSubmitTx against the Claimed UTxO → awaitTx (60s) - * 5. stores the terminal result and ALWAYS releases the supplier lock + * endChatSession is invoked from BOTH /v1/chat/end (user clicked "End chat") + * and the idle/hard-cap watchdog (abandoned session). Behavior branches on + * the record's settleMode: + * + * full — chat-session analogue of runChatJob's receipt+Submit tail: + * build + sign a receipt over the FULL transcript + * (response_hash = sha256(canonical(transcript))), then + * buildSubmitTx against the Claimed UTxO → awaitTx (60s), inside + * the wallet mutex (never race another wallet spend). + * ticket — no chain work at all: mark ended and free the slot. The escrow + * was never Claimed; the buyer's sweeper Reclaims it after + * deliver_by. * * Never throws. On any failure the session is marked "ended" with endFailure - * and the lock is released (the escrow then lingers Claimed until the buyer - * Reclaims after deliver_by — same fate as a failed one-off job). + * and the slot is released (a full-mode escrow then lingers Claimed until the + * buyer Reclaims after deliver_by — same fate as a failed one-off job). */ import { canonicalize } from "@marketplace/shared/cbor"; @@ -41,7 +43,7 @@ export interface EndChatSessionParams { } /** - * Settle a chat session. Returns the (now terminal) record, or null if no + * End a chat session. Returns the (now terminal) record, or null if no * session exists for the ref. Idempotent: a second call returns the same * record without re-submitting. */ @@ -59,6 +61,21 @@ export async function endChatSession( deps.sessions.clearTimers(record); try { + if (record.settleMode === "ticket") { + // No chain work: the un-Claimed escrow is the buyer's to reclaim. + record.status = "ended"; + return record; + } + const claimedRef = record.claimedRef; + if (!claimedRef) { + // Full-mode record without a Claim ref should be impossible; end + // defensively rather than throw (this path must never leak the slot). + console.warn(`[chat_end_failed] escrow=${escrowRef} trigger=${trigger ?? "end"} reason=missing_claimed_ref`); + record.endFailure = { reason: "missing_claimed_ref", message: "full-mode session has no claimedRef" }; + record.status = "ended"; + return record; + } + // ── Receipt over the full transcript ────────────────────────────────── const transcriptHash = sha256Hex(canonicalize(record.transcript)); const receipt = buildReceipt({ @@ -74,27 +91,35 @@ export async function endChatSession( const signed = signReceipt(receipt, deps.supplierKey.privateKeyHex); const resultHash = receiptResultHash(signed); - // ── Submit (Claimed → Submitted) ────────────────────────────────────── - let buildResult: { txCborHex: string; expectedTxHash: string }; - try { - buildResult = await buildSubmitTx({ - chain: deps.chain, - supplierKey: deps.supplierKey, - escrowRef: record.claimedRef, - receiptHash: resultHash, - }); - } catch (err) { - const message = err instanceof Error ? err.message : String(err); + // ── Submit (Claimed → Submitted): one wallet critical section ───────── + const submitOutcome = await deps.state.walletMutex.run(async () => { + let built: { txCborHex: string; expectedTxHash: string }; + try { + built = await buildSubmitTx({ + chain: deps.chain, + supplierKey: deps.supplierKey, + escrowRef: claimedRef, + receiptHash: resultHash, + }); + } catch (err) { + return { kind: "build_failed" as const, err }; + } + try { + await deps.chain.awaitTx(built.expectedTxHash, 60_000); + } catch (err) { + return { kind: "await_failed" as const, err }; + } + return { kind: "ok" as const, built }; + }); + if (submitOutcome.kind === "build_failed") { + const message = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err); console.warn(`[chat_end_failed] escrow=${escrowRef} trigger=${trigger ?? "end"} reason=submit_failed msg=${message}`); record.endFailure = { reason: "submit_failed", message: `Submit tx failed: ${message}` }; record.status = "ended"; return record; } - - try { - await deps.chain.awaitTx(buildResult.expectedTxHash, 60_000); - } catch (err) { - const message = err instanceof Error ? err.message : String(err); + if (submitOutcome.kind === "await_failed") { + const message = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err); console.warn(`[chat_end_failed] escrow=${escrowRef} trigger=${trigger ?? "end"} reason=submit_timeout msg=${message}`); record.endFailure = { reason: "submit_timeout", message: `Submit awaitTx failed: ${message}` }; record.status = "ended"; @@ -105,14 +130,14 @@ export async function endChatSession( receipt: signed.receipt as unknown as Record, receipt_signature: signed.signature, // The Submit tx output (index 0) is the new Submitted UTxO the buyer Accepts. - submitted_ref: `${buildResult.expectedTxHash}#0`, + submitted_ref: `${submitOutcome.built.expectedTxHash}#0`, }; record.status = "ended"; return record; } finally { - // Always release the single-slot lock so the next chat can start. + // Always release this session's slot so the next chat can start. try { - deps.state.release(); + deps.state.release(record.escrowRef); } catch { /* never let lock-release escape */ } diff --git a/supplier/src/config.ts b/supplier/src/config.ts index 9436147..2c6c5a4 100644 --- a/supplier/src/config.ts +++ b/supplier/src/config.ts @@ -66,6 +66,13 @@ export type CapabilityKind = "chat" | "tts" | "chat-session"; /** LLM backend selected for the chat capability. */ export type LlmBackend = "ollama" | "openai"; +/** Chain settlement mode for chat sessions. + * "full" — Claim at start, Submit receipt at end (buyer Accepts). Default. + * "ticket" — the buyer's escrow post is the only chain op; the supplier + * verifies it as an entry ticket but never Claims/Submits, and + * the buyer's sweeper Reclaims the Open escrow after deliver_by. */ +export type ChatSettleMode = "full" | "ticket"; + export interface SupplierConfig { supplierPrivKeyHex: string; ogmiosUrl: string; @@ -109,6 +116,13 @@ export interface SupplierConfig { * 300_000 (5 min). */ chatIdleTimeoutMs: number; + /** + * Concurrent chat-session slots (capabilityKind="chat-session" only). + * Default 1 = legacy single-slot behavior. + */ + maxChatSessions: number; + /** Chain settlement mode for chat sessions. Default "full". */ + chatSettleMode: ChatSettleMode; /** * When true, the supplier boots a LiveOgmiosProvider (real submitTx/awaitTx). * Default false → ReadOnlyOgmiosProvider (safe; no chain writes). @@ -278,6 +292,32 @@ export function loadConfig(env: Record): SupplierCon chatIdleTimeoutMs = Number(chatIdleStr); } + // MAX_CHAT_SESSIONS — concurrent chat-session slots (default 1 = legacy + // single-slot). Only meaningful for capabilityKind="chat-session". + const maxSessionsStr = env.MAX_CHAT_SESSIONS; + let maxChatSessions = 1; + if (maxSessionsStr !== undefined && maxSessionsStr !== "") { + if (!POS_INT_RE.test(maxSessionsStr)) { + throw new Error("loadConfig: MAX_CHAT_SESSIONS must be a positive integer"); + } + maxChatSessions = Number(maxSessionsStr); + } + if (maxChatSessions > 1 && capabilityKind !== "chat-session") { + throw new Error('loadConfig: MAX_CHAT_SESSIONS > 1 requires CAPABILITY_KIND="chat-session"'); + } + + // CHAT_SETTLE_MODE — "full" (default; Claim at start, Submit receipt at end) + // or "ticket" (the buyer's escrow post is the only chain op; the supplier + // never Claims/Submits and the buyer reclaims the escrow after deliver_by). + const settleModeStr = env.CHAT_SETTLE_MODE ?? "full"; + if (settleModeStr !== "full" && settleModeStr !== "ticket") { + throw new Error('loadConfig: CHAT_SETTLE_MODE must be "full" or "ticket"'); + } + if (settleModeStr === "ticket" && capabilityKind !== "chat-session") { + throw new Error('loadConfig: CHAT_SETTLE_MODE="ticket" requires CAPABILITY_KIND="chat-session"'); + } + const chatSettleMode: ChatSettleMode = settleModeStr; + const piperTimeoutStr = env.PIPER_TIMEOUT_MS; let piperTimeoutMs = 120_000; if (piperTimeoutStr !== undefined && piperTimeoutStr !== "") { @@ -320,6 +360,8 @@ export function loadConfig(env: Record): SupplierCon capabilityKind, llmBackend, chatIdleTimeoutMs, + maxChatSessions, + chatSettleMode, liveChain, walletHealthIntervalMs, }; diff --git a/supplier/src/index.ts b/supplier/src/index.ts index 2f9b19e..4404f71 100644 --- a/supplier/src/index.ts +++ b/supplier/src/index.ts @@ -42,7 +42,8 @@ async function main(): Promise { // M1-F-2: provider selection is driven by cfg.liveChain (LIVE_CHAIN=1). // ReadOnly is the safe default; live mode is explicit opt-in. const chain = buildChainProvider(cfg); - const state = new SupplierState(); + // chat-session kind serves N concurrent sessions; one-shot kinds stay 1. + const state = new SupplierState(cfg.capabilityKind === "chat-session" ? cfg.maxChatSessions : 1); const jobs = new JobStore(); const deps: SupplierDeps = { chain, state, config: cfg, supplierKey, jobs }; diff --git a/supplier/src/jobRunner.ts b/supplier/src/jobRunner.ts index 677fa62..746aec1 100644 --- a/supplier/src/jobRunner.ts +++ b/supplier/src/jobRunner.ts @@ -143,17 +143,28 @@ export async function runChatJob(params: RunChatJobParams): Promise { const signed = signReceipt(receipt, deps.supplierKey.privateKeyHex); const resultHash = receiptResultHash(signed); - // ── 4. Build + submit Submit tx ───────────────────────────────────── - let buildResult: { txCborHex: string; expectedTxHash: string }; - try { - buildResult = await buildSubmitTx({ - chain: deps.chain, - supplierKey: deps.supplierKey, - escrowRef: claimedRef, - receiptHash: resultHash, - }); - } catch (err) { - const message = err instanceof Error ? err.message : String(err); + // ── 4+5. Submit tx: build + confirm as ONE wallet critical section ── + const submitOutcome = await deps.state.walletMutex.run(async () => { + let built: { txCborHex: string; expectedTxHash: string }; + try { + built = await buildSubmitTx({ + chain: deps.chain, + supplierKey: deps.supplierKey, + escrowRef: claimedRef, + receiptHash: resultHash, + }); + } catch (err) { + return { kind: "build_failed" as const, err }; + } + try { + await deps.chain.awaitTx(built.expectedTxHash, 60_000); + } catch (err) { + return { kind: "await_failed" as const, err }; + } + return { kind: "ok" as const }; + }); + if (submitOutcome.kind === "build_failed") { + const message = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err); console.warn( `[job_failed] jobId=${jobId} reason=submit_failed httpStatus=502 msg=${message}`, ); @@ -164,12 +175,8 @@ export async function runChatJob(params: RunChatJobParams): Promise { }); return; } - - // ── 5. Submit confirmation (awaitTx, 60s budget) ─────────────────── - try { - await deps.chain.awaitTx(buildResult.expectedTxHash, 60_000); - } catch (err) { - const msg = err instanceof Error ? err.message : String(err); + if (submitOutcome.kind === "await_failed") { + const msg = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err); console.warn( `[job_failed] jobId=${jobId} reason=submit_timeout httpStatus=502 msg=${msg}`, ); @@ -201,9 +208,9 @@ export async function runChatJob(params: RunChatJobParams): Promise { }; deps.jobs.complete(jobId, payload); } finally { - // ── 7. Always release the lock ────────────────────────────────────── + // ── 7. Always release this job's slot ─────────────────────────────── try { - deps.state.release(); + deps.state.release(escrowRef); } catch { // never let a lock-release error escape the runner } @@ -292,17 +299,28 @@ export async function runTtsJob(params: RunTtsJobParams): Promise { const signed = signReceipt(receipt, deps.supplierKey.privateKeyHex); const resultHash = receiptResultHash(signed); - // ── 4. Build + submit Submit tx ─────────────────────────────────── - let buildResult: { txCborHex: string; expectedTxHash: string }; - try { - buildResult = await buildSubmitTx({ - chain: deps.chain, - supplierKey: deps.supplierKey, - escrowRef: claimedRef, - receiptHash: resultHash, - }); - } catch (err) { - const message = err instanceof Error ? err.message : String(err); + // ── 4+5. Submit tx: build + confirm as ONE wallet critical section ── + const submitOutcome = await deps.state.walletMutex.run(async () => { + let built: { txCborHex: string; expectedTxHash: string }; + try { + built = await buildSubmitTx({ + chain: deps.chain, + supplierKey: deps.supplierKey, + escrowRef: claimedRef, + receiptHash: resultHash, + }); + } catch (err) { + return { kind: "build_failed" as const, err }; + } + try { + await deps.chain.awaitTx(built.expectedTxHash, 60_000); + } catch (err) { + return { kind: "await_failed" as const, err }; + } + return { kind: "ok" as const }; + }); + if (submitOutcome.kind === "build_failed") { + const message = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err); console.warn( `[job_failed] jobId=${jobId} reason=submit_failed httpStatus=502 msg=${message}`, ); @@ -313,12 +331,8 @@ export async function runTtsJob(params: RunTtsJobParams): Promise { }); return; } - - // ── 5. Submit confirmation ──────────────────────────────────────── - try { - await deps.chain.awaitTx(buildResult.expectedTxHash, 60_000); - } catch (err) { - const msg = err instanceof Error ? err.message : String(err); + if (submitOutcome.kind === "await_failed") { + const msg = submitOutcome.err instanceof Error ? submitOutcome.err.message : String(submitOutcome.err); console.warn( `[job_failed] jobId=${jobId} reason=submit_timeout httpStatus=502 msg=${msg}`, ); @@ -344,7 +358,7 @@ export async function runTtsJob(params: RunTtsJobParams): Promise { deps.jobs.complete(jobId, payload); } finally { try { - deps.state.release(); + deps.state.release(escrowRef); } catch { /* never let lock-release escape */ } diff --git a/supplier/src/server.ts b/supplier/src/server.ts index ae16d75..e17379a 100644 --- a/supplier/src/server.ts +++ b/supplier/src/server.ts @@ -147,6 +147,8 @@ function makeStatusHandler(deps: ResolvedDeps) { const payload: Record = { status: snap.status, last_seen: snap.lastSeenIso, + active_sessions: snap.activeSessions, + max_sessions: snap.maxSessions, }; if (snap.status === "working" && snap.currentEscrowRef) { payload.current_escrow_ref = snap.currentEscrowRef; @@ -190,6 +192,9 @@ async function fetchActiveAdvert(deps: ResolvedDeps): Promise { function makeChatHandler(deps: ResolvedDeps) { return async (req: Request, res: Response, next: NextFunction) => { + // Slot held by THIS request (null until acquired / after hand-off to the + // job runner) — the catch-all must only release its own acquisition. + let acquiredRef: string | null = null; try { // ── 1. Header validation ────────────────────────────────────────── const headerVal = req.header("X-Escrow-Ref"); @@ -303,23 +308,37 @@ function makeChatHandler(deps: ResolvedDeps) { `now ${nowMs} >= deliver_by ${escrowDatum.deliver_by}`); } - // ── 8. Acquire single-slot lock ────────────────────────────────── + // ── 8. Acquire session slot ────────────────────────────────────── if (!deps.state.tryAcquire(escrowRefStr)) { return jsonError(res, 409, "supplier_busy", "supplier is already working another job"); } - // From here on, every error path MUST release the lock until the - // runner takes ownership of the lock. + acquiredRef = escrowRefStr; + // From here on, every error path MUST release the slot until the + // runner takes ownership of it. - // ── 9. Submit Claim tx ─────────────────────────────────────────── - let claimResult; - try { - claimResult = await buildClaimTx({ - chain: deps.chain, - supplierKey: deps.supplierKey, - escrowRef, - }); - } catch (err) { - deps.state.release(); + // ── 9+10. Claim tx: build + confirm as ONE wallet critical section ─ + // (splitting them would let another wallet spend coin-select mid-flight) + const claimOutcome = await deps.state.walletMutex.run(async () => { + let built; + try { + built = await buildClaimTx({ + chain: deps.chain, + supplierKey: deps.supplierKey, + escrowRef, + }); + } catch (err) { + return { kind: "build_failed" as const, err }; + } + try { + await deps.chain.awaitTx(built.expectedTxHash, 60_000); + } catch (err) { + return { kind: "await_failed" as const, err }; + } + return { kind: "ok" as const, built }; + }); + if (claimOutcome.kind === "build_failed") { + deps.state.release(escrowRefStr); + acquiredRef = null; // Backstop the periodic wallet-health ticker: most Claim build failures // we see in practice are wallet-shape problems (fragmentation, dust). // Fire-and-forget a debounced consolidate so the NEXT buyer retry @@ -330,17 +349,15 @@ function makeChatHandler(deps: ResolvedDeps) { supplierKey: deps.supplierKey, }); return jsonError(res, 503, "chain_submit_failed", - `Claim tx submit failed: ${(err as Error).message}`); + `Claim tx submit failed: ${(claimOutcome.err as Error).message}`); } - - // ── 10. Await Claim confirmation ───────────────────────────────── - try { - await deps.chain.awaitTx(claimResult.expectedTxHash, 60_000); - } catch (err) { - deps.state.release(); + if (claimOutcome.kind === "await_failed") { + deps.state.release(escrowRefStr); + acquiredRef = null; return jsonError(res, 504, "claim_timeout", - `Claim awaitTx failed: ${(err as Error).message}`); + `Claim awaitTx failed: ${(claimOutcome.err as Error).message}`); } + const claimResult = claimOutcome.built; // ── 11. Continuing-output ref + create job + fire-and-forget ───── const claimedRef: OutputReference = { @@ -349,7 +366,8 @@ function makeChatHandler(deps: ResolvedDeps) { }; const jobId = deps.jobs.create(escrowRefStr); - // Lock release now happens inside runChatJob's finally. + // Slot release now happens inside runChatJob's finally. + acquiredRef = null; void runChatJob({ deps: { chain: deps.chain, @@ -373,7 +391,9 @@ function makeChatHandler(deps: ResolvedDeps) { escrow_ref: escrowRefStr, }); } catch (err) { - try { deps.state.release(); } catch { /* ignore */ } + if (acquiredRef) { + try { deps.state.release(acquiredRef); } catch { /* ignore */ } + } next(err); return; } @@ -447,6 +467,8 @@ const ALLOWED_TTS_FORMATS = new Set(["mp3", "wav", "opus", "aac", "flac"]); function makeTtsHandler(deps: ResolvedDeps) { return async (req: Request, res: Response, next: NextFunction) => { + // Slot held by THIS request (null until acquired / after hand-off). + let acquiredRef: string | null = null; try { // ── 1. Header ─────────────────────────────────────────────────── const headerVal = req.header("X-Escrow-Ref"); @@ -550,39 +572,53 @@ function makeTtsHandler(deps: ResolvedDeps) { `now ${nowMs} >= deliver_by ${escrowDatum.deliver_by}`); } - // ── 8. Lock ───────────────────────────────────────────────────── + // ── 8. Acquire session slot ───────────────────────────────────── if (!deps.state.tryAcquire(escrowRefStr)) { return jsonError(res, 409, "supplier_busy", "supplier is already working another job"); } + acquiredRef = escrowRefStr; - // ── 9. Claim tx ───────────────────────────────────────────────── - let claimResult; - try { - claimResult = await buildClaimTx({ - chain: deps.chain, - supplierKey: deps.supplierKey, - escrowRef, - }); - } catch (err) { - deps.state.release(); + // ── 9. Claim tx: build + confirm as ONE wallet critical section ── + const claimOutcome = await deps.state.walletMutex.run(async () => { + let built; + try { + built = await buildClaimTx({ + chain: deps.chain, + supplierKey: deps.supplierKey, + escrowRef, + }); + } catch (err) { + return { kind: "build_failed" as const, err }; + } + try { + await deps.chain.awaitTx(built.expectedTxHash, 60_000); + } catch (err) { + return { kind: "await_failed" as const, err }; + } + return { kind: "ok" as const, built }; + }); + if (claimOutcome.kind === "build_failed") { + deps.state.release(escrowRefStr); + acquiredRef = null; return jsonError(res, 503, "chain_submit_failed", - `Claim tx submit failed: ${(err as Error).message}`); + `Claim tx submit failed: ${(claimOutcome.err as Error).message}`); } - try { - await deps.chain.awaitTx(claimResult.expectedTxHash, 60_000); - } catch (err) { - deps.state.release(); + if (claimOutcome.kind === "await_failed") { + deps.state.release(escrowRefStr); + acquiredRef = null; return jsonError(res, 504, "claim_timeout", - `Claim awaitTx failed: ${(err as Error).message}`); + `Claim awaitTx failed: ${(claimOutcome.err as Error).message}`); } // ── 10. Spawn runner + 202 ────────────────────────────────────── const claimedRef: OutputReference = { - txHash: claimResult.expectedTxHash, + txHash: claimOutcome.built.expectedTxHash, index: 0, }; const jobId = deps.jobs.create(escrowRefStr); + // Slot release now happens inside runTtsJob's finally. + acquiredRef = null; void runTtsJob({ deps: { chain: deps.chain, @@ -605,7 +641,9 @@ function makeTtsHandler(deps: ResolvedDeps) { escrow_ref: escrowRefStr, }); } catch (err) { - try { deps.state.release(); } catch { /* ignore */ } + if (acquiredRef) { + try { deps.state.release(acquiredRef); } catch { /* ignore */ } + } next(err); return; } @@ -705,6 +743,9 @@ function makeChatSessionHandlers(deps: ResolvedDeps) { } const start = async (req: Request, res: Response, next: NextFunction) => { + // Slot held by THIS request (null until acquired / after the session + // record takes ownership) — the catch-all only releases its own. + let acquiredRef: string | null = null; try { // ── 1. Header ─────────────────────────────────────────────────── const headerVal = req.header("X-Escrow-Ref"); @@ -717,6 +758,15 @@ function makeChatSessionHandlers(deps: ResolvedDeps) { } const escrowRefStr = `${escrowRef.txHash}#${escrowRef.index}`; + // ── 1.5 Duplicate-session guard ─────────────────────────────────── + // Ended records are retained, so this also blocks replaying a used + // ticket-mode escrow (which stays Open on-chain — the state check below + // wouldn't catch it). In full mode the on-chain Claim already prevents + // reuse; this check just fails faster. + if (deps.chatSessions.get(escrowRefStr)) { + return jsonError(res, 409, "session_exists", "a chat session already exists for this escrow ref"); + } + // ── 2. Body (session_nonce; no messages) ───────────────────────── const body = (req.body ?? {}) as ChatStartBody; const sessionNonce = typeof body.session_nonce === "string" ? body.session_nonce : ""; @@ -780,36 +830,68 @@ function makeChatSessionHandlers(deps: ResolvedDeps) { return jsonError(res, 408, "past_deliver_by", `now ${nowMs} >= deliver_by ${escrowDatum.deliver_by}`); } - // ── 8. Single-slot lock ────────────────────────────────────────── + // ── 8. Acquire session slot ────────────────────────────────────── if (!deps.state.tryAcquire(escrowRefStr)) { return jsonError(res, 409, "supplier_busy", "supplier is already working another chat"); } + acquiredRef = escrowRefStr; - // ── 9. Claim (Open → Claimed) — hold the lock on success ───────── - let claimResult; - try { - claimResult = await buildClaimTx({ chain: deps.chain, supplierKey: deps.supplierKey, escrowRef }); - } catch (err) { - deps.state.release(); - triggerOnFailureConsolidate({ chain: deps.chain, state: deps.state, supplierKey: deps.supplierKey }); - return jsonError(res, 503, "chain_submit_failed", `Claim tx submit failed: ${(err as Error).message}`); + if (deps.config.chatSettleMode === "ticket") { + // ── 9-ticket. No Claim: the verified Open escrow IS the entry + // ticket. No hard-cap timer either — there is no Submit deadline, so + // the session may outlive deliver_by (the buyer reclaims the escrow + // independently). The idle timer is the only lifetime bound. + const record = deps.chatSessions.create({ + escrowRef: escrowRefStr, + settleMode: "ticket", + advert, + escrowDatum, + }); + acquiredRef = null; // record owns the slot; endChatSession releases it + armIdleTimer(record); + return res.status(200).json({ status: "ticket", escrow_ref: escrowRefStr, settle_mode: "ticket" }); } - try { - await deps.chain.awaitTx(claimResult.expectedTxHash, 60_000); - } catch (err) { - deps.state.release(); - return jsonError(res, 504, "claim_timeout", `Claim awaitTx failed: ${(err as Error).message}`); + + // ── 9-full. Claim (Open → Claimed): build + confirm as ONE wallet + // critical section — hold the slot on success through Submit at end. + const claimOutcome = await deps.state.walletMutex.run(async () => { + let built; + try { + built = await buildClaimTx({ chain: deps.chain, supplierKey: deps.supplierKey, escrowRef }); + } catch (err) { + return { kind: "build_failed" as const, err }; + } + try { + await deps.chain.awaitTx(built.expectedTxHash, 60_000); + } catch (err) { + return { kind: "await_failed" as const, err }; + } + return { kind: "ok" as const, built }; + }); + if (claimOutcome.kind === "build_failed") { + deps.state.release(escrowRefStr); + acquiredRef = null; + triggerOnFailureConsolidate({ chain: deps.chain, state: deps.state, supplierKey: deps.supplierKey }); + return jsonError(res, 503, "chain_submit_failed", `Claim tx submit failed: ${(claimOutcome.err as Error).message}`); + } + if (claimOutcome.kind === "await_failed") { + deps.state.release(escrowRefStr); + acquiredRef = null; + return jsonError(res, 504, "claim_timeout", `Claim awaitTx failed: ${(claimOutcome.err as Error).message}`); } // ── 10. Create session + arm watchdog ──────────────────────────── - const claimedRef: OutputReference = { txHash: claimResult.expectedTxHash, index: 0 }; - const record = deps.chatSessions.create({ escrowRef: escrowRefStr, claimedRef, advert, escrowDatum }); + const claimedRef: OutputReference = { txHash: claimOutcome.built.expectedTxHash, index: 0 }; + const record = deps.chatSessions.create({ escrowRef: escrowRefStr, claimedRef, settleMode: "full", advert, escrowDatum }); + acquiredRef = null; // record owns the slot; endChatSession releases it armIdleTimer(record); armHardCapTimer(record, nowMs); return res.status(200).json({ status: "claimed", escrow_ref: escrowRefStr }); } catch (err) { - try { deps.state.release(); } catch { /* ignore */ } + if (acquiredRef) { + try { deps.state.release(acquiredRef); } catch { /* ignore */ } + } next(err); return; } @@ -954,6 +1036,11 @@ function makeChatSessionHandlers(deps: ResolvedDeps) { if (!record) { return jsonError(res, 404, "chat_session_not_found", `no chat session for ${escrowRefStr}`); } + if (record.settleMode === "ticket") { + // No receipt/Submit in ticket mode — the buyer reclaims the escrow. + // Idempotent: a re-end returns the same shape. + return res.status(200).json({ status: "ended", escrow_ref: escrowRefStr, settle_mode: "ticket" }); + } if (record.endFailure) { return jsonError(res, 502, record.endFailure.reason, record.endFailure.message); } diff --git a/supplier/src/state.ts b/supplier/src/state.ts index 5d7086b..85ac912 100644 --- a/supplier/src/state.ts +++ b/supplier/src/state.ts @@ -1,35 +1,68 @@ /** - * supplier/src/state.ts — In-process single-slot concurrency lock. + * supplier/src/state.ts — In-process session-slot admission + wallet mutex. * - * SupplierState tracks whether the node is free, working, or offline. + * SupplierState admits up to `capacity` concurrent sessions (capacity 1 for + * one-shot chat/tts kinds; MAX_CHAT_SESSIONS for chat-session kind). Status is + * free-until-full: "working" only when every slot is taken, so the indexer + * status poller and the buyer SDK's pre-check keep their existing semantics. * Cross-process coordination is M1-D; this is purely in-memory. * - * Methods: - * tryAcquire(escrowRef): boolean — atomic test-and-set; returns false unless free - * release() — transition working → free (no-op from free/offline) - * markOffline() — transition any state → offline; idempotent - * snapshot() — current status snapshot (defensive copy) + * walletMutex serializes every wallet-spending chain op (Claim/Submit/ + * consolidate): the wallet holds one working UTxO + one collateral UTxO, so + * concurrent txs would double-spend regardless of how many session slots + * exist. Hold it across build+awaitTx as one critical section. * * Invariants: - * - currentEscrowRef is set IFF status === "working" - * - markOffline() clears currentEscrowRef - * - tryAcquire never overwrites a held lock + * - a ref can be held at most once (duplicate tryAcquire returns false) + * - release(ref) frees exactly that ref's slot + * - markOffline() clears all slots and refuses new acquires + * - snapshot.currentEscrowRef is present IFF status === "working" (legacy + * consumers; holds the oldest active ref) * - lastSeenIso updates on every state-changing operation */ export type SupplierStatus = "free" | "working" | "offline"; +/** Tiny promise-chain mutex: run(fn) executes fns strictly one at a time. */ +export class Mutex { + private tail: Promise = Promise.resolve(); + + run(fn: () => Promise): Promise { + const result = this.tail.then(fn, fn); + // Keep the chain alive even if fn rejects; swallow here so the next + // run() isn't poisoned by a prior rejection. + this.tail = result.then( + () => undefined, + () => undefined, + ); + return result; + } +} + export interface SupplierSnapshot { status: SupplierStatus; + /** Oldest active ref; present ONLY when status === "working" (back-compat). */ currentEscrowRef?: string; + activeEscrowRefs: string[]; + activeSessions: number; + maxSessions: number; lastSeenIso: string; } export class SupplierState { - private status: SupplierStatus = "free"; - private currentEscrowRef: string | undefined = undefined; + /** Serializes ALL wallet-spending chain ops (Claim/Submit/consolidate). */ + readonly walletMutex = new Mutex(); + + private readonly active = new Set(); // insertion-ordered + private offline = false; private lastSeenIso: string = new Date().toISOString(); + constructor(readonly capacity: number = 1) { + if (!Number.isInteger(capacity) || capacity < 1) { + throw new Error(`SupplierState: capacity must be a positive integer, got ${capacity}`); + } + } + private touch(): void { // ISO 8601 with millisecond resolution. Coarse clocks may produce equal // values in tight succession — tests use >= rather than strict greater. @@ -37,40 +70,45 @@ export class SupplierState { } tryAcquire(escrowRef: string): boolean { - if (this.status !== "free") return false; - this.status = "working"; - this.currentEscrowRef = escrowRef; + if (this.offline) return false; + if (this.active.has(escrowRef)) return false; + if (this.active.size >= this.capacity) return false; + this.active.add(escrowRef); this.touch(); return true; } - release(): void { - if (this.status === "working") { - this.status = "free"; - this.currentEscrowRef = undefined; + release(escrowRef: string): void { + if (this.active.delete(escrowRef)) { this.touch(); } } markOffline(): void { - this.status = "offline"; - this.currentEscrowRef = undefined; + this.offline = true; + this.active.clear(); this.touch(); } snapshot(): SupplierSnapshot { - // Omit currentEscrowRef key entirely when not working so - // ('currentEscrowRef' in snap) === false (matches test expectation). - if (this.status === "working" && this.currentEscrowRef !== undefined) { - return { - status: this.status, - currentEscrowRef: this.currentEscrowRef, - lastSeenIso: this.lastSeenIso, - }; - } - return { - status: this.status, + const refs = [...this.active]; + const status: SupplierStatus = this.offline + ? "offline" + : refs.length >= this.capacity + ? "working" + : "free"; + const snap: SupplierSnapshot = { + status, + activeEscrowRefs: refs, + activeSessions: refs.length, + maxSessions: this.capacity, lastSeenIso: this.lastSeenIso, }; + // Omit currentEscrowRef key entirely when not working so + // ('currentEscrowRef' in snap) === false (matches legacy consumers). + if (status === "working" && refs.length > 0) { + snap.currentEscrowRef = refs[0]; + } + return snap; } } diff --git a/supplier/src/walletHealth.ts b/supplier/src/walletHealth.ts index 23c70e0..da8f1cf 100644 --- a/supplier/src/walletHealth.ts +++ b/supplier/src/walletHealth.ts @@ -16,9 +16,10 @@ * multiple consolidates. Returns 503 to the buyer for THIS request; the * next buyer retry (~30-60s later) finds the wallet healthy. * - * State-lock discipline: both layers use state.tryAcquire with a synthetic - * escrowRef so they share the existing single-slot lock with real Claims. - * If a Claim is in flight, the consolidation skips this tick. + * Wallet-lock discipline: consolidation spends EVERY wallet UTxO, so it runs + * inside state.walletMutex — the same mutex that serializes Claims/Submits — + * queueing behind any in-flight chain op instead of skipping. Session slots + * are not consumed: consolidation is a wallet op, not a session. */ import type { ChainProvider } from "@marketplace/shared/chain"; @@ -31,7 +32,6 @@ import { import type { SupplierState } from "./state.js"; -const WALLET_HEALTH_LOCK_REF = "__wallet_consolidate__"; const DEFAULT_COLLATERAL_LOVELACE = 5_000_000n; const DEFAULT_AWAIT_TIMEOUT_MS = 90_000; const DEFAULT_ON_FAILURE_DEBOUNCE_MS = 5 * 60_000; @@ -77,20 +77,20 @@ async function consolidateOnce( opts.log("skip: another consolidate already in flight"); return; } - if (!state.tryAcquire(WALLET_HEALTH_LOCK_REF)) { - opts.log("skip: supplier busy"); - return; - } inFlight.add(state); const consolidate: ConsolidateFn = deps.consolidate ?? runConsolidateWallet; try { - const result = await consolidate({ - chain: deps.chain, - walletKey: deps.supplierKey, - collateralLovelace: opts.collateralLovelace, - awaitTimeoutMs: opts.awaitTimeoutMs, - log: opts.log, - }); + // Queue behind any in-flight Claim/Submit — consolidation reshapes the + // whole wallet and must never race another wallet spend. + const result = await state.walletMutex.run(() => + consolidate({ + chain: deps.chain, + walletKey: deps.supplierKey, + collateralLovelace: opts.collateralLovelace, + awaitTimeoutMs: opts.awaitTimeoutMs, + log: opts.log, + }), + ); if (result.reason === "already-healthy") { opts.log("already-healthy"); } else { @@ -100,7 +100,6 @@ async function consolidateOnce( opts.log(`consolidate failed: ${(err as Error).message}`); } finally { inFlight.delete(state); - state.release(); } } diff --git a/tests/fixtures/supplier-side/sample-config.ts b/tests/fixtures/supplier-side/sample-config.ts index 45c1700..00d3507 100644 --- a/tests/fixtures/supplier-side/sample-config.ts +++ b/tests/fixtures/supplier-side/sample-config.ts @@ -31,6 +31,8 @@ export function buildSampleConfig(): SupplierConfig { piperUrl: "", piperTimeoutMs: 120_000, capabilityKind: "chat", + maxChatSessions: 1, + chatSettleMode: "full", liveChain: false, }; } diff --git a/tests/unit/buyer-sdk-chat-session.test.ts b/tests/unit/buyer-sdk-chat-session.test.ts new file mode 100644 index 0000000..2959b10 --- /dev/null +++ b/tests/unit/buyer-sdk-chat-session.test.ts @@ -0,0 +1,118 @@ +/** + * buyer-sdk-chat-session.test.ts — settle-mode handling in startChat/endChat. + * + * Ticket-mode suppliers answer /v1/chat/start with {status:"ticket"} and + * /v1/chat/end with {status:"ended", settle_mode:"ticket"}. The SDK must map + * those to settleMode:"ticket" — and, at end, skip receipt verification AND + * the Accept tx entirely (zero chain calls). Full-shaped responses keep the + * existing behavior. + */ + +import { describe, it, expect, vi } from "vitest"; +import { Marketplace } from "../../buyer/src/sdk/Marketplace.js"; +import { MockChainProvider } from "../../packages/shared/src/chain/MockChainProvider.js"; +import { encodeAdvertDatum } from "../../packages/shared/src/cbor/AdvertDatum.js"; +import type { AdvertDatum } from "../../packages/shared/src/cbor/types.js"; +import { buildBuyerWalletKey } from "../fixtures/buyer-side/wallet-keys.js"; + +const ADVERT_TX_HASH = "b".repeat(64); +const ESCROW_REF = { txHash: "f".repeat(64), index: 0 }; +const ESCROW_REF_STR = `${ESCROW_REF.txHash}#0`; + +function advertDatum(): AdvertDatum { + return { + supplier_pkh: "abcdef0123456789abcdef0123456789abcdef0123456789abcdef01", + capability_id: "llm.chat.v1", + model: "kimi", + max_output_tokens: 512, + max_processing_ms: 1_800_000, + price_lovelace: 200_000n, + supplier_bond_lovelace: 1_000_000n, + buyer_bond_lovelace: 1_000_000n, + endpoint_url: "http://supplier.test", + detail_uri: "ipfs://QmTest", + detail_hash: "a".repeat(64), + advertised_at: 1_745_500_000_000, + status: "Active", + }; +} + +/** fetch stub answering the supplier's /status, /v1/chat/start, /v1/chat/end. */ +function supplierFetch(responses: { start?: unknown; end?: unknown }) { + return vi.fn(async (url: RequestInfo | URL) => { + const u = String(url); + if (u.includes("/status")) { + return new Response(JSON.stringify({ status: "free" }), { status: 200 }); + } + if (u.includes("/v1/chat/start")) { + return new Response(JSON.stringify(responses.start ?? { status: "claimed" }), { status: 200 }); + } + if (u.includes("/v1/chat/end")) { + return new Response(JSON.stringify(responses.end ?? {}), { status: 200 }); + } + if (u.includes("/capability")) { + return new Response(JSON.stringify({ supplier_pkh: "", model: "kimi" }), { status: 200 }); + } + return new Response("{}", { status: 200 }); + }) as unknown as typeof globalThis.fetch; +} + +function makeMp(fetchImpl: typeof globalThis.fetch, chain = new MockChainProvider()) { + chain.seed({ + ref: { txHash: ADVERT_TX_HASH, index: 0 }, + address: "addr_test1wfakeadvert", + lovelace: 2_000_000n, + assets: {}, + datumHex: encodeAdvertDatum(advertDatum()), + scriptRef: null, + }); + return { mp: new Marketplace({ + chain, + indexerUrl: "http://indexer.test", + walletKey: buildBuyerWalletKey(), + networkParams: { networkId: 0 }, + _fetch: fetchImpl, + }), chain }; +} + +describe("Marketplace.startChat — settle-mode mapping", () => { + it("maps {status:'ticket'} to settleMode ticket", async () => { + const { mp } = makeMp(supplierFetch({ start: { status: "ticket", settle_mode: "ticket" } })); + const result = await mp.startChat({ advertRef: { txHash: ADVERT_TX_HASH, index: 0 }, payment_lovelace: 200_000n }); + expect(result.settleMode).toBe("ticket"); + }); + + it("maps {status:'claimed'} (and legacy responses) to settleMode full", async () => { + const { mp } = makeMp(supplierFetch({ start: { status: "claimed" } })); + const result = await mp.startChat({ advertRef: { txHash: ADVERT_TX_HASH, index: 0 }, payment_lovelace: 200_000n }); + expect(result.settleMode).toBe("full"); + }); +}); + +describe("Marketplace.endChat — ticket branch", () => { + it("returns the ticket result with zero chain calls and no receipt verification", async () => { + const chain = new MockChainProvider(); + const submitSpy = vi.spyOn(chain, "submitTx"); + const { mp } = makeMp( + supplierFetch({ end: { status: "ended", escrow_ref: ESCROW_REF_STR, settle_mode: "ticket" } }), + chain, + ); + + const result = await mp.endChat({ + escrowRef: ESCROW_REF, + sessionNonce: "nonce-abc", + supplierBaseUrl: "http://supplier.test", + }); + + expect(result.settleMode).toBe("ticket"); + expect("receipt" in result).toBe(false); + expect(submitSpy).not.toHaveBeenCalled(); // no Accept tx + }); + + it("still rejects a receipt-less response that does not declare ticket mode", async () => { + const { mp } = makeMp(supplierFetch({ end: { status: "submitted" } })); + await expect( + mp.endChat({ escrowRef: ESCROW_REF, sessionNonce: "nonce-abc", supplierBaseUrl: "http://supplier.test" }), + ).rejects.toThrow(/receipt/); + }); +}); diff --git a/tests/unit/supplier-chat-session-ticket.test.ts b/tests/unit/supplier-chat-session-ticket.test.ts new file mode 100644 index 0000000..4f52406 --- /dev/null +++ b/tests/unit/supplier-chat-session-ticket.test.ts @@ -0,0 +1,277 @@ +/** + * supplier-chat-session-ticket.test.ts — entry-ticket settle mode + N slots. + * + * Ticket mode (CHAT_SETTLE_MODE=ticket): /v1/chat/start verifies the Open + * escrow but never Claims (zero chain txs), arms no hard-cap timer, and + * /v1/chat/end returns a receipt-less {status:"ended"}. A used escrow ref can + * never open a second session (records are retained). MAX_CHAT_SESSIONS + * admits N concurrent sessions and the N+1th start gets 409 supplier_busy. + */ + +import { describe, it, expect, vi, afterEach } from "vitest"; +import request from "supertest"; +import { createHash } from "crypto"; +import type { Application } from "express"; +import { MockChainProvider } from "../../packages/shared/src/chain/MockChainProvider.js"; +import { canonicalize } from "../../packages/shared/src/cbor/canonical.js"; +import { encodeAdvertDatum } from "../../packages/shared/src/cbor/AdvertDatum.js"; +import { encodeEscrowDatum } from "../../packages/shared/src/cbor/EscrowDatum.js"; +import type { AdvertDatum, EscrowDatum } from "../../packages/shared/src/cbor/types.js"; +import { chatSessionPromptHash } from "../../packages/shared/src/tx/escrow/postChatEscrow.js"; +import type { SupplierConfig } from "../../supplier/src/config.js"; +import { SupplierState } from "../../supplier/src/state.js"; +import { JobStore } from "../../supplier/src/jobs.js"; +import { ChatSessionStore } from "../../supplier/src/chatSession.js"; +import { createApp } from "../../supplier/src/server.js"; +import { buildSampleConfig, SAMPLE_ADVERT_TX_HASH, SAMPLE_ADVERT_INDEX } from "../fixtures/supplier-side/sample-config.js"; +import { buildSupplierWalletKey, SUPPLIER_PKH } from "../fixtures/supplier-side/wallet-keys.js"; + +const MODEL = "kimi"; +const MAX_TOKENS = 512; +const CAPABILITY = "llm.chat.v1"; +const NONCE = "nonce-abc"; +const POSTED_AT = 1_745_500_000_000; +const DELIVER_BY = POSTED_AT + 90_000; + +function sha256Hex(s: string): string { + return createHash("sha256").update(s, "utf8").digest("hex"); +} + +function advertDatum(): AdvertDatum { + return { + supplier_pkh: SUPPLIER_PKH, + capability_id: CAPABILITY, + model: MODEL, + max_output_tokens: MAX_TOKENS, + max_processing_ms: 1_800_000, + price_lovelace: 200_000n, + supplier_bond_lovelace: 1_000_000n, + buyer_bond_lovelace: 1_000_000n, + endpoint_url: "http://supplier.example:8080", + detail_uri: "ipfs://QmTest", + detail_hash: "a".repeat(64), + advertised_at: POSTED_AT, + status: "Active", + }; +} + +function escrowDatum(nonce: string): EscrowDatum { + return { + buyer_pkh: "1234567890abcdef1234567890abcdef1234567890abcdef12345678", + supplier_pkh: SUPPLIER_PKH, + advert_ref: { txHash: SAMPLE_ADVERT_TX_HASH, index: SAMPLE_ADVERT_INDEX }, + capability_id: CAPABILITY, + request_spec_hash: sha256Hex(canonicalize({ capability_id: CAPABILITY, max_output_tokens: MAX_TOKENS, model: MODEL })), + prompt_hash: chatSessionPromptHash({ session_nonce: nonce }), + payment_lovelace: 200_000n, + buyer_bond_lovelace: 1_000_000n, + supplier_bond_lovelace: 1_000_000n, + deliver_by: DELIVER_BY, + posted_at: POSTED_AT, + submitted_at: null, + result_receipt_hash: null, + state: "Open", + }; +} + +interface Harness { + app: Application; + chain: MockChainProvider; + state: SupplierState; + chatSessions: ChatSessionStore; + seedEscrow: (txHashChar: string, nonce?: string) => string; +} + +function makeHarness(overrides?: Partial): Harness { + const chain = new MockChainProvider(); + chain.seed({ + ref: { txHash: SAMPLE_ADVERT_TX_HASH, index: SAMPLE_ADVERT_INDEX }, + address: "addr_test1wfakeadvertaddress", + lovelace: 2_000_000n, + assets: {}, + datumHex: encodeAdvertDatum(advertDatum()), + scriptRef: null, + }); + const config: SupplierConfig = { + ...buildSampleConfig(), + capabilityKind: "chat-session", + llmBackend: "openai", + ollamaUrl: "", + openaiBaseUrl: "http://up", + openaiApiKey: "", + openaiTimeoutMs: 5_000, + openaiMaxTokens: 0, + openaiReasoningDisabled: false, + chatIdleTimeoutMs: 60_000, + chatSettleMode: "ticket", + maxChatSessions: 1, + ...overrides, + } as SupplierConfig; + const state = new SupplierState(config.maxChatSessions); + const chatSessions = new ChatSessionStore(); + const app = createApp({ + chain, + state, + config, + supplierKey: buildSupplierWalletKey(), + jobs: new JobStore(), + chatSessions, + }); + const seedEscrow = (txHashChar: string, nonce = NONCE): string => { + const txHash = txHashChar.repeat(64); + chain.seed({ + ref: { txHash, index: 0 }, + address: "addr_test1wfakeescrowaddress", + lovelace: 2_200_000n, + assets: {}, + datumHex: encodeEscrowDatum(escrowDatum(nonce)), + scriptRef: null, + }); + return `${txHash}#0`; + }; + return { app, chain, state, chatSessions, seedEscrow }; +} + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("chat-session ticket mode — /v1/chat/start", () => { + it("verifies the escrow, skips Claim, and arms no hard-cap timer", async () => { + const h = makeHarness(); + const submitSpy = vi.spyOn(h.chain, "submitTx"); + const ref = h.seedEscrow("e"); + + const res = await request(h.app) + .post("/v1/chat/start") + .set("X-Escrow-Ref", ref) + .send({ session_nonce: NONCE }); + + expect(res.status).toBe(200); + expect(res.body).toEqual({ status: "ticket", escrow_ref: ref, settle_mode: "ticket" }); + expect(submitSpy).not.toHaveBeenCalled(); // zero chain txs + + const record = h.chatSessions.get(ref)!; + expect(record.settleMode).toBe("ticket"); + expect(record.claimedRef).toBeUndefined(); + expect(record.idleTimer).toBeDefined(); + expect(record.hardCapTimer).toBeUndefined(); + expect(h.state.snapshot().activeSessions).toBe(1); + }); + + it("still rejects a foreign-supplier escrow", async () => { + const h = makeHarness(); + const txHash = "c".repeat(64); + h.chain.seed({ + ref: { txHash, index: 0 }, + address: "addr_test1wfakeescrowaddress", + lovelace: 2_200_000n, + assets: {}, + datumHex: encodeEscrowDatum({ ...escrowDatum(NONCE), supplier_pkh: "9".repeat(56) }), + scriptRef: null, + }); + const res = await request(h.app) + .post("/v1/chat/start") + .set("X-Escrow-Ref", `${txHash}#0`) + .send({ session_nonce: NONCE }); + expect(res.status).toBe(403); + expect(res.body.reason).toBe("wrong_supplier"); + }); + + it("rejects reusing an escrow ref while the session is active AND after it ended", async () => { + const h = makeHarness(); + const ref = h.seedEscrow("e"); + + const first = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", ref).send({ session_nonce: NONCE }); + expect(first.status).toBe(200); + + const whileActive = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", ref).send({ session_nonce: NONCE }); + expect(whileActive.status).toBe(409); + expect(whileActive.body.reason).toBe("session_exists"); + + await request(h.app).post("/v1/chat/end").set("X-Escrow-Ref", ref).send({}); + // The escrow is still Open on-chain (never Claimed) — the retained record + // is the only thing standing between a used ticket and a free session. + const afterEnd = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", ref).send({ session_nonce: NONCE }); + expect(afterEnd.status).toBe(409); + expect(afterEnd.body.reason).toBe("session_exists"); + }); +}); + +describe("chat-session ticket mode — /v1/chat/end", () => { + it("ends without a receipt, frees the slot, and is idempotent", async () => { + const h = makeHarness(); + const submitSpy = vi.spyOn(h.chain, "submitTx"); + const ref = h.seedEscrow("e"); + await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", ref).send({ session_nonce: NONCE }); + + const end = await request(h.app).post("/v1/chat/end").set("X-Escrow-Ref", ref).send({}); + expect(end.status).toBe(200); + expect(end.body).toEqual({ status: "ended", escrow_ref: ref, settle_mode: "ticket" }); + expect(end.body.receipt).toBeUndefined(); + expect(submitSpy).not.toHaveBeenCalled(); + expect(h.state.snapshot().activeSessions).toBe(0); + + const again = await request(h.app).post("/v1/chat/end").set("X-Escrow-Ref", ref).send({}); + expect(again.status).toBe(200); + expect(again.body.status).toBe("ended"); + }); +}); + +describe("chat-session slots — MAX_CHAT_SESSIONS", () => { + it("admits N concurrent sessions and 409s the N+1th", async () => { + const h = makeHarness({ maxChatSessions: 2 }); + // NB: "b".repeat(64) is the advert tx hash — do not reuse it for escrows. + const refA = h.seedEscrow("d"); + const refB = h.seedEscrow("e"); + const refC = h.seedEscrow("f"); + + const a = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refA).send({ session_nonce: NONCE }); + const b = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refB).send({ session_nonce: NONCE }); + expect(a.status).toBe(200); + expect(b.status).toBe(200); + + const status = await request(h.app).get("/status"); + expect(status.body.status).toBe("working"); + expect(status.body.active_sessions).toBe(2); + expect(status.body.max_sessions).toBe(2); + + const c = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refC).send({ session_nonce: NONCE }); + expect(c.status).toBe(409); + expect(c.body.reason).toBe("supplier_busy"); + + // Ending one session readmits the third. + await request(h.app).post("/v1/chat/end").set("X-Escrow-Ref", refA).send({}); + const retry = await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refC).send({ session_nonce: NONCE }); + expect(retry.status).toBe(200); + }); + + it("serves concurrent message turns on distinct sessions", async () => { + const h = makeHarness({ maxChatSessions: 2 }); + const refA = h.seedEscrow("d"); + const refB = h.seedEscrow("e"); + await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refA).send({ session_nonce: NONCE }); + await request(h.app).post("/v1/chat/start").set("X-Escrow-Ref", refB).send({ session_nonce: NONCE }); + + const stream = [ + `data: ${JSON.stringify({ choices: [{ delta: { content: "Hello" }, finish_reason: null }] })}\n\n`, + `data: ${JSON.stringify({ choices: [{ delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 3, completion_tokens: 2 } })}\n\n`, + "data: [DONE]\n\n", + ].join(""); + vi.stubGlobal( + "fetch", + vi.fn().mockImplementation(async () => + new Response(stream, { status: 200, headers: { "content-type": "text/event-stream" } })), + ); + + const [ta, tb] = await Promise.all([ + request(h.app).post("/v1/chat/message").set("X-Escrow-Ref", refA).send({ messages: [{ role: "user", content: "hi" }] }), + request(h.app).post("/v1/chat/message").set("X-Escrow-Ref", refB).send({ messages: [{ role: "user", content: "yo" }] }), + ]); + expect(ta.text).toContain('"type":"done"'); + expect(tb.text).toContain('"type":"done"'); + expect(h.chatSessions.get(refA)!.transcript).toHaveLength(2); + expect(h.chatSessions.get(refB)!.transcript).toHaveLength(2); + vi.unstubAllGlobals(); + }); +}); diff --git a/tests/unit/supplier-config.test.ts b/tests/unit/supplier-config.test.ts index 9736b66..990307e 100644 --- a/tests/unit/supplier-config.test.ts +++ b/tests/unit/supplier-config.test.ts @@ -218,3 +218,39 @@ describe("parseAdvertRef()", () => { expect(() => parseAdvertRef(`${"a".repeat(64)}#-1`)).toThrow(); }); }); + +// ─── loadConfig — chat-session concurrency + settle mode ───────────────────── + +describe("loadConfig() — MAX_CHAT_SESSIONS / CHAT_SETTLE_MODE", () => { + const chatSessionEnv = () => ({ + ...buildSampleEnv(), + CAPABILITY_KIND: "chat-session", + LLM_BACKEND: "openai", + OPENAI_BASE_URL: "http://up", + }); + + it("defaults to 1 slot and full settle mode", () => { + const cfg = loadConfig(buildSampleEnv()); + expect(cfg.maxChatSessions).toBe(1); + expect(cfg.chatSettleMode).toBe("full"); + }); + + it("parses MAX_CHAT_SESSIONS and CHAT_SETTLE_MODE for chat-session kind", () => { + const cfg = loadConfig({ ...chatSessionEnv(), MAX_CHAT_SESSIONS: "8", CHAT_SETTLE_MODE: "ticket" }); + expect(cfg.maxChatSessions).toBe(8); + expect(cfg.chatSettleMode).toBe("ticket"); + }); + + it("rejects MAX_CHAT_SESSIONS > 1 for non-chat-session kinds", () => { + expect(() => loadConfig({ ...buildSampleEnv(), MAX_CHAT_SESSIONS: "4" })).toThrow(); + }); + + it("rejects CHAT_SETTLE_MODE=ticket for non-chat-session kinds", () => { + expect(() => loadConfig({ ...buildSampleEnv(), CHAT_SETTLE_MODE: "ticket" })).toThrow(); + }); + + it("rejects invalid values", () => { + expect(() => loadConfig({ ...chatSessionEnv(), MAX_CHAT_SESSIONS: "0" })).toThrow(); + expect(() => loadConfig({ ...chatSessionEnv(), CHAT_SETTLE_MODE: "nope" })).toThrow(); + }); +}); diff --git a/tests/unit/supplier-routes-status.test.ts b/tests/unit/supplier-routes-status.test.ts index 17f0b51..bc69725 100644 --- a/tests/unit/supplier-routes-status.test.ts +++ b/tests/unit/supplier-routes-status.test.ts @@ -117,6 +117,25 @@ describe("GET /status — response body when offline", () => { }); }); +describe("GET /status — session slot fields", () => { + it("reports active_sessions/max_sessions when free", async () => { + const state = new SupplierState(); + const res = await request(makeApp(state)).get("/status"); + expect(res.body.active_sessions).toBe(0); + expect(res.body.max_sessions).toBe(1); + }); + + it("stays free with slots remaining and reports the active count", async () => { + const state = new SupplierState(3); + state.tryAcquire(ESCROW_REF); + const res = await request(makeApp(state)).get("/status"); + expect(res.body.status).toBe("free"); + expect(res.body.active_sessions).toBe(1); + expect(res.body.max_sessions).toBe(3); + expect(res.body.current_escrow_ref).toBeUndefined(); + }); +}); + describe("GET /status — does not touch chain", () => { it("returns 200 even when chain has no UTxOs seeded", async () => { // If the handler incorrectly touches chain, it would throw because diff --git a/tests/unit/supplier-state.test.ts b/tests/unit/supplier-state.test.ts index 0a309c3..7ec87da 100644 --- a/tests/unit/supplier-state.test.ts +++ b/tests/unit/supplier-state.test.ts @@ -1,11 +1,13 @@ /** - * supplier-state.test.ts — RED phase tests for supplier/src/state.ts + * supplier-state.test.ts — tests for supplier/src/state.ts * - * Tests the SupplierState in-process slot lock: - * - tryAcquire(escrowRef): boolean — atomic test-and-set - * - release() + * Tests the SupplierState session-slot admission (capacity-aware) + the + * wallet mutex: + * - tryAcquire(escrowRef): boolean — slot admission (default capacity 1) + * - release(escrowRef) — frees exactly that ref's slot * - markOffline() - * - snapshot(): { status, currentEscrowRef?, lastSeenIso } + * - snapshot(): { status, currentEscrowRef?, activeSessions, maxSessions, ... } + * - walletMutex.run — strict serialization of wallet chain ops */ import { describe, it, expect, beforeEach } from "vitest"; @@ -13,6 +15,7 @@ import { SupplierState } from "../../supplier/src/state.js"; const ESCROW_REF_A = `${"f".repeat(64)}#0`; const ESCROW_REF_B = `${"e".repeat(64)}#1`; +const ESCROW_REF_C = `${"d".repeat(64)}#2`; describe("SupplierState — initial state", () => { it("starts in free state", () => { @@ -102,25 +105,25 @@ describe("SupplierState.release()", () => { it("transitions working → free", () => { state.tryAcquire(ESCROW_REF_A); - state.release(); + state.release(ESCROW_REF_A); expect(state.snapshot().status).toBe("free"); }); it("clears currentEscrowRef on release", () => { state.tryAcquire(ESCROW_REF_A); - state.release(); + state.release(ESCROW_REF_A); expect(state.snapshot().currentEscrowRef).toBeUndefined(); }); it("allows re-acquire after release", () => { state.tryAcquire(ESCROW_REF_A); - state.release(); + state.release(ESCROW_REF_A); expect(state.tryAcquire(ESCROW_REF_B)).toBe(true); }); it("re-acquire after release sets new currentEscrowRef", () => { state.tryAcquire(ESCROW_REF_A); - state.release(); + state.release(ESCROW_REF_A); state.tryAcquire(ESCROW_REF_B); expect(state.snapshot().currentEscrowRef).toBe(ESCROW_REF_B); }); @@ -128,7 +131,7 @@ describe("SupplierState.release()", () => { it("updates lastSeenIso on release", () => { state.tryAcquire(ESCROW_REF_A); const beforeRelease = state.snapshot().lastSeenIso; - state.release(); + state.release(ESCROW_REF_A); const afterRelease = state.snapshot().lastSeenIso; expect(new Date(afterRelease).getTime()).not.toBeNaN(); expect(afterRelease >= beforeRelease).toBe(true); @@ -199,4 +202,94 @@ describe("SupplierState.snapshot()", () => { // State must remain working expect(state.snapshot().status).toBe("working"); }); + + it("reports active/max session counts", () => { + const state = new SupplierState(); + expect(state.snapshot().activeSessions).toBe(0); + expect(state.snapshot().maxSessions).toBe(1); + state.tryAcquire(ESCROW_REF_A); + expect(state.snapshot().activeSessions).toBe(1); + expect(state.snapshot().activeEscrowRefs).toEqual([ESCROW_REF_A]); + }); +}); + +describe("SupplierState — capacity > 1", () => { + let state: SupplierState; + + beforeEach(() => { + state = new SupplierState(2); + }); + + it("stays free until every slot is taken (free-until-full)", () => { + expect(state.tryAcquire(ESCROW_REF_A)).toBe(true); + expect(state.snapshot().status).toBe("free"); + expect(state.snapshot().activeSessions).toBe(1); + expect("currentEscrowRef" in state.snapshot()).toBe(false); + + expect(state.tryAcquire(ESCROW_REF_B)).toBe(true); + expect(state.snapshot().status).toBe("working"); + expect(state.snapshot().currentEscrowRef).toBe(ESCROW_REF_A); // oldest + }); + + it("rejects the N+1th acquire", () => { + state.tryAcquire(ESCROW_REF_A); + state.tryAcquire(ESCROW_REF_B); + expect(state.tryAcquire(ESCROW_REF_C)).toBe(false); + }); + + it("rejects a duplicate ref while held", () => { + state.tryAcquire(ESCROW_REF_A); + expect(state.tryAcquire(ESCROW_REF_A)).toBe(false); + }); + + it("release frees exactly that ref's slot", () => { + state.tryAcquire(ESCROW_REF_A); + state.tryAcquire(ESCROW_REF_B); + state.release(ESCROW_REF_A); + expect(state.snapshot().status).toBe("free"); + expect(state.snapshot().activeEscrowRefs).toEqual([ESCROW_REF_B]); + expect(state.tryAcquire(ESCROW_REF_C)).toBe(true); + }); + + it("release of an unheld ref is a no-op", () => { + state.tryAcquire(ESCROW_REF_A); + state.release(ESCROW_REF_B); + expect(state.snapshot().activeSessions).toBe(1); + }); + + it("markOffline clears all slots and refuses new acquires", () => { + state.tryAcquire(ESCROW_REF_A); + state.tryAcquire(ESCROW_REF_B); + state.markOffline(); + expect(state.snapshot().status).toBe("offline"); + expect(state.snapshot().activeSessions).toBe(0); + expect(state.tryAcquire(ESCROW_REF_C)).toBe(false); + }); + + it("rejects a non-positive capacity", () => { + expect(() => new SupplierState(0)).toThrow(); + }); +}); + +describe("SupplierState.walletMutex", () => { + it("serializes runs in submission order and survives rejection", async () => { + const state = new SupplierState(4); + const order: number[] = []; + const p1 = state.walletMutex.run(async () => { + await new Promise((r) => setTimeout(r, 20)); + order.push(1); + }); + const p2 = state.walletMutex.run(async () => { + order.push(2); + }); + await Promise.all([p1, p2]); + expect(order).toEqual([1, 2]); + + await state.walletMutex + .run(async () => { + throw new Error("boom"); + }) + .catch(() => undefined); + expect(await state.walletMutex.run(async () => 42)).toBe(42); + }); }); diff --git a/tests/unit/supplier-walletHealth.test.ts b/tests/unit/supplier-walletHealth.test.ts index 2b34129..6813ffa 100644 --- a/tests/unit/supplier-walletHealth.test.ts +++ b/tests/unit/supplier-walletHealth.test.ts @@ -4,11 +4,11 @@ * * Coverage: * - Ticker calls consolidate when idle, logs already-healthy - * - Ticker skips when the supplier state is busy - * - Ticker releases the lock around runConsolidateWallet (success + failure) + * - Consolidation runs even while session slots are held (wallet ops don't + * consume session slots) but queues behind a held walletMutex + * - Failure paths never leave the walletMutex poisoned * - On-failure trigger fires consolidate the first time * - On-failure trigger debounces back-to-back invocations - * - On-failure trigger respects the state lock * * The consolidate function is injected via deps so the test never touches * a real Ogmios. @@ -91,7 +91,7 @@ describe("walletHealth — ticker", () => { ticker.stop(); }); - it("skips when supplier state is busy", async () => { + it("runs even while a session slot is held (wallet ops don't consume slots)", async () => { state.tryAcquire(ESCROW_REF); consolidate.mockResolvedValue(healthyResult); @@ -101,16 +101,39 @@ describe("walletHealth — ticker", () => { ); await vi.advanceTimersByTimeAsync(1_000); - await vi.advanceTimersByTimeAsync(0); + await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1)); - expect(consolidate).not.toHaveBeenCalled(); - expect(logLines).toContain("skip: supplier busy"); + // The held session slot is untouched. expect(state.snapshot().status).toBe("working"); + expect(state.snapshot().activeEscrowRefs).toEqual([ESCROW_REF]); ticker.stop(); }); - it("releases the lock after a successful consolidate", async () => { + it("queues behind a held walletMutex instead of racing it", async () => { + consolidate.mockResolvedValue(healthyResult); + let releaseChainOp: () => void = () => undefined; + const chainOp = state.walletMutex.run( + () => new Promise((resolve) => { releaseChainOp = resolve; }), + ); + + const ticker = startWalletHealthTicker( + { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate }, + { intervalMs: 1_000, log }, + ); + + await vi.advanceTimersByTimeAsync(1_000); + // The mutex is held by the fake chain op — consolidate must wait. + expect(consolidate).not.toHaveBeenCalled(); + + releaseChainOp(); + await chainOp; + await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1)); + + ticker.stop(); + }); + + it("logs success after a real consolidate", async () => { consolidate.mockResolvedValue(consolidatedResult); const ticker = startWalletHealthTicker( @@ -127,7 +150,7 @@ describe("walletHealth — ticker", () => { ticker.stop(); }); - it("releases the lock when consolidate throws", async () => { + it("does not poison the walletMutex when consolidate throws", async () => { consolidate.mockRejectedValue(new Error("ogmios unreachable")); const ticker = startWalletHealthTicker( @@ -138,8 +161,9 @@ describe("walletHealth — ticker", () => { await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1)); - expect(state.snapshot().status).toBe("free"); expect(logLines.some((l) => l.includes("consolidate failed"))).toBe(true); + // The mutex still serves the next wallet op. + expect(await state.walletMutex.run(async () => "ok")).toBe("ok"); ticker.stop(); }); @@ -202,7 +226,7 @@ describe("walletHealth — triggerOnFailureConsolidate", () => { expect(consolidate).toHaveBeenCalledTimes(1); }); - it("does not throw when supplier is busy — just skips", async () => { + it("runs even when session slots are held (queues on the wallet mutex)", async () => { state.tryAcquire(ESCROW_REF); consolidate.mockResolvedValue(healthyResult); @@ -213,9 +237,8 @@ describe("walletHealth — triggerOnFailureConsolidate", () => { ), ).not.toThrow(); - await Promise.resolve(); - expect(consolidate).not.toHaveBeenCalled(); - expect(logLines).toContain("skip: supplier busy"); + await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1)); + // The held session slot is untouched. expect(state.snapshot().status).toBe("working"); }); });