fix(gateway,supplier): demo sessions stick to their supplier; free pinned suppliers

The demo endpoint lost its session on nearly every second message: affinity
required clients to re-echo the assistant reply byte-identically, so clients
that strip <think> blocks or reshape whitespace (Open WebUI, litellm) missed
every turn, opened a fresh escrow session on the next free single-slot
supplier, and left the old one pinned. Open WebUI's background title/tag-gen
requests each pinned another supplier — one message could exhaust all three
kimi replicas ("no available chat supplier").

- affinity: lenient assistant-content matching (think-strip + whitespace
  collapse); user/system/tool messages and tool_calls stay byte-strict;
  strict prefixes outrank lenient ones. The mirror is never rewritten with
  client text, so receipt hashes stay valid.
- demoChat: trim trailing assistant/system messages (regenerate semantics)
  instead of 400 "the last message must have role user or tool".
- supplier: roll the transcript back when the LLM call fails so retries
  don't duplicate the delta (ChatSessionStore.truncateTranscript).
- demoChat: LRU-evict a demo session when no supplier is available (supplier
  /v1/chat/end frees the slot, sweeper Accepts the escrow) and retry the
  open with selectCandidates ignoreStatusFor to bypass the ~20s-stale
  indexer status poll. Max 2 evictions per request.
- sweeper: idle janitor closes demo sessions unused for DEMO_SESSION_IDLE_MS
  (default 3 min, below the supplier's 5 min CHAT_IDLE_TIMEOUT_MS) so
  abandoned one-off sessions release their suppliers quickly.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
david
2026-07-09 02:29:10 +02:00
parent 3d2b4bb203
commit 9a4f9f0699
16 changed files with 636 additions and 43 deletions

View File

@@ -30,5 +30,6 @@ GATEWAY_DB_DIR=./data/gateway
# KEY_RATE_MAX=60 # requests per API key per window
# 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)
# WALLET_HEALTH_INTERVAL_MS=600000
# SDK_REGISTRY_MAX=500 # max cached per-key Marketplace instances

View File

@@ -41,6 +41,10 @@ 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. */
demoSessionIdleMs: number;
walletHealthIntervalMs: number;
sdkRegistryMax: number;
corsOrigins: string[];
@@ -117,6 +121,7 @@ export function loadConfig(env: Record<string, string | undefined>): GatewayConf
windowMs: posInt(env, "DEMO_IP_RATE_WINDOW_MS", 60 * 1000),
},
sweeperIntervalMs: posInt(env, "SWEEPER_INTERVAL_MS", 60 * 1000),
demoSessionIdleMs: posInt(env, "DEMO_SESSION_IDLE_MS", 3 * 60 * 1000),
walletHealthIntervalMs: posInt(env, "WALLET_HEALTH_INTERVAL_MS", 10 * 60 * 1000),
sdkRegistryMax: posInt(env, "SDK_REGISTRY_MAX", 500),
corsOrigins: (env.GATEWAY_CORS_ORIGINS ?? "")

View File

@@ -3,15 +3,21 @@
*
* Stateless OpenAI clients resend the FULL message history each call; sessions
* are stateful. An incoming request belongs to an open session when that
* session's mirrored transcript is a strict prefix of the incoming messages —
* the suffix (delta) is the next turn. Agentic tool loops extend the
* transcript monotonically, so affinity hits nearly always; history divergence
* (IDE compaction, regenerate, retry) just misses and costs one new session.
* session's mirrored transcript is a prefix of the incoming messages — the
* suffix (delta) is the next turn. Client-authored messages (user/system/tool)
* must match byte-exactly, but clients routinely reshape ASSISTANT text when
* echoing it back (strip <think> reasoning, re-render whitespace), so
* assistant content is compared normalized. Strict (byte-exact) matches are
* preferred over lenient ones to disambiguate concurrent threads that share a
* user-message prefix. The mirror itself is NEVER rewritten with client text:
* the delta is forwarded verbatim and mirrored only after a successful turn,
* so the gateway mirror and the supplier transcript stay hash-identical for
* the receipt response_hash.
*
* Pure module — no I/O — so the matching rules are unit-testable.
*/
import { chatMessagesEquivalent, type ChatMessage } from "@marketplace/shared/tx";
import { chatMessagesEquivalent, toolCallsEqual, type ChatMessage } from "@marketplace/shared/tx";
export interface AffinityCandidate {
sessionId: string;
@@ -26,9 +32,39 @@ export interface AffinityMatch {
delta: ChatMessage[];
}
function isPrefix(mirror: ChatMessage[], incoming: ChatMessage[]): boolean {
if (mirror.length >= incoming.length) return false;
return mirror.every((m, i) => chatMessagesEquivalent(m, incoming[i]));
/** Canonicalize assistant text the way clients mangle it when re-echoing
* history: drop <think>…</think> reasoning blocks, collapse whitespace runs,
* trim. Exported for tests. */
export function normalizeAssistantContent(content: string): string {
return content
.replace(/<think>[\s\S]*?<\/think>/gi, "")
.replace(/\s+/g, " ")
.trim();
}
/** Lenient equality: strict for client-authored roles (the client sends the
* same bytes both times); assistant content is compared normalized while
* assistant tool_calls stay strict (ids anchor tool-result turns). */
function chatMessagesLenientEquivalent(a: ChatMessage, b: ChatMessage): boolean {
if (a.role !== b.role) return false;
if (a.role !== "assistant") return chatMessagesEquivalent(a, b);
return (
normalizeAssistantContent(a.content) === normalizeAssistantContent(b.content) &&
toolCallsEqual(a.tool_calls, b.tool_calls)
);
}
type PrefixQuality = "strict" | "lenient";
function classifyPrefix(mirror: ChatMessage[], incoming: ChatMessage[]): PrefixQuality | null {
if (mirror.length >= incoming.length) return null;
let strict = true;
for (let i = 0; i < mirror.length; i++) {
if (chatMessagesEquivalent(mirror[i], incoming[i])) continue;
if (!chatMessagesLenientEquivalent(mirror[i], incoming[i])) return null;
strict = false;
}
return strict ? "strict" : "lenient";
}
function deltaRespondable(delta: ChatMessage[]): boolean {
@@ -36,32 +72,45 @@ function deltaRespondable(delta: ChatMessage[]): boolean {
return last !== undefined && (last.role === "user" || last.role === "tool");
}
/** Drop trailing assistant/system messages so the history ends with a
* respondable message (regenerate semantics for clients that send a trailing
* assistant turn). Empty result = nothing to answer. */
export function trimToRespondable(messages: ChatMessage[]): ChatMessage[] {
let end = messages.length;
while (end > 0 && messages[end - 1].role !== "user" && messages[end - 1].role !== "tool") end--;
return messages.slice(0, end);
}
/**
* Pick the open session whose mirror best matches the incoming history.
* Longest mirror wins (most conversation reuse); ties go to the most recently
* used session. Returns null when nothing matches — caller opens a new
* session and replays the full history. Note `incoming.length <= mirror.length`
* never matches (a client retry of an already-answered request is a miss by
* design: the session transcript must stay append-only).
* Strict prefixes beat lenient ones; then longest mirror wins (most
* conversation reuse); ties go to the most recently used session. Returns
* null when nothing matches — caller opens a new session and replays the full
* history. Note `incoming.length <= mirror.length` never matches (a client
* retry of an already-answered request is a miss by design: the session
* transcript must stay append-only).
*/
export function matchSessionPrefix(
candidates: AffinityCandidate[],
incoming: ChatMessage[],
model: string,
): AffinityMatch | null {
let best: AffinityCandidate | null = null;
let best: { cand: AffinityCandidate; quality: PrefixQuality } | null = null;
for (const cand of candidates) {
if (cand.model !== model) continue;
if (!isPrefix(cand.mirror, incoming)) continue;
const quality = classifyPrefix(cand.mirror, incoming);
if (quality === null) continue;
if (!deltaRespondable(incoming.slice(cand.mirror.length))) continue;
if (
best === null ||
cand.mirror.length > best.mirror.length ||
(cand.mirror.length === best.mirror.length && cand.lastUsedAt > best.lastUsedAt)
(quality === "strict" && best.quality === "lenient") ||
(quality === best.quality &&
(cand.mirror.length > best.cand.mirror.length ||
(cand.mirror.length === best.cand.mirror.length && cand.lastUsedAt > best.cand.lastUsedAt)))
) {
best = cand;
best = { cand, quality };
}
}
if (!best) return null;
return { sessionId: best.sessionId, delta: incoming.slice(best.mirror.length) };
return { sessionId: best.cand.sessionId, delta: incoming.slice(best.cand.mirror.length) };
}

View File

@@ -22,7 +22,7 @@ import type { GatewayDeps } from "../deps.js";
import type { ApiKeyRow, SessionRow } from "../db/store.js";
import type { KeyContext } from "../sdk/registry.js";
import { transcripts, getSessionLock, dropSessionState } from "./transcripts.js";
import { matchSessionPrefix, type AffinityCandidate } from "./affinity.js";
import { matchSessionPrefix, trimToRespondable, type AffinityCandidate } from "./affinity.js";
import {
CAPABILITY,
SessionGoneError,
@@ -32,7 +32,7 @@ import {
} from "./sessions.js";
import type { ParsedChatRequest } from "./validate.js";
import { genId, buildChatCompletion, buildChunk, sseData, SSE_DONE, type Usage } from "./shapes.js";
import { badRequest, toGatewayError, toErrorBody } from "./errors.js";
import { GatewayError, badRequest, toGatewayError, toErrorBody } from "./errors.js";
interface DemoSessionMeta {
keyId: string;
@@ -87,6 +87,127 @@ function markSessionGone(deps: GatewayDeps, session: SessionRow): void {
recordSessionCost(deps, session);
}
/** Ask the supplier to settle (Submit) the session. NO gateway chain work —
* the sweeper Accepts the resulting Submitted escrow. The supplier releases
* its single slot as part of ending, so a truthy return means the supplier is
* free again. Best-effort: on failure the sweeper reclaims later. */
async function requestSupplierEnd(deps: GatewayDeps, session: SessionRow): Promise<boolean> {
try {
const res = await deps.fetchFn(`${session.supplier_base_url.replace(/\/+$/, "")}/v1/chat/end`, {
method: "POST",
headers: { "content-type": "application/json", "X-Escrow-Ref": session.escrow_ref },
body: JSON.stringify({}),
signal: AbortSignal.timeout(90_000),
});
await res.text().catch(() => "");
return res.ok;
} catch {
return false;
}
}
/** Close one demo session (evicted for capacity, or idle-janitored). Runs
* under the session lock so an in-flight turn finishes first; takes NO key
* mutex (no chain work here — see requestSupplierEnd). Returns the freed
* supplier_pkh, or null when the session was already gone or (janitor) was
* used again while we waited for the lock. */
async function closeDemoSession(
deps: GatewayDeps,
sessionId: string,
opts: { reason: "evicted" | "idle"; idleCutoffMs?: number; nowMs?: number },
): Promise<string | null> {
if (!demoSessions.has(sessionId)) return null;
return getSessionLock(sessionId).run(async () => {
const meta = demoSessions.get(sessionId);
if (!meta) return null; // lost a race with another closer
// Re-check idleness inside the lock: a turn that won the lock first has
// bumped lastUsedAt past the sweep's reference clock.
if (opts.idleCutoffMs !== undefined && (opts.nowMs ?? Date.now()) - meta.lastUsedAt < opts.idleCutoffMs) {
return null;
}
const session = deps.store.getSession(sessionId);
if (!session || session.state !== "open") {
dropDemoSession(sessionId);
return null;
}
await requestSupplierEnd(deps, session);
markSessionGone(deps, session);
// eslint-disable-next-line no-console
console.log(`[gateway:demo] closed session ${sessionId} (${opts.reason}), freed supplier ${session.supplier_pkh}`);
return session.supplier_pkh;
});
}
/** Least-recently-used open demo session for (key, model), excluding prior
* victims. Same-model scoping doubles as a guard: a bogus model name has no
* demo sessions, so eviction can never loop on a genuine model_not_found. */
function pickLruDemoSession(keyId: string, model: string, exclude: ReadonlySet<string>): string | null {
let lruId: string | null = null;
let lruAt = Infinity;
for (const [sessionId, meta] of demoSessions) {
if (meta.keyId !== keyId || meta.model !== model || exclude.has(sessionId)) continue;
if (meta.lastUsedAt < lruAt) {
lruAt = meta.lastUsedAt;
lruId = sessionId;
}
}
return lruId;
}
/** Close demo sessions idle past deps.config.demoSessionIdleMs so abandoned
* conversations (client gone, title-gen one-offs) release their single-slot
* suppliers. Called from the sweeper tick; sequential to keep supplier
* /v1/chat/end calls orderly. */
export async function sweepIdleDemoSessions(deps: GatewayDeps, nowMs = Date.now()): Promise<void> {
const idleMs = deps.config.demoSessionIdleMs;
for (const [sessionId, meta] of [...demoSessions]) {
if (nowMs - meta.lastUsedAt < idleMs) continue;
try {
await closeDemoSession(deps, sessionId, { reason: "idle", idleCutoffMs: idleMs, nowMs });
} catch (err) {
// eslint-disable-next-line no-console
console.error(`[gateway:demo] idle close of ${sessionId} failed:`, err instanceof Error ? err.message : err);
}
}
}
const MAX_EVICTIONS = 2;
function isNoSupplierError(err: unknown): boolean {
return err instanceof GatewayError && (err.code === "model_not_found" || err.code === "overloaded");
}
/** Open a session, evicting up to MAX_EVICTIONS LRU demo sessions when every
* supplier is pinned. Eviction frees the supplier immediately (its slot
* releases at Submit) but the indexer's status poll lags, so the retry passes
* the freed pkhs through selectCandidates' ignoreStatusFor. Lock ordering is
* safe: eviction holds only the session lock, the open holds only the key
* mutex — never nested. */
async function openSessionWithEviction(
deps: GatewayDeps,
keyRow: ApiKeyRow,
ctx: KeyContext,
model: string,
): Promise<SessionRow> {
const freedPkhs = new Set<string>();
const evicted = new Set<string>();
for (let attempt = 0; ; attempt++) {
try {
return await ctx.mutex.run(() =>
openSessionCore(deps, keyRow, ctx, model, freedPkhs.size > 0 ? { ignoreStatusFor: freedPkhs } : undefined),
);
} catch (err) {
if (attempt >= MAX_EVICTIONS || !isNoSupplierError(err)) throw err;
const victimId = pickLruDemoSession(keyRow.id, model, evicted);
if (!victimId) throw err;
evicted.add(victimId);
const pkh = await closeDemoSession(deps, victimId, { reason: "evicted" });
if (!pkh) throw err; // nothing actually freed — don't loop
freedPkhs.add(pkh);
}
}
}
type LockedTurnOutcome =
| { kind: "done"; result: TurnResult }
| { kind: "retry" } // candidate no longer valid — re-run the match
@@ -105,12 +226,17 @@ export async function runDemoChat(
parsed: ParsedChatRequest,
res: Response,
): Promise<void> {
const last = parsed.messages[parsed.messages.length - 1];
if (last.role !== "user" && last.role !== "tool") {
// Reject before any chain work — opening an escrow for an unanswerable
// request would burn a session fee for a guaranteed supplier 400.
throw badRequest("invalid_messages", "the last message must have role user or tool");
// Clients sometimes send histories ending in an assistant (continue/retry
// flows) or system message; the chat.v1 supplier can only respond to a
// user/tool-terminated turn. Trim trailing unrespondable messages
// (regenerate semantics) and reject — before any chain work — only when
// nothing respondable remains.
const trimmed = trimToRespondable(parsed.messages);
if (trimmed.length === 0) {
throw badRequest("invalid_messages", "no user or tool message to respond to");
}
const effective: ParsedChatRequest =
trimmed.length === parsed.messages.length ? parsed : { ...parsed, messages: trimmed };
const id = genId();
let keepalive: ReturnType<typeof setInterval> | undefined;
@@ -159,7 +285,7 @@ export async function runDemoChat(
};
try {
const result = await runWithAffinity(deps, keyRow, ctx, parsed, sink);
const result = await runWithAffinity(deps, keyRow, ctx, effective, sink);
if (keepalive) clearInterval(keepalive);
const usage: Usage = {
@@ -255,18 +381,24 @@ async function runWithAffinity(
// retry: candidate vanished/advanced — re-match against current state.
}
// Miss: open a new session (chain work → key mutex) and replay the full
// history as its first turn.
const session = await ctx.mutex.run(() => openSessionCore(deps, keyRow, ctx, parsed.model));
// Miss: open a new session (chain work → key mutex; evicts an LRU demo
// session when every supplier is pinned) and replay the full history as its
// first turn.
const session = await openSessionWithEviction(deps, keyRow, ctx, parsed.model);
demoSessions.set(session.id, { keyId: keyRow.id, model: parsed.model, lastUsedAt: Date.now() });
return getSessionLock(session.id).run(async () => {
try {
return await streamSupplierTurn(
const result = await streamSupplierTurn(
deps,
session,
{ messages: parsed.messages, tools: parsed.tools, tool_choice: parsed.toolChoice },
sink.onToken,
);
// Refresh recency after the (possibly long) first stream so this live
// thread isn't the LRU/janitor's first victim.
const meta = demoSessions.get(session.id);
if (meta) meta.lastUsedAt = Date.now();
return result;
} catch (err) {
if (err instanceof SessionGoneError) {
// Freshly opened yet already gone (supplier crashed mid-claim).

View File

@@ -77,12 +77,14 @@ export async function openSessionCore(
keyRow: ApiKeyRow,
ctx: KeyContext,
model: string,
opts?: { ignoreStatusFor?: ReadonlySet<string> },
): Promise<SessionRow> {
const candidates = await selectCandidates({
indexerUrl: deps.config.indexerUrl,
model,
capabilityId: CAPABILITY,
fetchFn: deps.fetchFn,
ignoreStatusFor: opts?.ignoreStatusFor,
});
if (candidates.length === 0) {
throw notFound("model_not_found", `no available chat (llm.chat.v1) supplier for model "${model}"`);

View File

@@ -56,6 +56,11 @@ export interface SelectSupplierOpts {
model: string;
capabilityId: string;
fetchFn?: typeof globalThis.fetch;
/** supplier_pkhs eligible regardless of polled status — used after the demo
* executor frees a supplier the indexer poller (~20 s cadence) still reports
* busy. sdk.startChat's live /status pre-check still guards genuinely-busy
* suppliers. */
ignoreStatusFor?: ReadonlySet<string>;
}
/**
@@ -79,7 +84,9 @@ export async function selectCandidates(opts: SelectSupplierOpts): Promise<Suppli
if (raw.capability_id !== opts.capabilityId) continue;
if (raw.model !== opts.model) continue;
if (raw.advert_status !== "Active") continue;
if (raw.status !== "free" && raw.status !== "unknown") continue;
if (raw.status !== "free" && raw.status !== "unknown" && !opts.ignoreStatusFor?.has(raw.supplier_pkh)) {
continue;
}
const advertRef = parseRef(raw.utxo_ref);
if (!advertRef) continue;
candidates.push({

View File

@@ -33,7 +33,7 @@ import type { SessionRow } from "./db/store.js";
import { fetchEscrows, acceptAndConfirm, reclaimAndConfirm, type EscrowRow } from "./onchain/settle.js";
import { parseRef } from "./routing/selectSupplier.js";
import { dropSessionState } from "./openai/transcripts.js";
import { dropDemoSession } from "./openai/demoChat.js";
import { dropDemoSession, sweepIdleDemoSessions } from "./openai/demoChat.js";
import { CAPABILITY as CHAT_CAPABILITY } from "./openai/sessions.js";
// Don't touch escrows newer than this — an in-flight request is settling them.
@@ -78,6 +78,12 @@ function closeSessionRow(
}
export async function runSweepOnce(deps: GatewayDeps): 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) =>
log(`demo idle janitor: ${err instanceof Error ? err.message : String(err)}`),
);
const now = Date.now();
for (const keyRow of deps.store.listKeys()) {
let rows: EscrowRow[];

View File

@@ -3,7 +3,12 @@
*/
import { describe, it, expect } from "vitest";
import { matchSessionPrefix, type AffinityCandidate } from "../src/openai/affinity.js";
import {
matchSessionPrefix,
normalizeAssistantContent,
trimToRespondable,
type AffinityCandidate,
} from "../src/openai/affinity.js";
import type { ChatMessage } from "@marketplace/shared/tx";
const u = (content: string): ChatMessage => ({ role: "user", content });
@@ -78,4 +83,61 @@ describe("matchSessionPrefix", () => {
const incoming = [u("hi"), { role: "assistant" as const, content: "", tool_calls: [{ ...TOOL_CALL }] }, { role: "tool" as const, content: "x", tool_call_id: "call_1" }];
expect(matchSessionPrefix([cand("s1", mirror)], incoming, "m")?.sessionId).toBe("s1");
});
it("matches leniently when the client reshapes the assistant echo (think-strip + whitespace)", () => {
const mirror = [u("hi"), a("<think>plan it</think>Hello there\n")];
const got = matchSessionPrefix([cand("s1", mirror)], [u("hi"), a("Hello there"), u("more")], "m");
expect(got).toEqual({ sessionId: "s1", delta: [u("more")] });
});
it("stays strict on user/tool divergence even with lenient assistant matching", () => {
const mirror = [u("hi "), a("Hello")];
// User content differs by whitespace — client-authored roles must be byte-exact.
expect(matchSessionPrefix([cand("s1", mirror)], [u("hi"), a("Hello"), u("more")], "m")).toBeNull();
});
it("rejects assistant tool_calls divergence under lenient matching", () => {
const mirror = [u("hi"), { role: "assistant" as const, content: "x", tool_calls: [TOOL_CALL] }];
const changed = { ...TOOL_CALL, function: { name: "get_time", arguments: '{"tz":"utc"}' } };
const incoming = [u("hi"), { role: "assistant" as const, content: "x ", tool_calls: [changed] }, { role: "tool" as const, content: "t", tool_call_id: "call_1" }];
expect(matchSessionPrefix([cand("s1", mirror)], incoming, "m")).toBeNull();
});
it("prefers a strict match over a lenient one regardless of recency", () => {
const incoming = [u("hi"), a("Hello"), u("more")];
const got = matchSessionPrefix(
[
cand("lenient", [u("hi"), a("Hello ")], 99),
cand("strict", [u("hi"), a("Hello")], 1),
],
incoming,
"m",
);
expect(got?.sessionId).toBe("strict");
});
});
describe("normalizeAssistantContent", () => {
it("strips think blocks, collapses whitespace and trims", () => {
expect(normalizeAssistantContent("<think>a\nb</think> Hi\n\nthere friend ")).toBe("Hi there friend");
expect(normalizeAssistantContent("<THINK>x</THINK>ok")).toBe("ok");
expect(normalizeAssistantContent("plain")).toBe("plain");
});
});
describe("trimToRespondable", () => {
it("drops trailing assistant/system messages", () => {
expect(trimToRespondable([u("hi"), a("partial")])).toEqual([u("hi")]);
expect(trimToRespondable([u("hi"), a("x"), { role: "system" as const, content: "nudge" }])).toEqual([u("hi")]);
});
it("returns empty when nothing respondable remains", () => {
expect(trimToRespondable([a("hello")])).toEqual([]);
expect(trimToRespondable([])).toEqual([]);
});
it("keeps already-respondable histories unchanged", () => {
const msgs = [u("hi"), a("x"), u("more")];
expect(trimToRespondable(msgs)).toEqual(msgs);
});
});

View File

@@ -4,6 +4,7 @@ import { join } from "path";
import { randomBytes, randomUUID } from "crypto";
import request from "supertest";
import { createApp } from "../src/server.js";
import { sweepIdleDemoSessions } from "../src/openai/demoChat.js";
import { GatewayStore } from "../src/db/store.js";
import { SdkRegistry } from "../src/sdk/registry.js";
import { genPrivKeyHex, deriveWalletKey } from "../src/wallet.js";
@@ -37,6 +38,7 @@ function makeDeps(
keyRate: { max: 1000, windowMs: 60_000 },
demoIpRate: overrides?.demoIpRate ?? { max: 1000, windowMs: 60_000 },
sweeperIntervalMs: 60_000,
demoSessionIdleMs: 180_000,
walletHealthIntervalMs: 600_000,
sdkRegistryMax: 100,
corsOrigins: [],
@@ -136,11 +138,14 @@ function doneFrame(message: Record<string, unknown>, finishReason: string, usage
return { type: "done", message, finish_reason: finishReason, usage };
}
/** fetchFn serving the indexer supplier list and scripted /v1/chat/message responses. */
function demoFetch(messageResponses: Array<() => Response>) {
/** fetchFn serving the indexer supplier list and scripted /v1/chat/message
* responses. Pass `suppliers` to serve a mutable list (eviction tests flip
* status mid-test); /v1/chat/end calls are recorded and 200-acked. */
function demoFetch(messageResponses: Array<() => Response>, suppliers?: () => unknown[]) {
const messageCalls: Array<Record<string, unknown>> = [];
const endCalls: string[] = [];
let i = 0;
const fetchFn = (async (url: unknown, init?: { body?: string }) => {
const fetchFn = (async (url: unknown, init?: { body?: string; headers?: Record<string, string> }) => {
const u = String(url);
if (u.includes("/v1/chat/message")) {
messageCalls.push(init?.body ? (JSON.parse(init.body) as Record<string, unknown>) : {});
@@ -148,10 +153,14 @@ function demoFetch(messageResponses: Array<() => Response>) {
i += 1;
return make();
}
if (u.includes("/suppliers")) return new Response(JSON.stringify(SUPPLIERS), { status: 200 });
if (u.includes("/v1/chat/end")) {
endCalls.push(init?.headers?.["X-Escrow-Ref"] ?? "");
return new Response(JSON.stringify({ status: "submitted" }), { status: 200 });
}
if (u.includes("/suppliers")) return new Response(JSON.stringify(suppliers ? suppliers() : SUPPLIERS), { status: 200 });
return new Response("[]", { status: 200 });
}) as unknown as typeof globalThis.fetch;
return { fetchFn, messageCalls };
return { fetchFn, messageCalls, endCalls };
}
/** Insert a demo-flagged key directly (public signup can't mint one) and stub
@@ -284,7 +293,7 @@ describe("gateway demo key", () => {
expect(usage.some((u) => u.kind === "chat_session" && u.cost_lovelace === "1000000")).toBe(true);
});
it("rejects histories that do not end with user/tool before any chain work", async () => {
it("rejects histories with nothing respondable before any chain work", async () => {
const deps = makeDeps(undefined, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const res = await request(createApp(deps))
@@ -292,9 +301,160 @@ describe("gateway demo key", () => {
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "assistant", content: "hello" }] });
expect(res.status).toBe(400);
expect(res.body.error.code).toBe("invalid_messages");
expect(startChat).not.toHaveBeenCalled();
});
it("answers a trailing-assistant history by trimming to the last user message (regenerate)", async () => {
const { fetchFn, messageCalls } = demoFetch([
() => supplierSse([{ type: "token", value: "Again" }, doneFrame({ role: "assistant", content: "Again" }, "stop")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const res = await request(createApp(deps))
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({
model: "kimi",
messages: [
{ role: "user", content: "hi" },
{ role: "assistant", content: "partial answer" },
],
});
expect(res.status).toBe(200);
expect(res.body.choices[0].message.content).toBe("Again");
expect(startChat).toHaveBeenCalledTimes(1);
// The trailing assistant message is dropped from the replay.
expect((messageCalls[0].messages as Array<{ role: string }>).map((m) => m.role)).toEqual(["user"]);
});
it("keeps the session when the client reshapes the assistant echo (lenient affinity)", async () => {
const { fetchFn, messageCalls } = demoFetch([
() => supplierSse([{ type: "token", value: "<think>hm</think>Hi there" }, doneFrame({ role: "assistant", content: "<think>hm</think>Hi there" }, "stop")]),
() => supplierSse([{ type: "token", value: "Sure" }, doneFrame({ role: "assistant", content: "Sure" }, "stop")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const app = createApp(deps);
const res1 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res1.status).toBe(200);
// Client echoes the assistant reply with the think block stripped and
// whitespace collapsed — still the same session, delta only.
const res2 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({
model: "kimi",
messages: [
{ role: "user", content: "hi" },
{ role: "assistant", content: "Hi there" },
{ role: "user", content: "more" },
],
});
expect(res2.status).toBe(200);
expect(res2.body.choices[0].message.content).toBe("Sure");
expect(startChat).toHaveBeenCalledTimes(1);
expect((messageCalls[1].messages as unknown[]).length).toBe(1);
});
it("evicts the LRU demo session when every supplier is pinned", async () => {
const suppliers = [{ ...SUPPLIERS[1] }]; // single chat supplier, mutable status
const { fetchFn, messageCalls, endCalls } = demoFetch(
[
() => supplierSse([{ type: "token", value: "Hi" }, doneFrame({ role: "assistant", content: "Hi" }, "stop")]),
() => supplierSse([{ type: "token", value: "New" }, doneFrame({ role: "assistant", content: "New" }, "stop")]),
],
() => suppliers,
);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const app = createApp(deps);
const res1 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res1.status).toBe(200);
// The lone supplier is now pinned by session 1 (indexer sees it working).
suppliers[0] = { ...suppliers[0], status: "working" };
// Unrelated history → affinity miss → no free supplier → evict session 1
// (supplier /v1/chat/end) and retry, bypassing the stale indexer status.
const res2 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "different topic" }] });
expect(res2.status).toBe(200);
expect(res2.body.choices[0].message.content).toBe("New");
expect(startChat).toHaveBeenCalledTimes(2);
expect(endCalls).toHaveLength(1);
expect((messageCalls[1].messages as unknown[]).length).toBe(1);
// Evicted session closed + billed.
const keyId = deps.store.listAllKeys().find((k) => k.demo === 1)!.id;
const usage = deps.store.listUsage(keyId, 10);
expect(usage.some((u) => u.kind === "chat_session" && u.cost_lovelace === "1000000")).toBe(true);
expect(deps.store.listOpenSessionsByKey(keyId)).toHaveLength(1); // only session 2
});
it("surfaces the original no-supplier error when there is nothing to evict", async () => {
const suppliers = [{ ...SUPPLIERS[1], status: "working" }];
const { fetchFn, endCalls } = demoFetch([() => new Response("unused", { status: 500 })], () => suppliers);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const res = await request(createApp(deps))
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res.status).toBe(404);
expect(res.body.error.code).toBe("model_not_found");
expect(startChat).not.toHaveBeenCalled();
expect(endCalls).toHaveLength(0);
});
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")]),
() => supplierSse([{ type: "token", value: "Fresh" }, doneFrame({ role: "assistant", content: "Fresh" }, "stop")]),
]);
const deps = makeDeps(fetchFn, { chain: fundedChain });
const { rawKey, startChat } = setupDemoKey(deps);
const app = createApp(deps);
const res1 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({ model: "kimi", messages: [{ role: "user", content: "hi" }] });
expect(res1.status).toBe(200);
// Pretend demoSessionIdleMs elapsed: the sweeper tick closes the session.
await sweepIdleDemoSessions(deps, Date.now() + deps.config.demoSessionIdleMs + 1);
expect(endCalls).toHaveLength(1);
const keyId = deps.store.listAllKeys().find((k) => k.demo === 1)!.id;
expect(deps.store.listOpenSessionsByKey(keyId)).toHaveLength(0);
// Continuing the thread now misses affinity and reopens with full replay.
const res2 = await request(app)
.post("/openai/v1/chat/completions")
.set("authorization", `Bearer ${rawKey}`)
.send({
model: "kimi",
messages: [
{ role: "user", content: "hi" },
{ role: "assistant", content: "Hi" },
{ role: "user", content: "more" },
],
});
expect(res2.status).toBe(200);
expect(startChat).toHaveBeenCalledTimes(2);
expect((messageCalls[1].messages as unknown[]).length).toBe(3);
});
it("blocks withdraw and hides the deposit address for demo keys", async () => {
const deps = makeDeps();
const { rawKey } = setupDemoKey(deps);

View File

@@ -92,6 +92,23 @@ describe("routing/selectSupplier", () => {
expect(parseRef("aa".repeat(32) + "#3")).toEqual({ txHash: "aa".repeat(32), index: 3 });
expect(parseRef("nope")).toBeNull();
});
it("ignoreStatusFor readmits a busy supplier by pkh (stale indexer status)", async () => {
const busy = [
{ ...rows[0], status: "working" },
{ ...rows[1], status: "offline" },
];
const base = { indexerUrl: "http://ix", model: "m", capabilityId: "llm.text.generate.v1" };
const excluded = await selectCandidates({ ...base, fetchFn: jsonResponse(busy) });
expect(excluded).toEqual([]);
const readmitted = await selectCandidates({
...base,
fetchFn: jsonResponse(busy),
ignoreStatusFor: new Set(["s1"]),
});
expect(readmitted.map((c) => c.supplierPkh)).toEqual(["s1"]);
});
});
describe("openai/validate", () => {

View File

@@ -62,7 +62,7 @@ export function normalizeChatMessage(raw: unknown): ChatMessage | null {
return msg;
}
function toolCallsEqual(a?: ToolCall[], b?: ToolCall[]): boolean {
export function toolCallsEqual(a?: ToolCall[], b?: ToolCall[]): boolean {
const la = a ?? [];
const lb = b ?? [];
if (la.length !== lb.length) return false;

View File

@@ -4,7 +4,7 @@
export type { ChatMessage, ToolCall, WalletKey, BuildResult, PostAdvertBuildResult, PostEscrowBuildResult } from "./types.js";
export { TxConstructionError } from "./types.js";
export { normalizeChatMessage, chatMessagesEquivalent } from "./chatMessage.js";
export { normalizeChatMessage, chatMessagesEquivalent, toolCallsEqual } from "./chatMessage.js";
export { mockSlotToWallclockMs, mockWallclockMsToSlot, NETWORK_BUFFER_MS } from "./internal/constants.js";
export { detectCborBackend, type CborBackend } from "./internal/cborBackend.js";
export type { Signer } from "./signer.js";

View File

@@ -113,6 +113,16 @@ export class ChatSessionStore {
r.lastActivityMs = Date.now();
}
/** Roll back a failed turn: truncate the transcript to `length`. The gateway
* mirror only appends after a successful turn, so the delta appended at turn
* start must be removed on LLM failure to keep both sides hash-identical. */
truncateTranscript(escrowRef: string, length: number): void {
const r = this.records.get(escrowRef);
if (!r || length < 0 || r.transcript.length <= length) return;
r.transcript.length = length;
r.lastActivityMs = Date.now();
}
touch(escrowRef: string): void {
const r = this.records.get(escrowRef);
if (r) r.lastActivityMs = Date.now();

View File

@@ -870,6 +870,7 @@ function makeChatSessionHandlers(deps: ResolvedDeps) {
// Pause the idle timer while a turn is in flight; re-arm when it ends so
// a long generation never trips the auto-end mid-stream.
if (record.idleTimer) { clearTimeout(record.idleTimer); record.idleTimer = undefined; }
const transcriptLenBefore = record.transcript.length;
deps.chatSessions.appendMessages(escrowRefStr, delta);
res.status(200);
@@ -919,6 +920,9 @@ function makeChatSessionHandlers(deps: ResolvedDeps) {
usage: { prompt_tokens: result.prompt_tokens, completion_tokens: result.completion_tokens },
});
} catch (err) {
// Failed turn: the gateway mirror never saw this delta, so drop it here
// too — otherwise a retry re-sends it and the transcripts diverge.
deps.chatSessions.truncateTranscript(escrowRefStr, transcriptLenBefore);
const errMsg = err instanceof Error ? err.message : String(err);
sse({ type: "error", message: errMsg });
} finally {

View File

@@ -8,7 +8,7 @@
*/
import { describe, it, expect } from "vitest";
import { normalizeChatMessage, chatMessagesEquivalent } from "../../packages/shared/src/tx/chatMessage.js";
import { normalizeChatMessage, chatMessagesEquivalent, toolCallsEqual } from "../../packages/shared/src/tx/chatMessage.js";
import { canonicalize } from "../../packages/shared/src/cbor/canonical.js";
const TOOL_CALL = { id: "call_1", type: "function", function: { name: "get_time", arguments: "{}" } };
@@ -94,3 +94,19 @@ describe("chatMessagesEquivalent", () => {
expect(chatMessagesEquivalent(a, c)).toBe(false);
});
});
describe("toolCallsEqual", () => {
const CALL = { id: "call_1", type: "function" as const, function: { name: "get_time", arguments: "{}" } };
it("treats missing and empty lists as equal (exported for the affinity matcher)", () => {
expect(toolCallsEqual(undefined, undefined)).toBe(true);
expect(toolCallsEqual(undefined, [])).toBe(true);
expect(toolCallsEqual([CALL], [{ ...CALL }])).toBe(true);
});
it("rejects id/name/arguments differences and length mismatches", () => {
expect(toolCallsEqual([CALL], [])).toBe(false);
expect(toolCallsEqual([CALL], [{ ...CALL, id: "call_2" }])).toBe(false);
expect(toolCallsEqual([CALL], [{ ...CALL, function: { name: "get_time", arguments: '{"tz":"utc"}' } }])).toBe(false);
});
});

View File

@@ -0,0 +1,122 @@
/**
* supplier-chat-session-rollback.test.ts — failed-turn transcript rollback.
*
* The chat-session message handler appends the incoming delta BEFORE calling
* the LLM backend; the gateway mirror only appends after a successful turn.
* On backend failure the supplier must truncate the delta back out, so a
* retried turn doesn't duplicate messages and both transcripts stay
* hash-identical for the receipt.
*/
import { describe, it, expect, vi, afterEach } from "vitest";
import request from "supertest";
import { MockChainProvider } from "../../packages/shared/src/chain/MockChainProvider.js";
import type { AdvertDatum, EscrowDatum } from "../../packages/shared/src/cbor/types.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 } from "../fixtures/supplier-side/sample-config.js";
import { buildSupplierWalletKey } from "../fixtures/supplier-side/wallet-keys.js";
const ESCROW_REF = `${"e".repeat(64)}#0`;
function chatSessionConfig(): SupplierConfig {
return {
...buildSampleConfig(),
capabilityKind: "chat-session",
llmBackend: "openai",
ollamaUrl: "",
openaiBaseUrl: "http://up",
openaiApiKey: "",
openaiTimeoutMs: 5_000,
openaiMaxTokens: 0,
openaiReasoningDisabled: false,
chatIdleTimeoutMs: 60_000,
} as SupplierConfig;
}
function makeAppWithSession() {
const chatSessions = new ChatSessionStore();
const record = chatSessions.create({
escrowRef: ESCROW_REF,
claimedRef: { txHash: "c".repeat(64), index: 0 },
advert: { model: "kimi" } as AdvertDatum,
escrowDatum: {} as EscrowDatum,
});
const app = createApp({
chain: new MockChainProvider(),
state: new SupplierState(),
config: chatSessionConfig(),
supplierKey: buildSupplierWalletKey(),
jobs: new JobStore(),
chatSessions,
});
return { app, chatSessions, record };
}
function sseBody(frames: unknown[]): string {
return frames.map((f) => `data: ${typeof f === "string" ? f : JSON.stringify(f)}\n\n`).join("");
}
const OK_STREAM = sseBody([
{ choices: [{ delta: { content: "Hello" }, finish_reason: null }] },
{ choices: [{ delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 3, completion_tokens: 2 } },
"[DONE]",
]);
afterEach(() => {
vi.unstubAllGlobals();
});
describe("ChatSessionStore.truncateTranscript", () => {
it("rolls an appended delta back and no-ops on stale targets", () => {
const store = new ChatSessionStore();
store.create({
escrowRef: ESCROW_REF,
claimedRef: { txHash: "c".repeat(64), index: 0 },
advert: { model: "kimi" } as AdvertDatum,
escrowDatum: {} as EscrowDatum,
});
store.appendMessages(ESCROW_REF, [{ role: "user", content: "hi" }]);
store.truncateTranscript(ESCROW_REF, 0);
expect(store.get(ESCROW_REF)!.transcript).toEqual([]);
store.appendMessages(ESCROW_REF, [{ role: "user", content: "hi" }]);
store.truncateTranscript(ESCROW_REF, 5); // target beyond current length
store.truncateTranscript(ESCROW_REF, -1); // nonsense target
store.truncateTranscript("missing", 0); // unknown session
expect(store.get(ESCROW_REF)!.transcript).toHaveLength(1);
});
});
describe("POST /v1/chat/message — failed-turn rollback", () => {
it("truncates the delta on backend failure, then a retry appends exactly once", async () => {
const { app, record } = makeAppWithSession();
const delta = [{ role: "user", content: "hi" }];
vi.stubGlobal("fetch", vi.fn().mockRejectedValue(new Error("boom")));
const failed = await request(app)
.post("/v1/chat/message")
.set("X-Escrow-Ref", ESCROW_REF)
.send({ messages: delta });
expect(failed.text).toContain('"type":"error"');
// The failed delta was rolled back — nothing to duplicate on retry.
expect(record.transcript).toEqual([]);
vi.stubGlobal(
"fetch",
vi.fn().mockResolvedValue(new Response(OK_STREAM, { status: 200, headers: { "content-type": "text/event-stream" } })),
);
const ok = await request(app)
.post("/v1/chat/message")
.set("X-Escrow-Ref", ESCROW_REF)
.send({ messages: delta });
expect(ok.text).toContain('"type":"done"');
expect(record.transcript).toEqual([
{ role: "user", content: "hi" },
{ role: "assistant", content: "Hello" },
]);
});
});