From 49e4c7e5c71a28aa61af792286dd5d1c6e748d0f Mon Sep 17 00:00:00 2001 From: ContextVM Date: Sat, 4 Jul 2026 12:20:32 +0200 Subject: [PATCH] feat(server): add model policy, admin tools, and billing prep - Introduce model visibility policies via MUXLL_MODELS_ALLOW / DENY and config file - Add gated admin tools (admin.models.setPolicy, admin.balance.credit) controlled by MUXLL_ADMIN_PUBKEYS - Implement prepaid billing infrastructure: ledger DB, markup, min charge, option to enable via MUXLL_BILLING - Update ENV example and developer docs to reflect new capabilities and code organization --- .env.example | 39 ++ AGENTS.md | 32 +- README.md | 47 ++- design/billing.md | 458 ++++++++++++++++++++++ design/idea.md | 13 + packages/client/src/client.ts | 5 +- packages/core/src/index.ts | 108 +++++ packages/server/src/catalog.ts | 179 +++++++++ packages/server/src/index.ts | 35 +- packages/server/src/ledger.ts | 208 ++++++++++ packages/server/src/main.ts | 80 +++- packages/server/src/pricing.ts | 124 ++++++ packages/server/src/server.ts | 329 ++++++++++++++-- packages/server/src/wire.ts | 47 +++ packages/server/tests/billing.test.ts | 284 ++++++++++++++ packages/server/tests/catalog.test.ts | 210 ++++++++++ packages/server/tests/integration.test.ts | 129 ++++++ packages/server/tests/policy.test.ts | 215 ++++++++++ 18 files changed, 2490 insertions(+), 52 deletions(-) create mode 100644 design/billing.md create mode 100644 design/idea.md create mode 100644 packages/server/src/catalog.ts create mode 100644 packages/server/src/ledger.ts create mode 100644 packages/server/src/pricing.ts create mode 100644 packages/server/tests/billing.test.ts create mode 100644 packages/server/tests/catalog.test.ts create mode 100644 packages/server/tests/policy.test.ts diff --git a/.env.example b/.env.example index 2ea7548..8238f1a 100644 --- a/.env.example +++ b/.env.example @@ -13,6 +13,45 @@ # discover the server. Default: off (private server, clients must know the pubkey). # MUXLL_ANNOUNCED=0 +# --- Model visibility policy ------------------------------------------------ +# The server trims/scope the model catalog exposed via models.list and resolvable +# via chat.complete. Rules are case-insensitive substrings matched against the +# `provider/id` tag (e.g. "anthropic" matches all Anthropic models, "gpt" +# matches every model whose tag contains "gpt" across providers). +# +# These env vars SEED the policy only when no config file exists yet; once +# admin.models.setPolicy persists a change to MUXLL_CONFIG_FILE, that file is +# authoritative (delete it + restart to re-seed from env). +# MUXLL_MODELS_ALLOW=anthropic,gpt +# MUXLL_MODELS_DENY=*-vision-* +# +# Path to the JSON policy file mutated by admin.models.setPolicy. +# Default: ./muxll-config.json +# MUXLL_CONFIG_FILE=muxll-config.json + +# --- Admin tools ------------------------------------------------------------ +# Comma-separated client pubkeys allowed to call admin.* tools (gated by the +# authenticated caller pubkey the Nostr transport injects). Empty/unset = no +# admin surface is registered at all. Not mutable at runtime — restart to change. +# MUXLL_ADMIN_PUBKEYS= + +# --- Billing (Phase 1: prepaid USD balance, admin-funded) -------------------- +# Set MUXLL_BILLING=1 to enable per-caller billing: chat.complete caps output +# tokens to what the caller's balance can afford and debits the real cost after +# the turn. Disabled by default (chat is free, as before). Lightning top-up and +# the CEP-8 payment wrap arrive in a later phase; for now balances are funded +# via admin.balance.credit. +# MUXLL_BILLING=1 +# +# SQLite file backing the ledger (default ./muxll-ledger.sqlite). +# MUXLL_LEDGER_DB=muxll-ledger.sqlite +# +# Markup over upstream cost in percent (10 = +10%). Default 0 = passthrough. +# MUXLL_MARKUP_PCT=10 +# +# Floor charge in USD applied to every chargeable request (anti-spam). +# MUXLL_MIN_CHARGE_USD=0.0001 + # --- Upstream LLM provider API keys ------------------------------------------ # Consumed by pi-ai's built-in providers (see @earendil-works/pi-ai). Only the # keys for providers you want to serve need to be set. diff --git a/AGENTS.md b/AGENTS.md index b5f9746..9e2ec66 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -148,16 +148,26 @@ asked. ## Where things live / change guide -- **Add an RPC tool** → `registerTools()` in +- **Add an RPC tool** → `registerTools()` (inference tools) or + `registerAdminTools()` (admin tools, gated by caller pubkey) in `packages/server/src/server.ts`; add the name to `METHODS` in `packages/core/src/index.ts`; add an integration test. - **Change provider configuration** → `packages/server/src/main.ts`. Today it is `builtinModels()` (env-var resolved). To inject a custom provider for tests, build a `Models` via `createModels()` + `models.setProvider(...)` and pass it to `startServer({ models })`. -- **Change model resolution** → `resolveModel()` in - `packages/server/src/wire.ts` (parses a `provider/id` tag; a bare id falls - back to first-match across providers). +- **Change model resolution / visibility** → `packages/server/src/catalog.ts`. + `Catalog` layers a `ModelPolicySet` over a pi-ai `Models` and is the single + seam for both `models.list` (`list`/`visibleModels`) and `chat.complete` + (`resolve` — a policy-hidden model returns `undefined`, i.e. "Unknown model", + so it can't be probed). Rules and the `models.list` `search` param share one + matcher: case-insensitive substring on the `provider/id` tag. Tag parsing / + bare-id first-match still lives in `resolveModel()` (`wire.ts`), which + `Catalog.resolve` wraps. Policy is seeded in `main.ts` from + `MUXLL_CONFIG_FILE` (authoritative once it exists) or `MUXLL_MODELS_ALLOW`/ + `MUXLL_MODELS_DENY`; `admin.models.setPolicy` patches it and persists via the + `onPolicyMutate` hook. Per-client overrides key on the caller pubkey the + transport injects (`injectClientPubkey: true` in `startServer`). - **Change the OpenAI wire shape** → `@muxll/core`. The server consumes the shapes in `chatResult()`/`toUsage()`/`toFinishReason()` (wire.ts) and `toContext()` (wire.ts, async — fetches http image URLs) maps the @@ -198,7 +208,17 @@ Configuration is entirely via environment variables (see `.env.example`). handler throws, the MCP SDK returns a result with `isError: true` and the message in `content`. `MuxllClient` turns these into thrown `Error`s for callers; raw `Client.callTool` tests must inspect `result.isError`, not - `.rejects`. + `.rejects`. This includes the admin guard (`admin.*` from a non-admin → + `isError` "unauthorized") and policy-hidden models (`chat.complete` on a + disallowed model → `isError` "Unknown model", same shape as a genuinely + unknown one). +- **Caller identity comes from the transport, not `_meta`.** `startServer` sets + `injectClientPubkey: true`, so the Nostr transport overwrites + `extra._meta.clientPubkey` with the authenticated gift-wrap signer pubkey + (it can't be spoofed by the client). Per-client model policy and admin + gating both read it via the `clientPubkey(extra)` helper in `server.ts`. + `MUXLL_ADMIN_PUBKEYS` is env-only and immutable at runtime (bootstrap, not + self-administered) — restart to change it. - **Streaming requires both ends to opt in.** The server sets `openStream: { enabled: true }`; the client must too, and must send a `progressToken`, or `extra._meta.stream` is absent and `getOpenStreamWriter()` @@ -257,4 +277,4 @@ Configuration is entirely via environment variables (see `.env.example`). becomes `{type:"tool",name}`; google can't pin a specific function). Still deferred via the `onPayload` seam: `response_format` and sampling params (`top_p`/`stop`/`seed`/penalties). -- Pricing/quotas are explicitly out of scope for the POC. +- Billing (Phase 1) is opt-in via `MUXLL_BILLING=1`: `Ledger` (`bun:sqlite`, USD stored as integer nano-dollars) keys balances on the authenticated caller pubkey; `chat.complete` caps output tokens to what the balance affords (`computeOutputCap`, in `pricing.ts`) and debits the real cost (recomputed via pi-ai's `calculateCost`, since the faux provider reports `cost.total = 0`) × `(1+MUXLL_MARKUP_PCT/100)`. `balance.*` / `admin.balance.*` tools fund/query balances. Native-Lightning top-up and the CEP-8 `withServerPayments` wrap are the next phases; see `docs/billing.md`. Billing is disabled entirely when no `ledger` is passed to `startServer`, so existing callers are unaffected. diff --git a/README.md b/README.md index c721222..2c16286 100644 --- a/README.md +++ b/README.md @@ -7,8 +7,9 @@ Nostr-native client can call them. Streamed responses ride CEP-41 open-ended streams. This repository currently holds a **proof of concept**: chat completions (with -and without streaming) and model listing. Payments, quotas, and provider config -files are intentionally out of scope for now. +and without streaming) and model listing. Billing is in progress (Phase 1: +a prepaid USD balance ledger funded by admin grants, with a markup over +upstream cost); see [`docs/billing.md`](docs/billing.md) for the design. ## How it works @@ -27,12 +28,15 @@ Two MCP tools are exposed (called via the standard `tools/call` method): #### `models.list` -Returns every model across configured providers. Shape follows OpenAI -`GET /models` (plus router-useful capability fields). +Returns the models the caller may use, across configured providers, after the +server's [model policy](#model-policy--admin-tools) is applied. Shape follows +OpenAI `GET /models` (plus router-useful capability fields). Optional arguments +narrow the policy-visible set; `{}` returns all of it. ```jsonc // request arguments -{} +{ "search": "gpt", // optional: case-insensitive substring on `provider/id`; absent = all visible + "limit": 20 } // optional: cap on models returned; absent = no limit // structuredContent { "object": "list", "data": [ @@ -45,6 +49,37 @@ Returns every model across configured providers. Shape follows OpenAI ] } ``` +#### Model policy & admin tools + +Operators control which models are exposed (to trim large upstream catalogs +and to scope per client). Rules are case-insensitive substrings matched +against the `provider/id` tag: `"anthropic"` matches every Anthropic model, +`"gpt"` matches every model whose tag contains `gpt` across providers. +Precedence: `deny` wins; when `allow` is present only matching tags pass. + +The policy is `{ default: { allow?, deny? }, byPubkey?: { : { allow?, deny? } } }` +— a per-client override is patch-merged over `default`. A hidden model neither +lists nor resolves: `chat.complete` reports it as `Unknown model`, so callers +can't probe for disallowed models. + +On startup the policy is seeded from `MUXLL_MODELS_ALLOW`/`MUXLL_MODELS_DENY` +(or a `MUXLL_CONFIG_FILE` JSON file once one exists). Two admin tools — gated +to the caller pubkeys in `MUXLL_ADMIN_PUBKEYS` (authenticated by the Nostr +transport, not self-reported) — manage it at runtime: + +```jsonc +// admin.models.getPolicy → { default, byPubkey? } of the current policy +{} + +// admin.models.setPolicy → patch, returns the new full policy. +// Per field: an array replaces, null clears (field/entry removed), absent unchanged. +{ "default": { "deny": ["*-vision-*"] }, + "byPubkey": { "": { "allow": ["gpt"] } } } +``` + +`admin.models.setPolicy` persists to `MUXLL_CONFIG_FILE`; once that file exists +it is authoritative (delete it + restart to re-seed from env). + #### `chat.complete` ```jsonc @@ -193,7 +228,7 @@ Proof of concept. Deliberately deferred until the core is validated: - **Stream cancellation** — a client CEP-41 `abort` does not yet cancel the upstream provider stream. - **Structured stream chunks** — `text_delta`, `thinking_delta`, and tool-call deltas are all streamed as `chat.completion.chunk` deltas (`delta.content`, `delta.reasoning_content`, `delta.tool_calls`). - **OpenAI feature gaps** — structured outputs (`response_format`) and sampling params (`top_p`, `stop`, `seed`, penalties) are supported per-API by pi-ai but not yet surfaced (need the `onPayload` seam). Function/tool calling is fully wired (`tools`, `tool_choice` normalized per target API, `tool` role); multimodal image input is wired (`user` content accepts `text` + `image_url` parts, data-URL or http(s)). -- **Pricing & quotas** — out of scope per the design brief. +- **Pricing & quotas** — Phase 1 landed: a prepaid USD balance ledger (`bun:sqlite`), a `maxTokens` cap that bounds each request to what the caller's balance affords, markup-over-upstream pricing, and `balance.*` / `admin.balance.*` tools (balances funded via `admin.balance.credit`). Disabled unless `MUXLL_BILLING=1`. Still to come: native-Lightning top-up (CEP-8 `balance.topup`), fixed per-Mtok pricing overrides, and the CEP-8 payment wrap. See [`docs/billing.md`](docs/billing.md). See [AGENTS.md](AGENTS.md) for contributor/agent instructions, and [`docs/idea.md`](docs/idea.md) for the original design note. diff --git a/design/billing.md b/design/billing.md new file mode 100644 index 0000000..c341430 --- /dev/null +++ b/design/billing.md @@ -0,0 +1,458 @@ +# Billing & Payments Design + +Status: **design** (not yet implemented). This document is the reference for the +next iteration. It supersedes the "Pricing & quotas — out of scope" line in +`README.md` and the matching deferral in `docs/idea.md`; it does **not** change +the inference-adapter principle in `VISION.md` (see +[Architectural tension](#architectural-tension) below). + +Companion material to re-read before implementing: + +- CEP-8 payments: `docs/contextvm-docs/src/content/docs/reference/ts-sdk/payments/overview.md`, + `…/how-to/payments/{getting-started,server,client,explicit-gating}.md` +- Routstr pricing: `docs/routstr-core/docs/provider/{pricing,advanced-pricing}.md`, + `docs/routstr-core/docs/client/payments.md` +- Routstr price/FX source: `docs/routstr-core/routstr/payment/price.py`, + `…/cost_calculation.py` +- `bun:sqlite`: `docs/bun-sqlite.md` + +## Goal + +Monetize inference served through muxll: clients **fund a balance with Lightning** +and spend it against per-request token cost. Operators set pricing policy +(markup over upstream, or fixed per-token rates), per model and per client. + +## Non-goals (deliberately deferred) + +- **Refunds / withdrawals.** Balance is one-way: Lightning in, USD credit, + consumed by usage. No redeem-to-sats, no Cashu minting, no CEP-8 `change` tag. +- **Subscriptions, quotas, free tiers** as first-class concepts. (A prepaid + balance + a 0-cost price rule can stand in for "free tier" if needed.) +- **Multi-currency.** USD only. +- **Per-request Lightning settlement for `chat.complete`.** Top-up is Lightning; + inference spends the prepaid balance (CEP-8 `waive`). See + [Why not per-request Lightning](#why-not-per-request-lightning-for-inference). + +## Decisions locked in this design pass + +1. **Balance is denominated in USD**, so users do not suffer BTC volatility on + held credit. Lightning only touches the **top-up edge**. +2. **Top-ups are native Lightning** (BOLT11 over NWC), self-serve. No + admin-grant-only mode is required (though `admin.balance.*` is still useful + for operator vouchers/credits). +3. **Over-serve protection uses `maxTokens`.** The server caps output tokens to + what the caller's remaining balance can afford, computed from a pre-call + input-token estimate. This bounds per-request spend without interrupting + streams. +4. **Pricing is configurable two ways, per model and per pubkey:** a percentage + markup over the upstream cost pi-ai already computes, **or** fixed USD rates + per 1M input/output tokens. +5. **The ledger is `bun:sqlite`**, integer fixed-point (no float money). +6. **The CEP-8 payment layer gates; the application settles.** `resolvePrice` + does authorization (waive/reject); the `chat.complete` handler does the real + accounting after generation. + +## How the references monetize + +| | routstr | yalr | CEP-8 native | +|---|---|---|---| +| Ledger unit | msats (sats) | sats | sats (BOLT11/NWC) | +| Funding | Lightning invoice *or* Cashu → balance/key | Lightning top-up (auto-invoice) | per-request invoice | +| Charging | balance debited per request from token usage | balance debited per request | invoice settled before response | +| Pricing | per-model > global per-token > upstream×markup; + min charge | per-request / per-token, admin-set | `resolvePrice` quote | +| FX | Kraken/Coinbase/Binance, `min`, 120s refresh | `ppq.ai` hosted top-up provider | — | + +The decisive pattern, shared by routstr and yalr: **both run their own balance +ledger on top of Lightning.** They fund a balance once, then decrement per +request — they do *not* settle a Lightning invoice per inference call. That is +exactly this design. Per-request invoices add a round-trip and a wallet +requirement to every call, which is hostile to streaming chat. + +### Why not per-request Lightning for inference + +CEP-8 fully supports it (`quotePrice` from `resolvePrice`, client pays via NWC, +`transparent` lifecycle). It would need **no ledger at all**. We reject it for +muxll because: (a) it adds an invoice round-trip to every request; (b) it forces +every client to run an NWC wallet handler; (c) it cannot represent prepaid +credit, operator vouchers, or rate limits. The balance model exists precisely +to avoid those. (Revisit if a pure pay-per-use, no-accounts mode is ever +wanted — it's a configuration of the same `withServerPayments` wrap, not a +rewrite.) + +## The core mechanism: gate vs settle + +**The timing tension.** CEP-8 settles a priced request *before* it is forwarded +to the MCP handler. LLM cost depends on output tokens, which are unknown until +generation completes. Therefore `resolvePrice` can never charge the exact +amount — it can only authorize. + +Resolution: split the two concerns. + +- **Gate — `resolvePrice` (runs before the handler).** Pure authorization. + - `chat.complete`: balance > floor → `waivePrice()` (no invoice emitted); else + `rejectPrice("insufficient balance")`. + - `balance.topup`: → `quotePrice(sats)` (this *is* the Lightning charge). +- **Settle — inside the `chat.complete` handler (runs after generation).** With + real `usage` known, compute the exact charge and **debit the ledger + atomically**. No-charge (or input-only charge) on upstream failure before any + tokens are produced. + +This is the design's central decision. CEP-8's payment layer authorizes; the +application accounts. The CEP-8 docs explicitly anticipate this ("if you enforce +quotas/one-time use, use a durable store"). + +## Over-serve protection: the `maxTokens` cap + +Confirmed in pi-ai: + +- `StreamOptions.maxTokens` caps **output** tokens (standard + `max_tokens` / `max_completion_tokens`). +- `estimateContextTokens(context)` (`utils/estimate.d.ts`) estimates **input** + tokens **before** the call. +- `clampMaxTokensToContext(model, context, maxTokens)` keeps it inside the + context window. +- `Model.cost` is **USD per 1M tokens** — `calculateCost` does + `(model.cost.input / 1000000) * usage.input`. + +Computation (per request, in the handler before `models.complete/stream`): + +``` +priceIn = priceBook.inputPerToken(pubkey, model) // resolves policy → USD/token +priceOut = priceBook.outputPerToken(pubkey, model) +inEst = estimateContextTokens(context) * INPUT_ESTIMATE_FUDGE // fudge ≈ 1.05 +inCostEst = inEst * priceIn +budget = ledger.balance(pubkey) - inCostEst +if budget < priceBook.minCharge(pubkey, model): + reject // can't afford even input +maxOut = floor(budget / priceOut) +maxTokens = clamp(min(userMax ?? model.maxTokens, maxOut), 1, model.contextWindow) +``` + +The provider then **stops generating at the ceiling** — no stream ever has to be +killed mid-flight for balance reasons. At settle, debit the **actuals** from +`usage.input` / `usage.output` / cache fields. + +**Why this matters.** A single uncapped stream could emit ~100k tokens and drain +the whole balance plus a large overdraft before the handler runs the debit. The +cap converts that unbounded risk into a bounded per-request allowance. It is the +single mechanism that makes prepaid streaming safe enough to defer the harder +concurrency work (see [Concurrency & safety](#concurrency--safety)). + +Caveats (mark, don't fix in v1): + +- **Input estimate accuracy.** We charge actuals, so the only leak is + `actual_input − estimated_input`. Bias the estimate *high* (`INPUT_ESTIMATE_FUDGE` + ≈ 1.05): overestimating → user capped a touch early (no money lost); + underestimating → small overdraft. Lean toward the operator. +- **Reasoning models.** `maxTokens` caps reasoning + visible output together + (reasoning tokens are billed as output). A tight cap on a reasoning model may + spend the whole budget "thinking" and emit little. UX wrinkle, not a money + risk. +- **Tool-call truncation.** A cap landing mid-tool-call can truncate JSON + `arguments` → a broken tool call. Set a sane floor (e.g. ≥ 256) when `tools` + are present. + +## Pricing model + +Two modes, resolving to a `(inputPerToken, outputPerToken)` pair for a given +`(pubkey, model)`: + +- **markup** (default): `charge = usage.cost.total * (1 + markupPct/100)`. + `usage.cost.total` is already populated by pi-ai's `calculateCost` and already + surfaced by `toUsage()` (`packages/server/src/wire.ts`). The markup is one + multiplication. No price table to maintain; always tracks upstream changes. + Correctly marks up cache read/write (Anthropic charges 2× for 1h writes — + folded into `usage.cost`) and reasoning tokens. +- **fixed**: `charge = (usage.input * inPerMtok + usage.output * outPerMtok) / 1e6`. + Computed from token counts. Operator sets explicit USD/Mtok rates. **Gap:** + fixed mode prices only raw input/output and silently eats cache/reasoning + cost. Acceptable as the "simple proxy" mode; document it. (Upgrade path: + add explicit cache rates to the fixed shape.) + +A **minimum charge per request** (USD floor) prevents dust/spam on tiny calls +(routstr's `min_request_cost`). + +Precedence (routstr's order, and the sensible one): **per-pubkey override > +per-model override > global default.** This reuses the `Catalog` mental model +operators already know (`byPubkey` / `default`), extended with a `byModel` +layer. + +Proposed config shape (persisted like model policy today; see +[Persistence](#persistence)): + +```jsonc +{ + "default": { + "mode": "markup", + "markupPct": 10, + "minCharge": 0.0001 + }, + "byModel": { + "anthropic/claude-opus-4-5": { + "mode": "fixed", + "fixed": { "inputPerMtok": 20, "outputPerMtok": 80 } + } + }, + "byPubkey": { + "": { "mode": "fixed", "fixed": { "inputPerMtok": 1, "outputPerMtok": 3 } } + } +} +``` + +`mode: "markup"` ignores `fixed`; `mode: "fixed"` ignores `markupPct`. +`inputPerToken = inputPerMtok / 1e6` (and likewise for output) — the canonical +internal unit the cap and debit consume. + +## Currency & FX + +The ledger is USD. Lightning settles sats. Because the balance is **non-refundable +and USD-denominated**, FX risk is a **single-edge, seconds-long** concern: quote +invoice → user pays. There is **no FX during usage** (unlike routstr, which prices +and charges in sats and so needs a live feed on every request). + +- One helper, `usdPerSat(): number`. Fetch on demand, cached ~60s. Endpoints: + Kraken → Coinbase → Binance, take `min` (favors operator; this is what + routstr does in `payment/price.py`). Single fallback is fine for v1. +- Apply a small **exchange buffer** (≈ 0.5%, routstr's `exchangeFee`) when + converting a requested USD amount → sat invoice, to absorb normal BTC drift in + the payment window. The operator's worst case is sub-minute FX drift on the + invoice amount — bounded and negligible. +- No background ticker needed for v1. Upgrade to routstr's 120s + jitter + background refresh only if top-up volume justifies it. + +## Top-up flow: a CEP-8 priced capability + +The cleanest expression: **`balance.topup` is itself a CEP-8 priced call**, and +`chat.complete` is a waived one. One `withServerPayments` wrap + one +`LnBolt11NwcPaymentProcessor` serves both; the `resolvePrice` outcome differs. + +| Tool | `resolvePrice` returns | Handler does | +|---|---|---| +| `chat.complete` | `waive` / `reject` (balance check) | cap `maxTokens`, run, debit actuals | +| `balance.topup` | `quote(amount = sats for requested USD)` | credit USD balance (payment already verified) | + +CEP-8 guarantees the crediting handler runs **only after** payment is verified — +which is exactly the security property a top-up needs. No out-of-band +invoice-and-poll, no separate verification endpoint. The client side is +symmetric: `withClientPayments` + `LnBolt11NwcPaymentHandler` pays top-ups +automatically and waives chat calls through. + +Top-up amount model (user-facing currency is USD): + +1. Client calls `balance.topup({ amountUsd })`. +2. `resolvePrice` computes `sats = ceil(amountUsd / usdPerSat() * (1 + buffer))` + and returns `quotePrice(sats, { description: "Top up $X" })`. +3. CEP-8 issues the BOLT11; client's NWC handler pays; server verifies. +4. Handler credits `amountUsd` (the requested USD — predictable for the user; + the buffer covers the operator's FX drift). + +`balance.get` returns the caller's own balance + recent usage. `admin.balance.*` +lets an operator grant/inspect balances (voucher/credit path; still useful even +with self-serve Lightning). + +## Architecture & seams + +All new wiring is **DI in `startServer`**, mirroring how `Catalog`/`policy` are +injected today. New components: + +- **`Ledger`** (`packages/server/src/ledger.ts`) — `bun:sqlite`, integer + fixed-point. Methods: `balance(pk)`, `charge(pk, model, usage) → chargeUsd`, + `credit(pk, usd, source)`, `history(pk, limit)`. See + [Data model](#data-model). +- **`PriceBook`** (`packages/server/src/pricing.ts`) — resolves + `(pubkey, modelTag) → { inputPerToken, outputPerToken, minCharge }` from the + config (markup or fixed). Reuses `Catalog`'s `matches()` + `byPubkey`/`default` + merge; adds `byModel`. Default mode is markup, so it reads `usage.cost.total` + rather than maintaining a price table. +- **`Fx`** (`packages/server/src/fx.ts`) — `usdPerSat()` with TTL cache and the + Kraken/Coinbase/Binance fallback. Only used by the top-up path. +- **`payments.ts`** (`packages/server/src/payments.ts`) — builds the + `resolvePrice` callback and the `withServerPayments` options; routes by + capability (`chat.complete` → waive/reject, `balance.topup` → quote). + +Wiring into existing code: + +- **`startServer`** (`packages/server/src/server.ts`): build the bare + `NostrServerTransport` as today, then + `transport = withServerPayments(transport, { processors: [lnBolt11], pricedCapabilities, resolvePrice })`. + `pricedCapabilities` marks `chat.complete` and `balance.topup` (for discovery / + `cap` tags). `Ledger`, `PriceBook`, `Fx` are new `StartServerOptions` fields. +- **`resolvePrice`**: one-line gate. For `chat.complete`, read + `ledger.balance(clientPubkey)` and return `waivePrice()` / `rejectPrice()`. + Keep CEP-8 coupling minimal — the real accounting is in the handler. +- **`runComplete` / `runStreamed`** (`packages/server/src/server.ts`): before + `models.complete/stream`, compute the `maxTokens` cap from + `ledger.balance(pk)` + `estimateContextTokens` + `PriceBook`; after generation, + `ledger.charge(pk, model, usage)`. Both already produce `usage` via `toUsage`. +- **`registerTools`**: add `balance.get` and `balance.topup` (and + `admin.balance.*` via `registerAdminTools`, admin-gated like the existing + `admin.models.*`). +- **`@muxll/core`** (`packages/core/src/index.ts`): add the new `METHODS` + (`balance.get`, `balance.topup`, `admin.balance.credit`, `admin.balance.get`) + and their zod raw shapes, same convention as the existing tools. +- **`main.ts`**: seed `Ledger` (db path), `PriceBook` (config file/env), `Fx`, + and the `LnBolt11NwcPaymentProcessor` from `NWC_SERVER_CONNECTION`. Mirror the + existing `readInitialPolicy` / `onPolicyMutate` pattern for price config. + +### Persistence + +Both model policy and price config persist the same way today's policy does +(`MUXLL_CONFIG_FILE`, atomic temp+rename write via `onPolicyMutate`). The ledger +is its own SQLite file (`MUXLL_LEDGER_DB`, default `muxll-ledger.sqlite`) — +append-mostly, never rewritten by config mutations, so it stays separate from +the JSON config. + +## Data model + +`bun:sqlite`, synchronous, single-writer. USD stored as **integer nano-dollars +(1e-9 USD)** so the smallest realistic charge (~$0.0000045, the README's own +example) is `4500` units with no float drift; int64 headroom is enormous. + +```sql +CREATE TABLE balances ( + pubkey TEXT PRIMARY KEY, + credit_nusd INTEGER NOT NULL DEFAULT 0, -- nano-USD, ≥ 0 + updated INTEGER NOT NULL -- epoch ms +); + +CREATE TABLE usage ( -- one row per charge (history + audit) + id INTEGER PRIMARY KEY AUTOINCREMENT, + pubkey TEXT NOT NULL, + ts INTEGER NOT NULL, -- epoch ms + model TEXT NOT NULL, -- provider/id tag + in_tokens INTEGER NOT NULL, + out_tokens INTEGER NOT NULL, + cache_read INTEGER NOT NULL DEFAULT 0, + cache_write INTEGER NOT NULL DEFAULT 0, + cost_nusd INTEGER NOT NULL, -- upstream cost (pi-ai usage.cost) + charge_nusd INTEGER NOT NULL, -- what was actually debited + mode TEXT NOT NULL -- 'markup' | 'fixed' +); +CREATE INDEX usage_pubkey_ts ON usage (pubkey, ts DESC); + +CREATE TABLE credits ( -- top-ups / admin grants (audit) + id INTEGER PRIMARY KEY AUTOINCREMENT, + pubkey TEXT NOT NULL, + ts INTEGER NOT NULL, + amount_nusd INTEGER NOT NULL, + source TEXT NOT NULL, -- 'lightning' | 'admin' | … + ref TEXT -- invoice hash / admin pubkey / note +); +CREATE INDEX credits_pubkey_ts ON credits (pubkey, ts DESC); +``` + +Atomic debit (no async race — `bun:sqlite` is synchronous, single-writer): + +```ts +// charge = min(actualCharge, balance) → clamp at 0; reject next call once negative-ish +db.query( + `UPDATE balances SET credit_nusd = MAX(0, credit_nusd - ?), updated = ? + WHERE pubkey = ?`, +).run(chargeNusd, now, pubkey); +``` + +For the gate we want a stricter `≥ floor` check; for the settle we clamp at 0 to +absorb the bounded concurrency overdraft (see below). Record the `usage` row +regardless (audit), storing the *intended* charge even if clamped. + +## RPC surface (new tools) + +Same shape as the existing tools (zod raw input shapes in `@muxll/core`, +registered in `server.ts`): + +```jsonc +// balance.get → own balance + recent usage (caller only) +{} + +// balance.topup → triggers CEP-8 Lightning quote (resolvePrice → quotePrice) +{ "amountUsd": 5.0 } + +// admin.balance.get → any pubkey's balance + history (admin-gated) +{ "pubkey": "…" } +// admin.balance.credit → operator grant (voucher/credit; admin-gated) +{ "pubkey": "…", "amountUsd": 5.0, "note": "…" } +``` + +`balance.topup`'s result (after payment is verified and the handler runs) is the +new balance + the credited amount. `admin.balance.*` are gated by +`MUXLL_ADMIN_PUBKEYS` exactly like `admin.models.*` (throw `unauthorized` from +non-admin → `isError` result, per the existing pattern). + +## Concurrency & safety + +- **The debit itself is race-free.** Synchronous `bun:sqlite`, single writer, + atomic `UPDATE`. No async between "read balance" and "write balance" — the + `UPDATE … SET credit_nusd = credit_nusd - ?` is one statement. +- **The gate-vs-settle gap is not serialized.** N concurrent streams from one + pubkey each pass `resolvePrice`, each computes its own `maxTokens` from the + *same* balance snapshot, each generates up to its cap, each debits. Overdraft + is bounded to roughly `(N−1) × perRequestCap` — **bounded by the cap**, which + is the whole point of using `maxTokens`. For v1: clamp the debit at 0 and + reject the next call once balance hits the floor. The airtight fix is a + `holds` table that reserves an estimate at gate time and reconciles at settle; + defer it (`ponytail:` — the cap is what makes this deferral safe). +- **Top-up crediting is safe by construction.** CEP-8 verifies payment before + the `balance.topup` handler runs; crediting is a single `credit(pk, …)` call. +- **No refund path** means no double-spend-via-withdrawal risk and no + change-token minting — one fewer surface to get wrong. + +## Testing + +The harness already proves DI for this (`@muxll/test-utils`): +`standUpFauxClient()` + `MockRelayHub`. pi-ai's `fauxProvider` yields +**deterministic token counts**, so: + +- **Cap + debit (Phase 1, no Lightning):** drive `chat.complete` through a faux + server with a `Ledger` over a temp SQLite file; assert the emitted `maxTokens` + matches the balance-derived ceiling and the post-call `usage` row holds the + expected charge. Cover: sufficient balance, balance-can-only-afford-input + (reject), balance-runs-out-mid-budget (cap < model default), markup vs fixed + pricing, min-charge floor. +- **Top-up flow:** use the SDK's `FakePaymentProcessor` / + `FakePaymentHandler` (confirmed exported from `@contextvm/sdk/payments`) so + `balance.topup` settles without a real NWC wallet; assert the USD credit lands + and `usage` history is queryable. `Fx` tested with a stubbed rate. +- **Client side:** `withClientPayments` + `FakePaymentHandler` in the faux + harness; assert chat calls waive through and top-ups pay. + +## Phasing + +1. **Accounting core (no Lightning).** `Ledger` + `PriceBook` (markup-only) + + the `maxTokens` cap + debit wired into `runComplete`/`runStreamed`, gated by + a balance check in `resolvePrice`. Admin grant via `admin.balance.credit` to + fund test balances. Tested over faux. **This alone is a demoable product.** +2. **Pricing overrides.** Fixed-per-Mtok mode + `byModel` / `byPubkey` + precedence in `PriceBook`; min-charge floor. Persisted config. +3. **Self-serve Lightning top-up.** `balance.topup` as a CEP-8 priced call; + `LnBolt11NwcPaymentProcessor` from `NWC_SERVER_CONNECTION`; `Fx.usdPerSat()` + with the exchange buffer. Client `withClientPayments` + NWC handler. +4. **Reservation/holds** (only if concurrency overdraft bites in practice): a + `holds` table, reserve-at-gate, reconcile-at-settle. + +Phase 1 is small: one `Ledger` class, one `PriceBook`, a `resolvePrice` gate, +two debit call-sites, one admin RPC, and tests over the existing harness. + +## Architectural tension + +`VISION.md` is explicit: *"Muxll is a thin inference adapter. Payments +settlement lives in its own component."* A ledger inside `packages/server` +bends that principle. The mitigation: keep the new surface behind clean, +DI'd interfaces (`Ledger`, `PriceBook`, `Fx`, `resolvePrice`) so the extraction +later is a **move**, not a rewrite. The long-term shape — a separate CEP-8 +payment gateway process wrapping a muxll inference server — is exactly what +`withServerPayments`' transport-middleware design is built for. For the POC, +in-server is the shortest path and CEP-8 explicitly supports the prepaid-waive +pattern; mark it `ponytail:` and plan the extraction. + +## Open questions + +- Ledger integer scale — nano-USD (1e-9) recommended; confirm before + implementation (any scale ≥ 1e-8 works; the README's $0.0000045 example is the + lower bound to keep integer). +- Should `balance.get` / usage history be paginated from day 1, or capped at + "recent N"? (Default: recent N; paginate when it matters.) +- Do we need an operator dashboard / metrics, or are `admin.balance.*` RPCs + enough for v1? (RPCs enough; dashboard is a separate surface.) +- Fallback FX behavior if **all** exchanges are unreachable at top-up time: hard + fail the top-up (recommended — never credit without a rate) vs. last-good-rate + with a max staleness window. diff --git a/design/idea.md b/design/idea.md new file mode 100644 index 0000000..474be75 --- /dev/null +++ b/design/idea.md @@ -0,0 +1,13 @@ +We want to build a CVM-based LLM router. This means that models served through a JSON-RPC API interface. The goal is to create a CVM server that can plug in providers and serve them through the CVM JSON-RPC interface over Nostr. Indeed, we thought that we can leverage the pi framework to achieve this. They provide a pretty nice library to plug in to clients, stream, and other features. + +We don’t need a complete API from the get go—just the core endpoints and models—so users can retrieve the list of available models. Additionally, CVM recently introduced CEP-41 open-ended streams, which align well with streamed LLM responses. + +Other references include: +- **Routstr**, which follows the same concept we want to achieve, but it is exposed as http server. +- **Yalr**, a general-purpose LLM router written in Rust that can help us address some of the nuances in building this project. + +For now, we don’t need to worry about pricing or charging for usage since we first want to validate this proof of concept for development purposes. Once everything is in place, we can explore paid options. For now, let’s focus solely on the router and proxy so we can serve LLM inference through our server. + +The goal is for operators to plug in APIs from providers—for example—and serve them through our CVM server. + +We will implement this in TypeScript, using the CVM SDK and the pi SDK. Codebased and docs are available at the `docs` directory. We also have an example of a server implementing streaming api at `cordn` directory. Also all cvm docs are available as well. \ No newline at end of file diff --git a/packages/client/src/client.ts b/packages/client/src/client.ts index 02d9b4a..1370c6c 100644 --- a/packages/client/src/client.ts +++ b/packages/client/src/client.ts @@ -27,6 +27,7 @@ import { type ChatCompletionChunk, type ChatInput, type ListModelsResponse, + type ModelsListInput, } from "@muxll/core"; export interface MuxllClientOptions { @@ -125,11 +126,11 @@ export class MuxllClient { } /** List models from the provider → parsed OpenAI list response. */ - async listModels(): Promise { + async listModels(params?: ModelsListInput): Promise { const result = await this.client.callTool( { name: METHODS.modelsList, - arguments: {}, + arguments: params ?? {}, }, undefined, { diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index a93de17..0e4b1a1 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -14,6 +14,11 @@ import { z } from "zod"; export const METHODS = { chatComplete: "chat.complete", modelsList: "models.list", + adminModelsGetPolicy: "admin.models.getPolicy", + adminModelsSetPolicy: "admin.models.setPolicy", + balanceGet: "balance.get", + adminBalanceGet: "admin.balance.get", + adminBalanceCredit: "admin.balance.credit", } as const; // --- /chat/completions input shapes ----------------------------------------- @@ -72,6 +77,28 @@ export const toolChoiceSchema = z.union([ }), ]); +// --- models.list input ----------------------------------------------------- + +// Raw zod shape. Both fields optional: `{}` reproduces today's unfiltered +// listing (still subject to the server's model policy). `search` narrows the +// policy-visible set; `limit` slices the first N in catalog order. +export const modelsListInput = { + search: z + .string() + .optional() + .describe( + "Case-insensitive substring matched against 'provider/id'; absent = all visible", + ), + limit: z + .number() + .int() + .positive() + .optional() + .describe("Cap on models returned; absent = no limit"), +}; + +// --- /chat/completions input shapes ----------------------------------------- + // Raw zod shape (not z.object) — McpServer.registerTool takes raw shapes. export const chatInput = { model: z @@ -98,6 +125,12 @@ export const chatInput = { "Tools the model may call; surfaced back as choices[0].message.tool_calls", ), tool_choice: toolChoiceSchema.optional(), + reasoning_effort: z + .enum(["minimal", "low", "medium", "high", "xhigh"]) + .optional() + .describe( + "OpenAI reasoning_effort: activates thinking on reasoning-capable models (ignored otherwise)", + ), temperature: z.number().optional(), max_tokens: z.number().optional().describe("Upper bound on generated tokens"), }; @@ -216,6 +249,68 @@ export const modelsOutput = { export const modelsListSchema = z.object(modelsOutput); +// --- model policy (admin.models.*) ----------------------------------------- + +// A single rule set. `allow`/`deny` are partial needles: case-insensitive +// substring matched against the `provider/id` tag. An absent field = no rule; +// an explicit [] = the empty set (allow:[] blocks everything, deny:[] denies +// nothing). Precedence: deny wins; when `allow` is present only matching tags +// pass, then `deny` removes. +export const modelPolicySchema = z.object({ + allow: z.array(z.string()).optional(), + deny: z.array(z.string()).optional(), +}); + +// The full policy: a default plus optional per-client overrides. A client's +// effective policy is the override patch-merged over `default` (override fields +// that are absent inherit from default). +export const modelPolicySetSchema = z.object({ + default: modelPolicySchema, + byPubkey: z.record(z.string(), modelPolicySchema).optional(), +}); + +// Patch body for admin.models.setPolicy. Per field: an array replaces, `null` +// clears (field falls back / entry removed), absent leaves it unchanged. +export const modelPolicyPatchSchema = z.object({ + allow: z.array(z.string()).nullable().optional(), + deny: z.array(z.string()).nullable().optional(), +}); + +// Raw zod shape (inputSchema for admin.models.setPolicy). +export const modelPolicySetPatchInput = { + default: modelPolicyPatchSchema.optional(), + byPubkey: z + .record(z.string(), z.union([modelPolicyPatchSchema, z.null()])) + .optional(), +}; + +// Raw output shape for admin.models.{getPolicy,setPolicy}: the resulting set. +export const modelPolicySetOutput = { + default: modelPolicySchema, + byPubkey: z.record(z.string(), modelPolicySchema).optional(), +}; + +// --- billing (balance.*, admin.balance.*) --------------------------------- +// Phase 1: prepaid USD balance keyed by the authenticated caller pubkey, +// funded via admin grants (Lightning top-up arrives in Phase 3). Pricing is +// markup-only here (see @muxll/server PriceBook); fixed-rate support is Phase 2. + +// balance.get: no arguments; reads the caller's own balance + recent usage. +export const balanceGetInput = {}; + +// admin.balance.get: read any caller's balance + recent usage (admin-only). +export const adminBalanceGetInput = { + pubkey: z.string(), + limit: z.number().int().positive().max(1000).optional(), +}; + +// admin.balance.credit: grant USD credit to a pubkey (admin-only). +export const adminBalanceCreditInput = { + pubkey: z.string(), + amountUsd: z.number().positive(), + note: z.string().optional(), +}; + // --- inferred TS types ------------------------------------------------------ export type FinishReason = z.infer; @@ -227,6 +322,19 @@ export type ToolChoice = z.infer; // z.input (not z.infer/z.output): callers build requests, so .default() fields // like `stream` stay optional on input and are filled by the server. export type ChatInput = z.input>; +export type ModelsListInput = z.input>; +export type ModelPolicy = z.infer; +export type ModelPolicySet = z.infer; +export type ModelPolicyPatch = z.input; +export type ModelPolicySetPatch = z.input< + z.ZodObject +>; export type ChatCompletion = z.infer; export type ChatCompletionChunk = z.infer; export type ListModelsResponse = z.infer; +export type AdminBalanceGetInput = z.input< + z.ZodObject +>; +export type AdminBalanceCreditInput = z.input< + z.ZodObject +>; diff --git a/packages/server/src/catalog.ts b/packages/server/src/catalog.ts new file mode 100644 index 0000000..236e48d --- /dev/null +++ b/packages/server/src/catalog.ts @@ -0,0 +1,179 @@ +/** + * Server-side model catalog: a visibility policy layered over a pi-ai `Models`. + * + * One seam for both `models.list` (what's visible) and `chat.complete` (what + * resolves), so a hidden model neither lists nor resolves — callers learn + * nothing of its existence. Policy rules and the `models.list` `search` param + * share one matcher: case-insensitive substring on the `provider/id` tag. + * + * Transport-agnostic. `clientPubkey` is threaded through so a per-client + * override can be applied; the transport injects it at `_meta.clientPubkey` + * (authenticated — derived from the Nostr gift-wrap signer, not self-reported). + */ +import type { Api, Model, Models } from "@earendil-works/pi-ai"; +import type { + ModelPolicy, + ModelPolicySet, + ModelPolicySetPatch, +} from "@muxll/core"; +import { resolveModel } from "./wire.ts"; + +/** `provider/id` — the tag clients send back as `model` and policies match on. */ +export function modelTag(m: Model): string { + return `${m.provider}/${m.id}`; +} + +/** + * Case-insensitive substring match on a tag. Empty/whitespace needle never + * matches (so an empty allow-list blocks all, and an empty search is a no-op). + */ +export function matches(tag: string, needle: string): boolean { + const n = needle.trim().toLowerCase(); + return n !== "" && tag.toLowerCase().includes(n); +} + +/** + * Does a tag pass one policy? `deny` wins; a present `allow` (even `[]`) + * restricts to matching tags; absent `allow` = no restriction. + */ +export function policyAllows(tag: string, policy: ModelPolicy): boolean { + if ((policy.deny ?? []).some((n) => matches(tag, n))) return false; + if (Array.isArray(policy.allow)) { + return policy.allow.some((n) => matches(tag, n)); + } + return true; +} + +/** + * Merge `over` onto `base`: a defined field (array, incl. `[]`) replaces; an + * absent field inherits from `base`. (`null` never reaches here — `applyPolicyPatch` + * turns clear-into-`undefined` first.) This is the per-client override model: + * the effective policy is `mergePolicy(default, byPubkey[pk] ?? {})`. + */ +export function mergePolicy(base: ModelPolicy, over: ModelPolicy): ModelPolicy { + return { + ...(base.allow !== undefined ? { allow: base.allow } : {}), + ...(base.deny !== undefined ? { deny: base.deny } : {}), + ...(over.allow !== undefined ? { allow: over.allow } : {}), + ...(over.deny !== undefined ? { deny: over.deny } : {}), + }; +} + +function patchPolicy( + base: ModelPolicy, + patch: ModelPolicySetPatch["default"], +): ModelPolicy { + const out: ModelPolicy = {}; + // Start from base (dropping undefined fields along the way). + if (base.allow !== undefined) out.allow = base.allow; + if (base.deny !== undefined) out.deny = base.deny; + if (patch?.allow !== undefined) { + // null clears (→ absent → inherits upward / no rule); array replaces. + if (patch.allow !== null) out.allow = [...patch.allow]; + else delete out.allow; + } + if (patch?.deny !== undefined) { + if (patch.deny !== null) out.deny = [...patch.deny]; + else delete out.deny; + } + return out; +} + +/** + * Apply a set patch → new set (immutable). `null` under `byPubkey[pk]` removes + * the entry; a patch object merges onto the existing per-client policy. + */ +export function applyPolicyPatch( + set: ModelPolicySet, + patch: ModelPolicySetPatch, +): ModelPolicySet { + const byPubkey: Record = set.byPubkey + ? { ...set.byPubkey } + : {}; + if (patch.byPubkey) { + for (const [pk, p] of Object.entries(patch.byPubkey)) { + if (p === null) { + delete byPubkey[pk]; + } else { + byPubkey[pk] = patchPolicy(byPubkey[pk] ?? {}, p); + } + } + } + const next: ModelPolicySet = { + default: patchPolicy(set.default, patch.default), + }; + if (Object.keys(byPubkey).length > 0) next.byPubkey = byPubkey; + return next; +} + +function effectivePolicy( + set: ModelPolicySet, + clientPubkey?: string, +): ModelPolicy { + const override = clientPubkey ? set.byPubkey?.[clientPubkey] : undefined; + return override ? mergePolicy(set.default, override) : set.default; +} + +/** + * Mutable model catalog over a pi-ai `Models`. `onMutate` (if given) is invoked + * with the new set after every `setPolicy` so `main.ts` can persist it; tests + * omit it for pure in-memory behavior. + */ +export class Catalog { + constructor( + private readonly models: Models, + private policy: ModelPolicySet, + private readonly onMutate?: (set: ModelPolicySet) => void, + ) {} + + /** Current policy (deep clone; safe to hand to callers / serialize). */ + getPolicy(): ModelPolicySet { + return structuredClone(this.policy); + } + + /** Apply a patch, notify `onMutate`, return the new set. */ + setPolicy(patch: ModelPolicySetPatch): ModelPolicySet { + this.policy = applyPolicyPatch(this.policy, patch); + this.onMutate?.(this.getPolicy()); + return this.getPolicy(); + } + + /** Models visible to this caller, in catalog order. */ + visibleModels(clientPubkey?: string): Model[] { + const policy = effectivePolicy(this.policy, clientPubkey); + return this.models + .getModels() + .filter((m) => policyAllows(modelTag(m), policy)); + } + + /** `models.list` view: visible set, narrowed by `search`, capped by `limit`. */ + list( + params: { search?: string; limit?: number }, + clientPubkey?: string, + ): Model[] { + let out = this.visibleModels(clientPubkey); + if (params.search) { + const needle = params.search.trim().toLowerCase(); + if (needle) + out = out.filter((m) => modelTag(m).toLowerCase().includes(needle)); + } + if (params.limit) out = out.slice(0, params.limit); + return out; + } + + /** + * Resolve a `provider/id` tag (bare id = first-match) for this caller. + * Returns undefined if unknown OR not visible — callers can't tell which, so + * hidden models don't leak through `chat.complete`. + */ + resolve(tag: string, clientPubkey?: string): Model | undefined { + const model = resolveModel(this.models, tag); + if (!model) return undefined; + return policyAllows( + modelTag(model), + effectivePolicy(this.policy, clientPubkey), + ) + ? model + : undefined; + } +} diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index a3ef19e..4225545 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -1,23 +1,50 @@ /** * @muxll/server — the CVM/Nostr inference adapter. Operators run this, point it - * at their upstream provider keys via pi-ai, and expose chat.complete / - * models.list over MCP-on-Nostr with CEP-41 streaming. + * at their upstream provider keys via pi-ai, and exposes chat.complete / + * models.list (plus admin.models.* policy tools) over MCP-on-Nostr with + * CEP-41 streaming. * * Public surface only. Internals are split: - * ./server.ts — MCP tool registration + CEP-41 transport glue - * ./wire.ts — OpenAI ↔ pi-ai translation (transport-agnostic) + * ./server.ts — MCP tool registration + CEP-41 transport glue + * ./catalog.ts — model visibility policy over a pi-ai Models + * ./wire.ts — OpenAI ↔ pi-ai translation (transport-agnostic) */ export { registerTools, + registerAdminTools, + registerBalanceTools, startServer, type StartServerOptions, type StartedServer, } from "./server.ts"; +export { + Catalog, + applyPolicyPatch, + mergePolicy, + policyAllows, + matches, + modelTag, +} from "./catalog.ts"; export { resolveModel, mapToolChoice, + mapReasoning, toImageContent, toFinishReason, toUsage, } from "./wire.ts"; +export { + Ledger, + type Balance, + type UsageRecord, + type CreditRecord, +} from "./ledger.ts"; +export { + PriceBook, + computeOutputCap, + type PricingConfig, + type TokenRates, + type CapInput, + type CapResult, +} from "./pricing.ts"; export type { Models } from "@earendil-works/pi-ai"; diff --git a/packages/server/src/ledger.ts b/packages/server/src/ledger.ts new file mode 100644 index 0000000..b0ff5e5 --- /dev/null +++ b/packages/server/src/ledger.ts @@ -0,0 +1,208 @@ +/** + * Prepaid balance ledger for billing. Keys on the authenticated caller pubkey + * (injected by the Nostr transport at `_meta.clientPubkey`). Stores USD as + * integer nano-dollars (1e-9 USD) to avoid float drift on the cheapest real + * charges (~$0.0000045); the public API is float USD so callers never touch the + * integer scale. + * + * Synchronous + single-writer by design (bun:sqlite): the atomic debit is one + * statement, no async race between read and write. Phase 1 funds balances via + * `admin.balance.credit`; Lightning top-up (Phase 3) calls `credit` too. + * + * `charge` debits `min(chargeUsd, balance)` (clamps at 0) but always records + * the full `chargeUsd` in the `usage` row for audit. The `maxTokens` cap in + * `pricing.ts` bounds each request so any clamp is the bounded input-estimate + * error, not a runaway stream. + */ +import { Database } from "bun:sqlite"; + +/** Nano-dollars per USD (internal integer storage scale). */ +export const NUSD_PER_USD = 1e9; + +const nusd = (usd: number): number => + Math.max(0, Math.round(usd * NUSD_PER_USD)); +const usd = (n: number): number => n / NUSD_PER_USD; + +const SCHEMA = /* sql */ ` +CREATE TABLE IF NOT EXISTS balances ( + pubkey TEXT PRIMARY KEY, + credit_nusd INTEGER NOT NULL DEFAULT 0, + updated INTEGER NOT NULL +); +CREATE TABLE IF NOT EXISTS usage ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + pubkey TEXT NOT NULL, + ts INTEGER NOT NULL, + model TEXT NOT NULL, + in_tokens INTEGER NOT NULL, + out_tokens INTEGER NOT NULL, + cache_read INTEGER NOT NULL DEFAULT 0, + cache_write INTEGER NOT NULL DEFAULT 0, + cost_usd REAL NOT NULL, + charge_usd REAL NOT NULL, + mode TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS usage_pubkey_ts ON usage (pubkey, ts DESC); +CREATE TABLE IF NOT EXISTS credits ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + pubkey TEXT NOT NULL, + ts INTEGER NOT NULL, + amount_usd REAL NOT NULL, + source TEXT NOT NULL, + ref TEXT +); +CREATE INDEX IF NOT EXISTS credits_pubkey_ts ON credits (pubkey, ts DESC); +`; + +export interface UsageRecord { + ts: number; + model: string; + inputTokens: number; + outputTokens: number; + cacheRead: number; + cacheWrite: number; + costUsd: number; + chargeUsd: number; + mode: string; +} + +export interface CreditRecord { + ts: number; + amountUsd: number; + source: string; + ref: string | null; +} + +export interface Balance { + pubkey: string; + balanceUsd: number; +} + +export class Ledger { + private readonly db: Database; + private readonly balanceStmt; + private readonly creditBalanceStmt; + private readonly creditLogStmt; + private readonly chargeUsageStmt; + private readonly chargeDebitStmt; + private readonly usageStmt; + private readonly creditsStmt; + + constructor(path: string = ":memory:") { + this.db = new Database(path); + // ponytail: default journal (not WAL); fine for single-process low + // concurrency. Switch to WAL if multi-reader throughput ever matters. + this.db.exec(SCHEMA); + this.balanceStmt = this.db.prepare( + `SELECT credit_nusd FROM balances WHERE pubkey = ?`, + ); + this.creditBalanceStmt = this.db.prepare( + `INSERT INTO balances (pubkey, credit_nusd, updated) VALUES (?, ?, ?) + ON CONFLICT(pubkey) DO UPDATE SET + credit_nusd = credit_nusd + excluded.credit_nusd, + updated = excluded.updated`, + ); + this.creditLogStmt = this.db.prepare( + `INSERT INTO credits (pubkey, ts, amount_usd, source, ref) VALUES (?, ?, ?, ?, ?)`, + ); + this.chargeUsageStmt = this.db.prepare( + `INSERT INTO usage + (pubkey, ts, model, in_tokens, out_tokens, cache_read, cache_write, cost_usd, charge_usd, mode) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ); + this.chargeDebitStmt = this.db.prepare( + // Clamp at 0: any residual is the bounded input-estimate error, not a + // runaway. Next call sees 0 and rejects at the gate (pricing.ts). + `UPDATE balances SET credit_nusd = MAX(0, credit_nusd - ?), updated = ? + WHERE pubkey = ?`, + ); + this.usageStmt = this.db.prepare( + `SELECT ts, model, in_tokens, out_tokens, cache_read, cache_write, cost_usd, charge_usd, mode + FROM usage WHERE pubkey = ? ORDER BY ts DESC LIMIT ?`, + ); + this.creditsStmt = this.db.prepare( + `SELECT ts, amount_usd, source, ref FROM credits WHERE pubkey = ? ORDER BY ts DESC LIMIT ?`, + ); + } + + /** Current balance in USD (0 when the pubkey has no row). */ + balanceUsd(pubkey: string): number { + const row = this.balanceStmt.get(pubkey) as + { credit_nusd: number } | undefined; + return usd(row?.credit_nusd ?? 0); + } + + /** Grant USD credit to a pubkey (Lightning top-up / admin grant). */ + credit( + pubkey: string, + amountUsd: number, + source: string, + ref: string | null = null, + ): void { + const ts = Date.now(); + this.creditBalanceStmt.run(pubkey, nusd(amountUsd), ts); + this.creditLogStmt.run(pubkey, ts, amountUsd, source, ref); + } + + /** Record a charge and debit `min(chargeUsd, balance)`. */ + charge(pubkey: string, rec: Omit): void { + const ts = Date.now(); + this.chargeUsageStmt.run( + pubkey, + ts, + rec.model, + rec.inputTokens, + rec.outputTokens, + rec.cacheRead, + rec.cacheWrite, + rec.costUsd, + rec.chargeUsd, + rec.mode, + ); + this.chargeDebitStmt.run(nusd(rec.chargeUsd), ts, pubkey); + } + + history(pubkey: string, limit: number = 50): UsageRecord[] { + const rows = this.usageStmt.all(pubkey, limit) as Array<{ + ts: number; + model: string; + in_tokens: number; + out_tokens: number; + cache_read: number; + cache_write: number; + cost_usd: number; + charge_usd: number; + mode: string; + }>; + return rows.map((r) => ({ + ts: r.ts, + model: r.model, + inputTokens: r.in_tokens, + outputTokens: r.out_tokens, + cacheRead: r.cache_read, + cacheWrite: r.cache_write, + costUsd: r.cost_usd, + chargeUsd: r.charge_usd, + mode: r.mode, + })); + } + + credits(pubkey: string, limit: number = 50): CreditRecord[] { + const rows = this.creditsStmt.all(pubkey, limit) as Array<{ + ts: number; + amount_usd: number; + source: string; + ref: string | null; + }>; + return rows.map((r) => ({ + ts: r.ts, + amountUsd: r.amount_usd, + source: r.source, + ref: r.ref, + })); + } + + close(): void { + this.db.close(); + } +} diff --git a/packages/server/src/main.ts b/packages/server/src/main.ts index d405559..bea4971 100644 --- a/packages/server/src/main.ts +++ b/packages/server/src/main.ts @@ -1,10 +1,14 @@ /** * Muxll server entry point. Builds the pi-ai provider registry from * ambient env (built-in providers read their API keys from the environment), - * then starts the CVM server. + * loads the model policy, then starts the CVM server. */ +import { renameSync, readFileSync, writeFileSync } from "node:fs"; import { ApplesauceRelayPool, PrivateKeySigner } from "@contextvm/sdk"; import { builtinModels } from "@earendil-works/pi-ai/providers/all"; +import type { ModelPolicy, ModelPolicySet } from "@muxll/core"; +import { Ledger } from "./ledger.ts"; +import { PriceBook } from "./pricing.ts"; import { startServer } from "./server.ts"; function parseRelays(): string[] { @@ -18,17 +22,91 @@ function parseRelays(): string[] { return ["wss://relay.contextvm.org"]; } +/** Comma-separated env value → trimmed, non-empty list. */ +function parseList(raw: string | undefined): string[] { + if (!raw) return []; + return raw + .split(",") + .map((s) => s.trim()) + .filter(Boolean); +} + +/** + * The model policy is file-authoritative once a config file exists; env vars + * (`MUXLL_MODELS_ALLOW`/`MUXLL_MODELS_DENY`) seed it only when there's no file. + * The file is created lazily on the first `admin.models.setPolicy` mutation, so + * env vars keep re-applying on each start until an admin change persists state. + * Delete the file + restart to go back to env-driven. + */ +function readInitialPolicy(path: string): ModelPolicySet { + let parsed: unknown; + try { + parsed = JSON.parse(readFileSync(path, "utf-8")); + } catch { + return envSeed(); + } + // Trust our own output / an operator's hand-edit, but guard against a corrupt + // or hand-mangled file: require `default` to be an object, else re-seed. + const set = parsed as Partial; + if ( + !set || + typeof set !== "object" || + set.default === null || + typeof set.default !== "object" + ) { + return envSeed(); + } + return set as ModelPolicySet; +} + +function envSeed(): ModelPolicySet { + const allow = parseList(process.env.MUXLL_MODELS_ALLOW); + const deny = parseList(process.env.MUXLL_MODELS_DENY); + const def: ModelPolicy = {}; + if (allow.length) def.allow = allow; + if (deny.length) def.deny = deny; + return { default: def }; +} + +/** Atomic write (temp + rename) so a crash mid-write can't brick the config. */ +function writePolicyFile(path: string, set: ModelPolicySet): void { + const tmp = `${path}.tmp`; + writeFileSync(tmp, `${JSON.stringify(set, null, 2)}\n`, "utf-8"); + renameSync(tmp, path); +} + async function main() { const signer = new PrivateKeySigner(process.env.MUXLL_PRIVATE_KEY); const relayUrls = parseRelays(); const pubkey = await signer.getPublicKey(); + const policyFile = process.env.MUXLL_CONFIG_FILE ?? "muxll-config.json"; + + // Billing is opt-in via env: set MUXLL_BILLING=1 to enable. USD balances are + // funded by admin grants (Phase 1); Lightning top-up arrives in Phase 3. + // markupPct/minCharge default to 0 (cost passthrough) when unset. + const billingEnabled = process.env.MUXLL_BILLING === "1"; + const ledger = billingEnabled + ? new Ledger(process.env.MUXLL_LEDGER_DB ?? "muxll-ledger.sqlite") + : undefined; + const priceBook = billingEnabled + ? new PriceBook({ + markupPct: Number(process.env.MUXLL_MARKUP_PCT ?? 0), + minChargeUsd: Number(process.env.MUXLL_MIN_CHARGE_USD ?? 0), + }) + : undefined; + const instance = await startServer({ signer, relayHandler: new ApplesauceRelayPool(relayUrls), models: builtinModels(), // built-in providers read keys from env serverInfo: { name: "muxll" }, isAnnouncedServer: process.env.MUXLL_ANNOUNCED === "1", + policy: readInitialPolicy(policyFile), + adminPubkeys: parseList(process.env.MUXLL_ADMIN_PUBKEYS), + onPolicyMutate: (set) => writePolicyFile(policyFile, set), + ledger, + priceBook, }); console.log("Muxll CVM server running."); diff --git a/packages/server/src/pricing.ts b/packages/server/src/pricing.ts new file mode 100644 index 0000000..6f46d50 --- /dev/null +++ b/packages/server/src/pricing.ts @@ -0,0 +1,124 @@ +/** + * Pricing + the output-token cap that protects operators from overserving when + * a caller's balance runs low. + * + * Phase 1 is **markup-only**: the charge is the upstream cost pi-ai computes + * from `Model.cost` (USD per 1M tokens via `calculateCost`) times `(1 + + * markupPct/100)`. Phase 2 adds fixed per-Mtok rates and per-model/per-pubkey + * overrides (the `byModel`/`byPubkey` matrix from docs/billing.md); the + * `PriceBook` constructor and `perTokenRates`/`chargeUsd` shapes are the seam + * it widens. + * + * Why we recompute cost server-side rather than trusting `usage.cost.total`: + * the faux provider reports `cost.total = 0`, and real providers may omit it. + * The router is the pricing authority; `calculateCost(model, usage)` is + * deterministic and works uniformly. (It also populates `usage.cost` in place, + * so the response's `usage.cost` reflects the same computation.) + */ +import { + calculateCost, + type Api, + type Model, + type Usage, +} from "@earendil-works/pi-ai"; + +export interface PricingConfig { + /** Markup over upstream cost, in percent (10 = +10%). */ + markupPct: number; + /** Floor charge in USD applied to every chargeable request (anti-spam). */ + minChargeUsd: number; +} + +export interface TokenRates { + inputPerToken: number; + outputPerToken: number; +} + +export const MARKUP_MODE = "markup" as const; + +export class PriceBook { + constructor(private readonly cfg: PricingConfig) {} + + get mode(): "markup" { + return MARKUP_MODE; + } + + get minChargeUsd(): number { + return this.cfg.minChargeUsd; + } + + /** Per-token USD rates for the cap. Markup mode derives these from + * `Model.cost` (USD/Mtok); Phase 2 fixed mode returns operator-set rates. */ + perTokenRates(model: Model): TokenRates { + const m = 1 + this.cfg.markupPct / 100; + return { + inputPerToken: ((model.cost.input ?? 0) / 1e6) * m, + outputPerToken: ((model.cost.output ?? 0) / 1e6) * m, + }; + } + + /** Upstream USD cost for a real usage, computed from `Model.cost`. Mutates + * `usage.cost` in place (idempotent) so the response reflects it. */ + costUsd(model: Model, usage: Usage): number { + return calculateCost(model, usage).total; + } + + /** Final charge for a usage: upstream cost × markup, floored at minCharge. */ + chargeUsd(model: Model, usage: Usage): number { + const charge = this.costUsd(model, usage) * (1 + this.cfg.markupPct / 100); + return Math.max(charge, this.cfg.minChargeUsd); + } +} + +export interface CapInput { + balanceUsd: number; + /** Pre-call estimate of input tokens (conservative/overestimate is safe). */ + inputEstimateTokens: number; + rates: TokenRates; + minChargeUsd: number; + userMaxTokens?: number; + modelMaxTokens?: number; + contextWindow?: number; +} + +export type CapResult = + { reject: true; reason: string } | { maxTokens: number }; + +/** + * Bound a request's output so the caller's balance can afford it. Returns the + * `maxTokens` to pass to the provider, or `reject` when even the input floor + * isn't covered. The provider then stops at the ceiling — no stream is ever + * killed mid-flight for balance reasons. + * + * Not a concurrency lock: N concurrent requests each compute their own cap from + * the same balance snapshot, so overdraft is bounded to ~(N-1) × per-request + * cap. The cap is what makes that deferral safe (docs/billing.md). + */ +export function computeOutputCap(i: CapInput): CapResult { + const inputCostEst = i.inputEstimateTokens * i.rates.inputPerToken; + const budget = i.balanceUsd - inputCostEst; + if (budget < i.minChargeUsd) { + return { reject: true, reason: "insufficient balance" }; + } + // Free output (no priced output): cap by the caller/model limit only. + if (i.rates.outputPerToken <= 0) { + const cap = Math.min( + i.userMaxTokens ?? Infinity, + i.modelMaxTokens ?? Infinity, + ); + return Number.isFinite(cap) + ? { maxTokens: Math.max(1, Math.floor(cap)) } + : { maxTokens: 1 }; + } + const maxOut = Math.floor(budget / i.rates.outputPerToken); + if (maxOut < 1) { + return { reject: true, reason: "insufficient balance for output tokens" }; + } + let cap = Math.min( + i.userMaxTokens ?? Infinity, + maxOut, + i.modelMaxTokens ?? Infinity, + i.contextWindow ?? Infinity, + ); + return { maxTokens: Math.max(1, Math.floor(cap)) }; +} diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index 1270755..3f8aadf 100644 --- a/packages/server/src/server.ts +++ b/packages/server/src/server.ts @@ -20,28 +20,75 @@ import { type OpenStreamWriter, type RelayHandler, } from "@contextvm/sdk"; +import { clampThinkingLevel } from "@earendil-works/pi-ai"; import type { Api, AssistantMessageEvent, Context, Model, Models, + StreamOptions, + Usage, } from "@earendil-works/pi-ai"; import { + adminBalanceCreditInput, + adminBalanceGetInput, + balanceGetInput, chatInput, - modelsOutput, METHODS, + modelsListInput, + modelsOutput, + modelPolicySetOutput, + modelPolicySetPatchInput, type FinishReason, + type ModelPolicySet, } from "@muxll/core"; +import { Catalog, modelTag } from "./catalog.ts"; import { ZERO_USAGE, chatResult, + mapReasoning, mapToolChoice, - resolveModel, toContext, toFinishReason, toUsage, } from "./wire.ts"; +import { Ledger } from "./ledger.ts"; +import { PriceBook, computeOutputCap } from "./pricing.ts"; + +/** Billing context threaded into a chat handler when a ledger + pricebook are configured. */ +interface BillingCtx { + ledger: Ledger; + priceBook: PriceBook; + pubkey: string; +} + +/** Conservative pre-call input-token estimate. Biased high (safe for operator). */ +function estimateInputTokens(messages: unknown): number { + // ponytail: pi-ai's estimateContextTokens lives under utils/ (not in its + // exports map). JSON length / 4 is the same chars-per-token heuristic and + // overestimates (role names, JSON syntax) — over-reserving input grants + // fewer output tokens, never an over-charge. + return Math.ceil(JSON.stringify(messages).length / 4); +} + +/** Compute the real charge and debit the ledger after a successful turn. */ +function settleBilling(b: BillingCtx, target: Model, usage: Usage): void { + // chargeUsd recomputes upstream cost from Model.cost via calculateCost + // (faux reports 0; real providers may omit) and populates usage.cost in + // place, so the response's usage.cost reflects the same computation. + const chargeUsd = b.priceBook.chargeUsd(target, usage); + b.ledger.charge(b.pubkey, { + model: modelTag(target), + inputTokens: usage.input, + outputTokens: usage.output, + cacheRead: usage.cacheRead, + cacheWrite: usage.cacheWrite, + costUsd: usage.cost.total, + chargeUsd, + mode: b.priceBook.mode, + }); +} function getOpenStreamWriter(extra: { _meta?: Record; @@ -56,8 +103,22 @@ function getOpenStreamWriter(extra: { return stream; } +/** Pull the authenticated caller pubkey the Nostr transport injected into _meta. */ +function clientPubkey(extra: { + _meta?: Record; +}): string | undefined { + const pk = (extra._meta as { clientPubkey?: unknown } | undefined) + ?.clientPubkey; + return typeof pk === "string" ? pk : undefined; +} + /** Register chat.complete and models.list on an McpServer. */ -export function registerTools(server: McpServer, models: Models): void { +export function registerTools( + server: McpServer, + models: Models, + catalog: Catalog, + billingOpts?: { ledger: Ledger; priceBook: PriceBook }, +): void { // chat.complete registers no MCP outputSchema because the two result shapes // differ: non-streaming returns the full CreateChatCompletionResponse (built // by chatResult), streaming returns a usage metadata summary. The SDK would @@ -72,7 +133,9 @@ export function registerTools(server: McpServer, models: Models): void { inputSchema: chatInput, }, async (input, extra) => { - const target = resolveModel(models, input.model); + // catalog.resolve hides policy-blocked models as "unknown" so callers + // can't probe for disallowed models through chat.complete. + const target = catalog.resolve(input.model, clientPubkey(extra)); if (!target) { throw new Error(`Unknown model: ${input.model}`); } @@ -81,23 +144,55 @@ export function registerTools(server: McpServer, models: Models): void { tools: input.tools, target, }); + + // Billing: cap output tokens to what the caller's balance can afford, + // so no stream is ever served beyond prepaid credit. Disabled when no + // ledger is configured (chat stays free, preserving the pre-billing path). + let maxTokens = input.max_tokens; + let billing: BillingCtx | undefined; + if (billingOpts) { + const pk = clientPubkey(extra); + if (!pk) throw new Error("cannot determine caller for billing"); + const cap = computeOutputCap({ + balanceUsd: billingOpts.ledger.balanceUsd(pk), + inputEstimateTokens: estimateInputTokens(input.messages), + rates: billingOpts.priceBook.perTokenRates(target), + minChargeUsd: billingOpts.priceBook.minChargeUsd, + userMaxTokens: input.max_tokens, + modelMaxTokens: target.maxTokens, + contextWindow: target.contextWindow, + }); + if ("reject" in cap) throw new Error(cap.reason); + maxTokens = cap.maxTokens; + billing = { ...billingOpts, pubkey: pk }; + } + const options = { temperature: input.temperature, - maxTokens: input.max_tokens, - // tool_choice is normalized per target API in mapToolChoice - // (anthropic/bedrock/google use "any" for required; the object form - // becomes {type:"tool",name}; openai-responses drops it). - // ponytail: response_format, top_p/stop/seed/penalties, reasoning_effort - // still not wired (need onPayload; deferred). + maxTokens, + // tool_choice / reasoning_effort are normalized per target API: + // mapToolChoice (anthropic/bedrock/google use "any" for required; the + // object form becomes {type:"tool",name}; openai-responses drops it), + // mapReasoning (thinking on + effort/budget per provider). + // clampThinkingLevel respects each model's thinkingLevelMap, so a + // level the model can't do (or a non-reasoning model) clamps to "off" + // and mapReasoning enables nothing. + // ponytail: response_format, top_p/stop/seed/penalties still not wired. ...(input.tool_choice !== undefined ? { toolChoice: mapToolChoice(target.api, input.tool_choice) } : {}), + ...(input.reasoning_effort + ? mapReasoning( + target.api, + clampThinkingLevel(target, input.reasoning_effort), + ) + : {}), }; if (input.stream) { - return runStreamed(models, target, context, options, extra); + return runStreamed(models, target, context, options, extra, billing); } - return runComplete(models, target, context, options); + return runComplete(models, target, context, options, billing); }, ); @@ -105,29 +200,161 @@ export function registerTools(server: McpServer, models: Models): void { METHODS.modelsList, { title: "List Models", - description: "List all models available across configured providers.", - inputSchema: {}, + description: + "List models available across configured providers (after the server's model policy). Optional `search` narrows by provider/id substring; `limit` caps the count.", + inputSchema: modelsListInput, outputSchema: modelsOutput, }, - async () => { - const all = models.getModels(); + async (input, extra) => { + const data = catalog.list(input, clientPubkey(extra)).map((m) => ({ + id: `${m.provider}/${m.id}`, + object: "model", + // ponytail: pi-ai carries no model creation time; 0 stands in for OpenAI's `created`. + created: 0, + owned_by: m.provider, + name: m.name, + api: m.api, + contextWindow: m.contextWindow, + maxTokens: m.maxTokens, + reasoning: m.reasoning, + input: m.input, + })); + return { + content: [], + structuredContent: { object: "list", data }, + }; + }, + ); +} + +/** + * Register admin.model policy tools, gated by the caller's authenticated + * pubkey. With no admins configured the surface isn't registered at all (no + * dead, always-unauthorized tool advertised via tools/list). + */ +export function registerAdminTools( + server: McpServer, + catalog: Catalog, + adminPubkeys: string[], +): void { + if (adminPubkeys.length === 0) return; + const adminKeys = new Set(adminPubkeys); + + const requireAdmin = (extra: { _meta?: Record }): void => { + if (!adminKeys.has(clientPubkey(extra) ?? "")) { + throw new Error("unauthorized: admin pubkey required"); + } + }; + + server.registerTool( + METHODS.adminModelsGetPolicy, + { + title: "Get Model Policy", + description: + "Return the current model visibility policy (default + per-client overrides). Admin-only.", + inputSchema: {}, + outputSchema: modelPolicySetOutput, + }, + async (_input, extra) => { + requireAdmin(extra); + return { content: [], structuredContent: catalog.getPolicy() }; + }, + ); + + server.registerTool( + METHODS.adminModelsSetPolicy, + { + title: "Set Model Policy", + description: + "Patch the model visibility policy. Per field: array replaces, null clears, absent unchanged. Returns the new full policy. Admin-only.", + inputSchema: modelPolicySetPatchInput, + outputSchema: modelPolicySetOutput, + }, + async (input, extra) => { + requireAdmin(extra); + return { content: [], structuredContent: catalog.setPolicy(input) }; + }, + ); +} + +/** + * Register balance.* + admin.balance.* tools. `balance.get` is own-balance + * (the caller's pubkey); `admin.balance.*` are gated like admin.models.* and + * only registered when admins are configured. Only called when a ledger exists. + */ +export function registerBalanceTools( + server: McpServer, + ledger: Ledger, + adminPubkeys: string[], +): void { + server.registerTool( + METHODS.balanceGet, + { + title: "Get Balance", + description: + "Return the caller's prepaid balance (USD) and recent usage. Balances are funded via admin.balance.credit (Lightning top-up arrives in a later phase).", + inputSchema: balanceGetInput, + }, + async (_input, extra) => { + const pk = clientPubkey(extra); + if (!pk) throw new Error("cannot determine caller"); + return { + content: [], + structuredContent: { + pubkey: pk, + balanceUsd: ledger.balanceUsd(pk), + recent: ledger.history(pk, 20), + }, + }; + }, + ); + + if (adminPubkeys.length === 0) return; + const adminKeys = new Set(adminPubkeys); + const requireAdmin = (extra: { _meta?: Record }): void => { + if (!adminKeys.has(clientPubkey(extra) ?? "")) { + throw new Error("unauthorized: admin pubkey required"); + } + }; + + server.registerTool( + METHODS.adminBalanceGet, + { + title: "Get Balance (admin)", + description: + "Return any caller's balance (USD) + recent usage. Admin-only.", + inputSchema: adminBalanceGetInput, + }, + async (input, extra) => { + requireAdmin(extra); + return { + content: [], + structuredContent: { + pubkey: input.pubkey, + balanceUsd: ledger.balanceUsd(input.pubkey), + recent: ledger.history(input.pubkey, input.limit ?? 50), + }, + }; + }, + ); + + server.registerTool( + METHODS.adminBalanceCredit, + { + title: "Credit Balance (admin)", + description: + "Grant USD credit to a caller pubkey (voucher / operator-funded). Admin-only.", + inputSchema: adminBalanceCreditInput, + }, + async (input, extra) => { + requireAdmin(extra); + ledger.credit(input.pubkey, input.amountUsd, "admin", input.note ?? null); return { content: [], structuredContent: { - object: "list", - data: all.map((m) => ({ - id: `${m.provider}/${m.id}`, - object: "model", - // ponytail: pi-ai carries no model creation time; 0 stands in for OpenAI's `created`. - created: 0, - owned_by: m.provider, - name: m.name, - api: m.api, - contextWindow: m.contextWindow, - maxTokens: m.maxTokens, - reasoning: m.reasoning, - input: m.input, - })), + pubkey: input.pubkey, + balanceUsd: ledger.balanceUsd(input.pubkey), + creditedUsd: input.amountUsd, }, }; }, @@ -138,9 +365,11 @@ async function runComplete( models: Models, target: Model, context: Context, - options: { temperature?: number; maxTokens?: number; toolChoice?: unknown }, + options: StreamOptions & Record, + billing?: BillingCtx, ) { const message = await models.complete(target, context, options); + if (billing && message.usage) settleBilling(billing, target, message.usage); return chatResult(target, message, Math.floor(message.timestamp / 1000)); } @@ -148,8 +377,9 @@ async function runStreamed( models: Models, target: Model, context: Context, - options: { temperature?: number; maxTokens?: number; toolChoice?: unknown }, + options: StreamOptions & Record, extra: { _meta?: Record }, + billing?: BillingCtx, ) { const stream = getOpenStreamWriter(extra); // ponytail: contentType can't be set here — the CVM transport owns @@ -233,6 +463,8 @@ async function runStreamed( } const message = final && final.type === "done" ? final.message : null; + if (billing && message?.usage) + settleBilling(billing, target, message.usage); // Final chunk: empty delta + finish_reason + usage (always included — // equivalent to stream_options.include_usage: true). await writeChunk( @@ -273,6 +505,16 @@ export interface StartServerOptions { models: Models; serverInfo?: { name?: string; website?: string }; isAnnouncedServer?: boolean; + /** Initial model visibility policy. Defaults to allow-all. */ + policy?: ModelPolicySet; + /** Caller pubkeys allowed to use admin.* tools. Empty = no admin surface. */ + adminPubkeys?: string[]; + /** Invoked with the new policy set after every admin.models.setPolicy (persistence). */ + onPolicyMutate?: (set: ModelPolicySet) => void; + /** Prepaid balance ledger. When set, chat.complete charges against balances. */ + ledger?: Ledger; + /** Pricing config. Required when `ledger` is set (markup-only in Phase 1). */ + priceBook?: PriceBook; } export interface StartedServer { @@ -289,13 +531,33 @@ export async function startServer( name: options.serverInfo?.name ?? "muxll", version: "0.1.0", }); - registerTools(server, options.models); + const catalog = new Catalog( + options.models, + options.policy ?? { default: {} }, + options.onPolicyMutate, + ); + if (options.ledger && !options.priceBook) { + throw new Error("priceBook is required when ledger is provided"); + } + const billingOpts = + options.ledger && options.priceBook + ? { ledger: options.ledger, priceBook: options.priceBook } + : undefined; + registerTools(server, options.models, catalog, billingOpts); + registerAdminTools(server, catalog, options.adminPubkeys ?? []); + if (options.ledger) { + registerBalanceTools(server, options.ledger, options.adminPubkeys ?? []); + } const transport = new NostrServerTransport({ signer: options.signer, relayHandler: options.relayHandler, serverInfo: options.serverInfo, isAnnouncedServer: options.isAnnouncedServer ?? false, + // injectClientPubkey is the authenticated caller identity (derived from + // the Nostr gift-wrap signer, overwritten by the transport so it can't be + // spoofed via _meta). Enables per-client model policy + admin gating. + injectClientPubkey: true, openStream: { enabled: true }, oversizedTransfer: { enabled: true }, }); @@ -307,6 +569,7 @@ export async function startServer( transport, close: async () => { await server.close(); + options.ledger?.close(); }, }; } diff --git a/packages/server/src/wire.ts b/packages/server/src/wire.ts index c945c80..680c29d 100644 --- a/packages/server/src/wire.ts +++ b/packages/server/src/wire.ts @@ -12,6 +12,7 @@ import { type ImageContent, type Message, type Model, + type ModelThinkingLevel, type Models, type ToolCall, type Usage, @@ -105,6 +106,52 @@ export function mapToolChoice(api: Api, choice: ToolChoice): unknown { return choice; } +/** + * Map a clamped thinking level to a target API's provider-specific reasoning + * option (the same per-API dispatch shape as mapToolChoice). The caller should + * clampThinkingLevel() first; an "off" result (a non-reasoning model, or a + * level the model's thinkingLevelMap rejects) returns no fields, leaving + * thinking off. Unknown APIs likewise return no fields — nothing invented. + */ +export function mapReasoning( + api: Api, + level: ModelThinkingLevel, +): Record { + if (level === "off") return {}; + switch (api) { + case "anthropic-messages": + // Adaptive models read `effort`; budget-based (older) models ignore it + // and fall back to a default token budget. "minimal" isn't an Anthropic + // effort, so omit it and let the model pick its default. + // ponytail: no level→budget mapping for older budget models + // (low/medium/high → 2048/8192/16384); they get the 1024 default. + return { + thinkingEnabled: true, + ...(level !== "minimal" ? { effort: level } : {}), + }; + case "openai-completions": + case "openai-responses": + return { reasoningEffort: level }; + case "bedrock-converse-stream": + // Bedrock reads the uniform ThinkingLevel directly and does its own + // budget math — the cleanest of the providers. + return { reasoning: level }; + case "google-generative-ai": + case "google-vertex": + // ponytail: enabled-only; level→budgetTokens mapping deferred. + return { thinking: { enabled: true } }; + case "mistral-conversations": + // ponytail: mistral exposes only none|high and splits activation across + // promptMode/reasoningEffort per model; map the high end and skip the + // rest until a mistral-reasoning caller actually needs it. + return level === "high" || level === "xhigh" + ? { reasoningEffort: "high" } + : {}; + default: + return {}; + } +} + /** * Convert an OpenAI image URL (data: or http(s):) to a pi-ai ImageContent * block. OpenAI accepts URLs directly, but base64 is universal — Anthropic and diff --git a/packages/server/tests/billing.test.ts b/packages/server/tests/billing.test.ts new file mode 100644 index 0000000..72ed31f --- /dev/null +++ b/packages/server/tests/billing.test.ts @@ -0,0 +1,284 @@ +/** + * Phase 1 billing end-to-end over the real Nostr transports + an in-process + * mock relay: admin grants USD credit, chat.complete caps output to what the + * balance affords and debits the real cost, balance.get reflects it, and + * unfunded callers are rejected. No Lightning (that's Phase 3); the ledger is + * funded via admin.balance.credit. Mirrors the policy.test.ts harness. + * + * Charges read the *actual* usage back from the response and recompute the + * expected cost, so faux's randomized output-token sizing doesn't make the + * assertions flaky. + */ +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import { Client } from "@contextvm/mcp-sdk/client"; +import { NostrClientTransport, PrivateKeySigner } from "@contextvm/sdk"; +import { + createModels, + fauxAssistantMessage, + fauxProvider, + type Context, + type StreamOptions, +} from "@earendil-works/pi-ai"; +import { METHODS } from "@muxll/core"; +import { + Ledger, + PriceBook, + startServer, + type StartedServer, +} from "@muxll/server"; +import { MockRelayHub } from "@muxll/test-utils"; + +/** One faux model with explicit USD/Mtok pricing so charges are non-zero. */ +const MODEL_COST = { input: 1, output: 2, cacheRead: 0.5, cacheWrite: 1.25 }; +const MARKUP_PCT = 10; + +function buildModels() { + const models = createModels(); + const handle = fauxProvider({ + provider: "anthropic", + models: [ + { + id: "claude-opus", + cost: MODEL_COST, + contextWindow: 200_000, + maxTokens: 8_000, + }, + ], + }); + models.setProvider(handle.provider); + handle.setResponses([fauxAssistantMessage("anthropic reply")]); + return { models, handle }; +} + +let server: StartedServer; +let hub: MockRelayHub; +let serverPk: string; +let adminClient: Client; +let handle: ReturnType["handle"]; + +// Subject clients, each with its own pubkey/balance so tests don't interfere. +let clientA: Client; // funded $1 — charge test +let clientB: Client; // funded $1 — stream test +let clientC: Client; // funded $0.001 — cap test +let clientD: Client; // unfunded — reads-0 / rejected / non-admin tests +let pkA: string; +let pkD: string; + +async function connectClient(name: string, signer = new PrivateKeySigner()) { + const client = new Client({ name, version: "0.0.0" }); + await client.connect( + new NostrClientTransport({ + signer, + serverPubkey: serverPk, + relayHandler: hub.createRelayHandler(), + openStream: { enabled: true }, + oversizedTransfer: { enabled: true }, + isStateless: true, + logLevel: "silent", + }), + ); + return { client, pk: await signer.getPublicKey() }; +} + +function call(c: Client, name: string, args: Record) { + return c.callTool({ name, arguments: args }, undefined, { + onprogress: () => undefined, + resetTimeoutOnProgress: true, + }); +} + +const balanceOf = async (c: Client): Promise => + ( + (await call(c, METHODS.balanceGet, {})).structuredContent as { + balanceUsd: number; + } + ).balanceUsd; + +beforeAll(async () => { + hub = new MockRelayHub(); + const serverSigner = new PrivateKeySigner(); + serverPk = await serverSigner.getPublicKey(); + const adminSigner = new PrivateKeySigner(); + + const built = buildModels(); + server = await startServer({ + signer: serverSigner, + relayHandler: hub.createRelayHandler(), + models: built.models, + adminPubkeys: [await adminSigner.getPublicKey()], + ledger: new Ledger(":memory:"), + priceBook: new PriceBook({ markupPct: MARKUP_PCT, minChargeUsd: 0 }), + }); + handle = built.handle; + + adminClient = (await connectClient("admin", adminSigner)).client; + const a = await connectClient("a"); + const b = await connectClient("b"); + const c = await connectClient("c"); + const d = await connectClient("d"); + clientA = a.client; + clientB = b.client; + clientC = c.client; + clientD = d.client; + pkA = a.pk; + pkD = d.pk; + + await call(adminClient, METHODS.adminBalanceCredit, { + pubkey: a.pk, + amountUsd: 1.0, + }); + await call(adminClient, METHODS.adminBalanceCredit, { + pubkey: b.pk, + amountUsd: 1.0, + }); + await call(adminClient, METHODS.adminBalanceCredit, { + pubkey: c.pk, + amountUsd: 0.001, + }); +}); + +afterAll(async () => { + await adminClient.close(); + await clientA.close(); + await clientB.close(); + await clientC.close(); + await clientD.close(); + await server.close(); +}); + +describe("balance.* / admin.balance.*", () => { + test("admin grant is reflected in admin and own balance reads", async () => { + const viaAdmin = await call(adminClient, METHODS.adminBalanceGet, { + pubkey: pkA, + }); + expect(viaAdmin.structuredContent).toMatchObject({ + pubkey: pkA, + balanceUsd: 1.0, + }); + expect( + (await call(clientA, METHODS.balanceGet, {})).structuredContent, + ).toMatchObject({ balanceUsd: 1.0 }); + }); + + test("a pubkey with no credit reads zero", async () => { + expect( + (await call(clientD, METHODS.balanceGet, {})).structuredContent, + ).toMatchObject({ balanceUsd: 0 }); + }); + + test("non-admin caller cannot grant credit", async () => { + const res = await call(clientD, METHODS.adminBalanceCredit, { + pubkey: pkD, + amountUsd: 5, + }); + expect(res.isError).toBe(true); + expect(JSON.stringify(res.content)).toContain("unauthorized"); + }); + + test("admin can grant credit to any pubkey", async () => { + const g = await connectClient("g"); + try { + const res = await call(adminClient, METHODS.adminBalanceCredit, { + pubkey: g.pk, + amountUsd: 0.25, + }); + expect(res.structuredContent).toMatchObject({ + balanceUsd: 0.25, + creditedUsd: 0.25, + }); + } finally { + await g.client.close(); + } + }); +}); + +describe("chat.complete — billing", () => { + test("a funded call debits cost×markup and records usage", async () => { + handle.setResponses([fauxAssistantMessage("the quick brown fox")]); + const before = await balanceOf(clientA); + + const res = await call(clientA, METHODS.chatComplete, { + model: "anthropic/claude-opus", + messages: [{ role: "user", content: "hello" }], + }); + expect(res.isError).toBeFalsy(); + + const usage = ( + res.structuredContent as { + usage: { + prompt_tokens: number; + completion_tokens: number; + cost: { total: number }; + }; + } + ).usage; + // Server recomputes cost from Model.cost (faux reports 0). + const expectedCost = + (usage.prompt_tokens * MODEL_COST.input + + usage.completion_tokens * MODEL_COST.output) / + 1e6; + expect(usage.cost.total).toBeCloseTo(expectedCost, 12); + + const expectedCharge = expectedCost * (1 + MARKUP_PCT / 100); + expect(before - (await balanceOf(clientA))).toBeCloseTo(expectedCharge, 12); + + const admin = await call(adminClient, METHODS.adminBalanceGet, { + pubkey: pkA, + }); + const recent = ( + admin.structuredContent as { + recent: { chargeUsd: number; mode: string }[]; + } + ).recent; + expect(recent[0]!.mode).toBe("markup"); + expect(recent[0]!.chargeUsd).toBeCloseTo(expectedCharge, 12); + }); + + test("a funded streaming call also debits (runStreamed settle path)", async () => { + handle.setResponses([fauxAssistantMessage("streamed reply text")]); + const before = await balanceOf(clientB); + const res = await call(clientB, METHODS.chatComplete, { + model: "anthropic/claude-opus", + messages: [{ role: "user", content: "stream me" }], + stream: true, + }); + expect(res.isError).toBeFalsy(); + const after = await balanceOf(clientB); + expect(after).toBeLessThan(before); + expect(after).toBeGreaterThan(0); + }); + + test("output is capped to what the balance can afford", async () => { + let captured: number | undefined; + handle.setResponses([ + (_ctx: Context, options: StreamOptions | undefined) => { + captured = options?.maxTokens; + return fauxAssistantMessage("capped"); + }, + ]); + + const res = await call(clientC, METHODS.chatComplete, { + model: "anthropic/claude-opus", + messages: [{ role: "user", content: "hi" }], + }); + expect(res.isError).toBeFalsy(); + + // The cap kicked in: maxTokens is well under the model's 8000, bounded by + // what $0.001 affords at the marked-up output rate. + expect(captured).toBeDefined(); + expect(captured!).toBeGreaterThan(0); + expect(captured!).toBeLessThan(8_000); + const outputPerToken = (MODEL_COST.output / 1e6) * (1 + MARKUP_PCT / 100); + expect(captured!).toBeLessThanOrEqual(Math.floor(0.001 / outputPerToken)); + }); + + test("a caller with zero balance is rejected", async () => { + handle.setResponses([fauxAssistantMessage("should not reach")]); + const res = await call(clientD, METHODS.chatComplete, { + model: "anthropic/claude-opus", + messages: [{ role: "user", content: "hi" }], + }); + expect(res.isError).toBe(true); + expect(JSON.stringify(res.content)).toContain("insufficient balance"); + }); +}); diff --git a/packages/server/tests/catalog.test.ts b/packages/server/tests/catalog.test.ts new file mode 100644 index 0000000..46831c9 --- /dev/null +++ b/packages/server/tests/catalog.test.ts @@ -0,0 +1,210 @@ +/** + * Unit tests for the model catalog: pure policy logic (match / precedence / + * patch / merge) plus Catalog behavior over a real pi-ai Models with several + * faux models. End-to-end (over the wire) coverage lives in policy.test.ts. + */ +import { describe, expect, test } from "bun:test"; +import { createModels, fauxProvider } from "@earendil-works/pi-ai"; +import { + applyPolicyPatch, + Catalog, + matches, + mergePolicy, + modelTag, + policyAllows, +} from "@muxll/server"; + +/** 4 faux models across 2 providers: anthropic/{opus,haiku}, openai/{4o,mini}. */ +function buildModels() { + const models = createModels(); + const a = fauxProvider({ + provider: "anthropic", + models: [{ id: "claude-opus" }, { id: "claude-haiku" }], + }); + const b = fauxProvider({ + provider: "openai", + models: [{ id: "gpt-4o" }, { id: "gpt-4o-mini" }], + }); + models.setProvider(a.provider); + models.setProvider(b.provider); + return models; +} + +const tags = (ms: { provider: string; id: string }[]) => + ms.map((m) => `${m.provider}/${m.id}`); + +describe("matches", () => { + test("case-insensitive substring on the tag", () => { + expect(matches("anthropic/claude-opus", "CLAUDE")).toBe(true); + expect(matches("anthropic/claude-opus", "anthropic")).toBe(true); + expect(matches("openai/gpt-4o", "gpt")).toBe(true); + expect(matches("anthropic/claude-opus", "gemini")).toBe(false); + }); + test("empty / whitespace needles never match", () => { + expect(matches("anthropic/claude-opus", "")).toBe(false); + expect(matches("anthropic/claude-opus", " ")).toBe(false); + }); +}); + +describe("policyAllows", () => { + test("no rules → allowed", () => { + expect(policyAllows("anthropic/claude-opus", {})).toBe(true); + }); + test("deny wins over allow", () => { + expect( + policyAllows("anthropic/claude-opus", { + allow: ["anthropic"], + deny: ["opus"], + }), + ).toBe(false); + }); + test("allow restricts to matching tags", () => { + expect( + policyAllows("anthropic/claude-opus", { allow: ["anthropic"] }), + ).toBe(true); + expect(policyAllows("openai/gpt-4o", { allow: ["anthropic"] })).toBe(false); + }); + test("absent allow = no restriction; empty allow = block all", () => { + expect(policyAllows("openai/gpt-4o", { allow: undefined })).toBe(true); + expect(policyAllows("openai/gpt-4o", { allow: [] })).toBe(false); + }); +}); + +describe("mergePolicy", () => { + test("defined fields override, absent fields inherit", () => { + expect( + mergePolicy( + { allow: ["anthropic"], deny: ["haiku"] }, + { allow: ["gpt"] }, + ), + ).toEqual({ allow: ["gpt"], deny: ["haiku"] }); + expect(mergePolicy({ allow: ["anthropic"] }, {})).toEqual({ + allow: ["anthropic"], + }); + }); + test("empty array overrides (block-all propagates)", () => { + expect(mergePolicy({ allow: ["anthropic"] }, { allow: [] })).toEqual({ + allow: [], + }); + }); +}); + +describe("applyPolicyPatch", () => { + test("array replaces, absent unchanged", () => { + expect( + applyPolicyPatch( + { default: { allow: ["anthropic"] } }, + { default: { deny: ["mini"] } }, + ), + ).toEqual({ default: { allow: ["anthropic"], deny: ["mini"] } }); + }); + test("null clears a field", () => { + expect( + applyPolicyPatch( + { default: { allow: ["anthropic"], deny: ["mini"] } }, + { default: { deny: null } }, + ), + ).toEqual({ default: { allow: ["anthropic"] } }); + }); + test("byPubkey: add, merge, then remove with null", () => { + const start = { default: {} }; + const added = applyPolicyPatch(start, { + byPubkey: { pk1: { allow: ["gpt"] } }, + }); + expect(added.byPubkey?.pk1).toEqual({ allow: ["gpt"] }); + const merged = applyPolicyPatch(added, { + byPubkey: { pk1: { deny: ["mini"] } }, + }); + expect(merged.byPubkey?.pk1).toEqual({ allow: ["gpt"], deny: ["mini"] }); + const removed = applyPolicyPatch(merged, { byPubkey: { pk1: null } }); + expect(removed.byPubkey).toBeUndefined(); + }); +}); + +describe("Catalog", () => { + test("no policy → all visible; resolve finds by tag and bare id", () => { + const c = new Catalog(buildModels(), { default: {} }); + expect(tags(c.visibleModels())).toEqual([ + "anthropic/claude-opus", + "anthropic/claude-haiku", + "openai/gpt-4o", + "openai/gpt-4o-mini", + ]); + expect(modelTag(c.resolve("anthropic/claude-opus")!)).toBe( + "anthropic/claude-opus", + ); + expect(modelTag(c.resolve("claude-opus")!)).toBe("anthropic/claude-opus"); + }); + + test("allow narrows visibleModels and resolve", () => { + const c = new Catalog(buildModels(), { default: { allow: ["anthropic"] } }); + expect(tags(c.visibleModels())).toEqual([ + "anthropic/claude-opus", + "anthropic/claude-haiku", + ]); + expect(c.resolve("openai/gpt-4o")).toBeUndefined(); // hidden = unknown + }); + + test("deny removes from both list and resolve", () => { + const c = new Catalog(buildModels(), { + default: { deny: ["mini", "haiku"] }, + }); + expect(tags(c.visibleModels())).toEqual([ + "anthropic/claude-opus", + "openai/gpt-4o", + ]); + expect(c.resolve("anthropic/claude-haiku")).toBeUndefined(); + }); + + test("list: search narrows, limit caps", () => { + const c = new Catalog(buildModels(), { default: {} }); + expect(tags(c.list({ search: "gpt" }))).toEqual([ + "openai/gpt-4o", + "openai/gpt-4o-mini", + ]); + expect(tags(c.list({ search: "openai" }))).toEqual([ + "openai/gpt-4o", + "openai/gpt-4o-mini", + ]); + expect(c.list({ limit: 1 })).toHaveLength(1); + expect(c.list({})).toHaveLength(4); // {} = unfiltered (today's behavior) + }); + + test("per-client override replaces fields over default", () => { + const c = new Catalog(buildModels(), { + default: { allow: ["anthropic"] }, + byPubkey: { pkGpt: { allow: ["gpt"] } }, + }); + // default caller: anthropic only + expect(tags(c.visibleModels())).toEqual([ + "anthropic/claude-opus", + "anthropic/claude-haiku", + ]); + // pkGpt caller: gpt only (allow replaced) + expect(tags(c.visibleModels("pkGpt"))).toEqual([ + "openai/gpt-4o", + "openai/gpt-4o-mini", + ]); + // override without allow inherits default's allow, adds its deny + const c2 = new Catalog(buildModels(), { + default: { allow: ["anthropic"] }, + byPubkey: { pk: { deny: ["haiku"] } }, + }); + expect(tags(c2.visibleModels("pk"))).toEqual(["anthropic/claude-opus"]); + }); + + test("setPolicy patches in place, fires onMutate, getPolicy clones", () => { + const mutated: unknown[] = []; + const c = new Catalog(buildModels(), { default: {} }, (s) => + mutated.push(s), + ); + const next = c.setPolicy({ default: { deny: ["mini"] } }); + expect(next.default.deny).toEqual(["mini"]); + expect(c.getPolicy().default.deny).toEqual(["mini"]); + expect(mutated).toHaveLength(1); + // getPolicy returns a detached copy + const snap = c.getPolicy(); + snap.default.deny = ["x"]; + expect(c.getPolicy().default.deny).toEqual(["mini"]); + }); +}); diff --git a/packages/server/tests/integration.test.ts b/packages/server/tests/integration.test.ts index 49b534e..26af86c 100644 --- a/packages/server/tests/integration.test.ts +++ b/packages/server/tests/integration.test.ts @@ -20,6 +20,7 @@ import { } from "@earendil-works/pi-ai"; import { mapToolChoice, + mapReasoning, startServer, toImageContent, type StartedServer, @@ -393,6 +394,134 @@ describe("mapToolChoice", () => { }); }); +describe("mapReasoning", () => { + test('"off" enables nothing on any api', () => { + expect(mapReasoning("anthropic-messages", "off")).toEqual({}); + expect(mapReasoning("openai-completions", "off")).toEqual({}); + }); + + test("anthropic: thinkingEnabled + effort, except minimal omits effort", () => { + expect(mapReasoning("anthropic-messages", "high")).toEqual({ + thinkingEnabled: true, + effort: "high", + }); + expect(mapReasoning("anthropic-messages", "minimal")).toEqual({ + thinkingEnabled: true, + }); + }); + + test("openai-completions/responses → reasoningEffort; bedrock → reasoning", () => { + expect(mapReasoning("openai-completions", "medium")).toEqual({ + reasoningEffort: "medium", + }); + expect(mapReasoning("openai-responses", "xhigh")).toEqual({ + reasoningEffort: "xhigh", + }); + expect(mapReasoning("bedrock-converse-stream", "high")).toEqual({ + reasoning: "high", + }); + }); + + test("google → thinking.enabled; mistral maps only the high end", () => { + expect(mapReasoning("google-generative-ai", "low")).toEqual({ + thinking: { enabled: true }, + }); + expect(mapReasoning("mistral-conversations", "high")).toEqual({ + reasoningEffort: "high", + }); + expect(mapReasoning("mistral-conversations", "low")).toEqual({}); + }); + + test("unknown api → empty (no thinking fields invented)", () => { + expect(mapReasoning("some-future-api", "high")).toEqual({}); + }); +}); + +describe("chat.complete (reasoning forwarding)", () => { + // Dedicated harness: the default faux provider uses a random api string, but + // mapReasoning dispatches on target.api — so this stands up a faux provider + // configured with a real api (anthropic-messages) and a reasoning-capable + // model, then captures the stream options the handler built. + let rServer: StartedServer; + let rClient: Client; + let rFaux: ReturnType; + + beforeAll(async () => { + const hub = new MockRelayHub(); + const signer = new PrivateKeySigner(); + const pubkey = await signer.getPublicKey(); + rFaux = fauxProvider({ + api: "anthropic-messages", + models: [{ id: "claude-reason", reasoning: true }], + }); + const models = createModels(); + models.setProvider(rFaux.provider); + rServer = await startServer({ + signer, + relayHandler: hub.createRelayHandler(), + models, + }); + rClient = new Client({ name: "muxll-test", version: "0.0.0" }); + await rClient.connect( + new NostrClientTransport({ + signer: new PrivateKeySigner(), + serverPubkey: pubkey, + relayHandler: hub.createRelayHandler(), + openStream: { enabled: true }, + oversizedTransfer: { enabled: true }, + }), + ); + }); + + afterAll(async () => { + await rClient.close(); + await rServer.close(); + }); + + test("forwards reasoning_effort as per-API thinking options to the provider", async () => { + let captured: Record | undefined; + rFaux.setResponses([ + (_ctx, options) => { + captured = options as Record; + return fauxAssistantMessage("thoughtful"); + }, + ]); + const result = await rClient.callTool({ + name: METHODS.chatComplete, + arguments: { + model: "claude-reason", + messages: [{ role: "user", content: "think hard" }], + reasoning_effort: "high", + }, + }); + const sc = result.structuredContent as { + choices: { message: { content: string } }[]; + }; + expect(sc.choices[0]!.message.content).toBe("thoughtful"); + expect(captured!.thinkingEnabled).toBe(true); + expect(captured!.effort).toBe("high"); + }); + + test("omits thinking options when reasoning_effort is not set", async () => { + let captured: Record | undefined; + rFaux.setResponses([ + (_ctx, options) => { + captured = options as Record; + return fauxAssistantMessage("plain"); + }, + ]); + await rClient.callTool({ + name: METHODS.chatComplete, + arguments: { + model: "claude-reason", + messages: [{ role: "user", content: "hi" }], + }, + }); + expect(captured!.thinkingEnabled).toBeUndefined(); + expect(captured!.effort).toBeUndefined(); + }); +}); + describe("toImageContent", () => { test("parses a data URL into mimeType + base64 data", async () => { expect(await toImageContent("data:image/png;base64,iVBORw0KGgo=")).toEqual({ diff --git a/packages/server/tests/policy.test.ts b/packages/server/tests/policy.test.ts new file mode 100644 index 0000000..40b6ed5 --- /dev/null +++ b/packages/server/tests/policy.test.ts @@ -0,0 +1,215 @@ +/** + * End-to-end model-policy tests over the real Nostr transports and an in-process + * mock relay: models.list search/limit, policy-driven hiding on both list and + * chat.complete, admin gating via injectClientPubkey, patch semantics, and the + * persistence hook. Pure policy logic lives in catalog.test.ts. + */ +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import { Client } from "@contextvm/mcp-sdk/client"; +import { NostrClientTransport, PrivateKeySigner } from "@contextvm/sdk"; +import { + createModels, + fauxAssistantMessage, + fauxProvider, +} from "@earendil-works/pi-ai"; +import { METHODS, type ModelPolicySet } from "@muxll/core"; +import { startServer, type StartedServer } from "@muxll/server"; +import { MockRelayHub } from "@muxll/test-utils"; + +/** 4 faux models across 2 providers; both handles so tests can queue replies. */ +function buildModels() { + const models = createModels(); + const a = fauxProvider({ + provider: "anthropic", + models: [{ id: "claude-opus" }, { id: "claude-haiku" }], + }); + const b = fauxProvider({ + provider: "openai", + models: [{ id: "gpt-4o" }, { id: "gpt-4o-mini" }], + }); + models.setProvider(a.provider); + models.setProvider(b.provider); + a.setResponses([fauxAssistantMessage("anthropic reply")]); + b.setResponses([fauxAssistantMessage("openai reply")]); + return models; +} + +const ids = (sc: unknown) => + (sc as { data: { id: string }[] }).data.map((m) => m.id); + +let server: StartedServer; +let adminClient: Client; +let plainClient: Client; +let plainPk: string; +let mutations: ModelPolicySet[]; + +async function connectClient( + hub: MockRelayHub, + serverPk: string, + name: string, + signer: PrivateKeySigner = new PrivateKeySigner(), +) { + const client = new Client({ name, version: "0.0.0" }); + await client.connect( + new NostrClientTransport({ + signer, + serverPubkey: serverPk, + relayHandler: hub.createRelayHandler(), + openStream: { enabled: true }, + oversizedTransfer: { enabled: true }, + isStateless: true, + logLevel: "silent", + }), + ); + return { client, pk: await signer.getPublicKey() }; +} + +function call( + client: Client, + name: string, + arguments_: Record, +) { + return client.callTool({ name, arguments: arguments_ }, undefined, { + onprogress: () => undefined, + resetTimeoutOnProgress: true, + }); +} + +beforeAll(async () => { + const hub = new MockRelayHub(); + const serverSigner = new PrivateKeySigner(); + const serverPk = await serverSigner.getPublicKey(); + + const adminSigner = new PrivateKeySigner(); + const adminPk = await adminSigner.getPublicKey(); + + mutations = []; + server = await startServer({ + signer: serverSigner, + relayHandler: hub.createRelayHandler(), + models: buildModels(), + policy: { default: { deny: ["haiku"] } }, // haiku hidden by default + adminPubkeys: [adminPk], + onPolicyMutate: (s) => mutations.push(s), + }); + + const admin = await connectClient(hub, serverPk, "admin", adminSigner); + adminClient = admin.client; + const plain = await connectClient(hub, serverPk, "plain"); + plainClient = plain.client; + plainPk = plain.pk; +}); + +afterAll(async () => { + await adminClient.close(); + await plainClient.close(); + await server.close(); +}); + +describe("models.list — search / limit", () => { + test("{} returns all policy-visible models", async () => { + const res = await call(plainClient, METHODS.modelsList, {}); + // default policy denies haiku → 3 remain + expect(ids(res.structuredContent)).toEqual([ + "anthropic/claude-opus", + "openai/gpt-4o", + "openai/gpt-4o-mini", + ]); + }); + + test("search narrows by provider substring", async () => { + const res = await call(plainClient, METHODS.modelsList, { + search: "openai", + }); + expect(ids(res.structuredContent)).toEqual([ + "openai/gpt-4o", + "openai/gpt-4o-mini", + ]); + }); + + test("search narrows by model substring across providers", async () => { + const res = await call(plainClient, METHODS.modelsList, { search: "gpt" }); + expect(ids(res.structuredContent)).toEqual([ + "openai/gpt-4o", + "openai/gpt-4o-mini", + ]); + }); + + test("limit caps the count", async () => { + const res = await call(plainClient, METHODS.modelsList, { limit: 1 }); + expect(ids(res.structuredContent)).toHaveLength(1); + }); +}); + +describe("chat.complete — hidden models are unknown", () => { + test("an allowed model completes", async () => { + const res = await call(plainClient, METHODS.chatComplete, { + model: "anthropic/claude-opus", + messages: [{ role: "user", content: "hi" }], + }); + expect(res.isError).toBeFalsy(); + const sc = res.structuredContent as { + choices: { message: { content: string } }[]; + }; + expect(sc.choices[0]!.message.content).toBe("anthropic reply"); + }); + + test("a policy-hidden model errors as 'unknown'", async () => { + const res = await call(plainClient, METHODS.chatComplete, { + model: "anthropic/claude-haiku", // denied by default policy + messages: [{ role: "user", content: "hi" }], + }); + expect(res.isError).toBe(true); + expect(JSON.stringify(res.content)).toContain("Unknown model"); + }); +}); + +describe("admin.models.* — gated by caller pubkey", () => { + test("non-admin caller is rejected", async () => { + const res = await call(plainClient, METHODS.adminModelsGetPolicy, {}); + expect(res.isError).toBe(true); + expect(JSON.stringify(res.content)).toContain("unauthorized"); + }); + + test("admin caller reads the current policy", async () => { + const res = await call(adminClient, METHODS.adminModelsGetPolicy, {}); + expect(res.isError).toBeFalsy(); + expect(res.structuredContent).toEqual({ default: { deny: ["haiku"] } }); + }); + + test("admin patches (null clears), mutation persists, effect is live", async () => { + const before = mutations.length; + // Clear the deny → all 4 models visible to plain client. + const res = await call(adminClient, METHODS.adminModelsSetPolicy, { + default: { deny: null }, + }); + expect(res.isError).toBeFalsy(); + expect(res.structuredContent).toEqual({ default: {} }); + expect(mutations.length).toBe(before + 1); + expect(mutations.at(-1)).toEqual({ default: {} }); + + const list = await call(plainClient, METHODS.modelsList, {}); + expect(ids(list.structuredContent)).toHaveLength(4); + }); + + test("per-client override scopes one caller without touching others", async () => { + // Restrict the plain client to gpt only; admin (no override) still sees all. + await call(adminClient, METHODS.adminModelsSetPolicy, { + byPubkey: { [plainPk]: { allow: ["gpt"] } }, + }); + const plainList = await call(plainClient, METHODS.modelsList, {}); + expect(ids(plainList.structuredContent)).toEqual([ + "openai/gpt-4o", + "openai/gpt-4o-mini", + ]); + const adminList = await call(adminClient, METHODS.modelsList, {}); + expect(ids(adminList.structuredContent)).toHaveLength(4); + + // Remove the override → plain client falls back to default (all 4). + await call(adminClient, METHODS.adminModelsSetPolicy, { + byPubkey: { [plainPk]: null }, + }); + const restored = await call(plainClient, METHODS.modelsList, {}); + expect(ids(restored.structuredContent)).toHaveLength(4); + }); +});