feat(supplier,gateway,buyer): N-slot chat suppliers + entry-ticket settle mode

Two structural fixes for demo concurrency (3 single-slot suppliers thrashed
under Open WebUI's ~3 conversations per message):

Multi-slot suppliers (MAX_CHAT_SESSIONS, default 1):
- SupplierState is now capacity-aware: a Set of active escrow refs with
  free-until-full status semantics, ref-scoped release(ref) (also fixes the
  latent bug where handler catch-alls released another request's slot), and
  a walletMutex serializing every wallet-spending chain op — Claim, Submit,
  and wallet-health consolidation queue instead of racing the single
  working/collateral UTxO. /status gains active_sessions/max_sessions.

Entry-ticket settle mode (CHAT_SETTLE_MODE=ticket, default full, both sides):
- The buyer's escrow post at session open is the ONLY chain op. The supplier
  verifies the Open escrow as an entry ticket but never Claims/Submits
  (/v1/chat/start → {status:"ticket"}; /v1/chat/end → receipt-less
  {status:"ended"}); no hard-cap timer, so sessions may outlive deliver_by.
  Used tickets can't reopen sessions (records retained; 409 session_exists).
- Buyer SDK: startChat/endChat are settle-mode aware, driven by the
  supplier's response — mixed-mode rollout windows settle correctly.
- Gateway: ticket closes bill cost 0; the sweeper Reclaims Open tickets
  after deliver_by WITHOUT closing live sessions (the conversation and its
  affinity mirror survive their reclaimed ticket), gates reclaims on
  deliver_by, and keeps an in-process settled-refs memory so terminal
  indexer rows aren't retried every sweep forever.

Deployment (kimi-chat trio + gateway): ticket mode, 8 slots per supplier,
idle policy gateway 10 min / supplier 15 min (fixes the inversion where the
supplier idle-closed before the gateway janitor).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
david
2026-07-09 20:48:52 +02:00
parent 9a4f9f0699
commit 2f7b0194a5
34 changed files with 1451 additions and 235 deletions

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -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<string, string | undefined>, name: string): st
return v;
}
function chatSettleMode(env: Record<string, string | undefined>): "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<string, string | undefined>,
name: string,
@@ -122,6 +135,7 @@ export function loadConfig(env: Record<string, string | undefined>): 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 ?? "")

View File

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

View File

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

View File

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

View File

@@ -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<string>();
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<void> {
/** 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<void> {
// 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<void> {
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<void> {
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<void> {
// 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<void> {
}
} 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 });
}
}
}
}

View File

@@ -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")]),

View File

@@ -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>): 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);
});
});

View File

@@ -126,3 +126,9 @@ ADVERT_DETAIL_HASH=
# `docker compose down && up -d` to apply.
# ESCROW_REF_UTXO=<txhash>#0
# ADVERT_REF_UTXO=<txhash>#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)

View File

@@ -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: [],

View File

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

View File

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

View File

@@ -42,7 +42,8 @@ async function main(): Promise<void> {
// 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 };

View File

@@ -143,17 +143,28 @@ export async function runChatJob(params: RunChatJobParams): Promise<void> {
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<void> {
});
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<void> {
};
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<void> {
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<void> {
});
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<void> {
deps.jobs.complete(jobId, payload);
} finally {
try {
deps.state.release();
deps.state.release(escrowRef);
} catch {
/* never let lock-release escape */
}

View File

@@ -147,6 +147,8 @@ function makeStatusHandler(deps: ResolvedDeps) {
const payload: Record<string, unknown> = {
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<AdvertResult> {
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);
}

View File

@@ -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<unknown> = Promise.resolve();
run<T>(fn: () => Promise<T>): Promise<T> {
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<string>(); // 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;
}
}

View File

@@ -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();
}
}

View File

@@ -31,6 +31,8 @@ export function buildSampleConfig(): SupplierConfig {
piperUrl: "",
piperTimeoutMs: 120_000,
capabilityKind: "chat",
maxChatSessions: 1,
chatSettleMode: "full",
liveChain: false,
};
}

View File

@@ -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/);
});
});

View File

@@ -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<SupplierConfig>): 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();
});
});

View File

@@ -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();
});
});

View File

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

View File

@@ -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);
});
});

View File

@@ -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<void>((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");
});
});