diff --git a/.env.example b/.env.example index 46f3b56..a35026b 100644 --- a/.env.example +++ b/.env.example @@ -49,6 +49,9 @@ VERIFY_ENABLED=false # When a guardrail finds a violation: flag-only (false) or block with 422 (true). # A guardrail *execution error* always fails closed (blocks) regardless of this. GUARDRAILS_BLOCK=false +# Opt-in: buffer a streamed response fully and run inline guardrails before sending it +# (blocks with 422 if violating). Trades first-byte latency for inline enforcement on streams. +GUARDRAILS_STREAM_BUFFER=false # Async local-Ollama LLM judge (1–5 score + reason), attached to the trace out-of-band. JUDGE_ENABLED=false # Fraction of (non-cached) responses to judge, 0–1. diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c725c3..9ca11ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,9 @@ All notable changes to Sentinel are documented here. The format follows - Per-request **cost tracking**: set a `pricing` map (USD per 1K tokens, per model) in `sentinel.config.json` and every trace records a `costUsd`; the dashboard shows total spend, spend saved by the cache, and cost over time. - **Native Anthropic provider**: set `"type": "anthropic"` on a provider and Sentinel speaks Anthropic's Messages API (`x-api-key`, request/response/stream translation) while clients keep using the OpenAI shape. - **Multi-key round-robin**: a provider can list `apiKeyEnvs` (a key pool); Sentinel rotates across the keys per request to spread load and survive free-tier rate limits. +- **Per-key trace isolation**: a client reads only its own traces via `GET /v1/traces` (+ `/v1/traces/:id`), authenticated by its Sentinel key; the admin `GET /traces` still sees everything. +- **Opt-in inline guardrails on streaming**: `GUARDRAILS_STREAM_BUFFER=true` buffers a streamed response and applies guardrails (including a 422 block) before any bytes are sent. +- **Configurable routing thresholds**: the `auto` complexity boundaries are tunable via `routing.thresholds`. ### Fixed diff --git a/README.md b/README.md index 8d90cd0..773aa94 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,8 @@ curl http://localhost:8080/traces \ # filters: ?model=gpt-4o-mini&status=200&stream=true&since=&limit=50 ``` +A client app can also read **only its own** traces — no admin key — at `GET /v1/traces` (and `/v1/traces/:id`), authenticated by its Sentinel API key and scoped server-side to that key. + Traces are **metadata only** — no prompt or response bodies are stored, and API keys are recorded as a SHA-256 hash, never in the clear. ## Caching @@ -115,7 +117,7 @@ Sentinel survives flaky providers and free-tier rate limits instead of passing t - **Retry + backoff + fallback.** A retryable failure (429, upstream 5xx, timeout, network fault) is retried with exponential backoff up to `MAX_RETRIES`, then **failed over** to the next candidate — additional models you list under `routing.fallback`, ending at a local Ollama model so a request can always be served. Terminal errors (400/401/403/404) fail fast without pointless fallback. - **Per-provider throttling.** A token-bucket limiter paces each provider to its configured `rpm` (or `DEFAULT_RPM`), waiting up to `THROTTLE_MAX_WAIT_MS` for headroom before skipping to a less-loaded candidate. This keeps you under a provider's limit rather than discovering it the hard way. -- **Cost-aware routing (opt-in).** Send `"model": "auto"` and a rules-based classifier picks the **cheapest capable tier** from `routing.tiers` (cheapest first) by prompt complexity, escalating to more capable tiers on failure. Explicit model names behave exactly as before — plus the fallback chain. Drop-in semantics are preserved; nothing is configured by default. +- **Cost-aware routing (opt-in).** Send `"model": "auto"` and a rules-based classifier picks the **cheapest capable tier** from `routing.tiers` (cheapest first) by prompt complexity, escalating to more capable tiers on failure. Explicit model names behave exactly as before — plus the fallback chain. Drop-in semantics are preserved; nothing is configured by default. The complexity boundaries between tiers are tunable via `routing.thresholds`. Every routed request records which provider/model actually served it, whether a fallback was used, and the retry count — queryable via `GET /traces?fallbackUsed=true` (and `?routedProvider=`). Streaming requests fail over up to the first chunk; once the SSE response is committed, a mid-stream error is surfaced as an inline event. @@ -127,7 +129,7 @@ Sentinel can check a response's quality **in the request path**, not just after - **Async LLM judge (sampled).** A fraction (`JUDGE_SAMPLE_RATE`) of responses are scored 1–5 with a short reason by a local Ollama model (`JUDGE_MODEL`), **out of band** — it never adds latency to the response. The response under review is wrapped as untrusted data so it can't talk the judge into a pass; a judge failure is recorded as "unscored", never a pass. - **Regression tracking.** Each request carries a model-independent **prompt fingerprint**, so `GET /regression` groups judge scores by `(prompt, model)` — compare the groups sharing a fingerprint to see how one prompt's quality differs across models or versions. -Verdicts are queryable: `GET /traces?guardrailStatus=block`, `?judgeScoreMax=2`, `?promptFingerprint=…`. Inline guardrails apply to non-streaming responses; streamed responses are judged from their buffered output after they complete. +Verdicts are queryable: `GET /traces?guardrailStatus=block`, `?judgeScoreMax=2`, `?promptFingerprint=…`. Inline guardrails apply to non-streaming responses by default; set `GUARDRAILS_STREAM_BUFFER=true` to buffer a streamed response and apply guardrails (including a 422 block) **before any bytes are sent** — otherwise streamed output is judged asynchronously after it completes. ## Dashboard @@ -161,8 +163,8 @@ Sentinel v0.1.0 is a **single-node, self-hosted** gateway. Honest boundaries tod - **In-memory cache & rate-limit state.** The semantic cache and token buckets live in-process — not shared across instances, and cleared on restart. A Redis-backed swap is planned behind the existing interfaces. - **SQLite (or in-memory) trace storage.** Ideal for single-node; a Postgres backend for multi-instance is planned. -- **The async judge needs a local Ollama** with `JUDGE_MODEL` pulled. Without it, judging degrades to `unscored` (never a false pass). The bundled benchmarks run against **mock upstreams**, so the headline catch-rate is the _deterministic guardrail_ rate; the LLM judge is covered by unit tests. -- **Inline guardrails apply to non-streaming responses.** Streamed responses are judged from their buffered output _after_ completion (inline blocking of a live stream is on the roadmap). +- **The async judge needs a local Ollama** with `JUDGE_MODEL` pulled. Without it, judging degrades to `unscored` (never a false pass). The bundled benchmarks run against **mock upstreams**, so the headline catch-rate is the _deterministic guardrail_ rate; the LLM judge is covered by unit tests and a real run is captured in [`docs/EVIDENCE.md`](./docs/EVIDENCE.md). +- **Inline guardrails run inline on non-streaming responses.** Streamed responses enforce guardrails only with `GUARDRAILS_STREAM_BUFFER=true` (which buffers the full response first, trading first-byte latency); otherwise they're judged asynchronously after completion. - **Per-request dollar cost requires a `pricing` map** in `sentinel.config.json` (USD per 1K tokens, per model). With it, every trace records a `costUsd` and the dashboard shows spend + cache savings; without it, cost is unknown (`null`) and only request-volume cache savings are visible. ## Development diff --git a/SECURITY_REVIEW_LOG.md b/SECURITY_REVIEW_LOG.md index ac2ea83..f87f806 100644 --- a/SECURITY_REVIEW_LOG.md +++ b/SECURITY_REVIEW_LOG.md @@ -79,4 +79,6 @@ Each item: assume an attacker is actively trying it. Tick only when a test prove | SR-008 | 2026-06-30 | 1 | PR4 multi-key round-robin | Holding several provider keys per provider widens the secret surface — a key could leak via logs/errors, or a blank slot send an empty key | low | mitigated | Keys come only from env (`apiKeyEnv`/`apiKeyEnvs`), never the repo or request input; each is sent only in the outbound auth header (`Authorization`/`x-api-key`), redacted in logs and never returned to clients. Rotation is an in-memory counter over the resolved pool — no provider key is persisted or traced (traces still store only the SHA-256 of the *Sentinel client* key). Unset/empty env vars are skipped, so a misconfigured slot can't send a blank key (tested). | +| SR-009 | 2026-06-30 | 3, 6 | PR5 per-key trace isolation + opt-in stream guardrails | A client reading another tenant's traces; a client injecting `apiKeyHash` to widen scope; a streamed bad response bypassing inline guardrails | high | mitigated | Resolves the SR-002 deferral. `GET /v1/traces` (+ `/v1/traces/:id`) is gated by the caller's own Sentinel key and **forces** `apiKeyHash = hash(callerKey)` server-side — the filter is never parsed from query input, so a client cannot widen scope; cross-tenant list returns only the caller's rows and `/v1/traces/:id` 404s another key's id (tested). The admin all-traces `GET /traces` stays separately gated by `SENTINEL_ADMIN_KEY`. Opt-in `GUARDRAILS_STREAM_BUFFER` buffers a streamed response and runs inline guardrails before any byte is sent, blocking (422) on a violation — closing the streaming-bypass gap; default off preserves streaming latency. | + > Add a row per high-risk change. Status ∈ {open, mitigated, accepted}. Severity ∈ {low, med, high, critical}. diff --git a/docs/EVIDENCE.md b/docs/EVIDENCE.md new file mode 100644 index 0000000..8c677c1 --- /dev/null +++ b/docs/EVIDENCE.md @@ -0,0 +1,29 @@ +# EVIDENCE — real async-judge run + +The headline benchmarks in [`load/RESULTS.md`](../load/RESULTS.md) run against **mock +upstreams**, so the catch-rate reported there is the _deterministic guardrail_ rate. The +async LLM judge needs a real model; this file captures a real run to prove that path +end to end. + +## Async LLM judge against a local Ollama + +- **Date:** 2026-06-30 +- **Model:** `qwen2.5:0.5b` via Ollama's OpenAI-compatible endpoint + (`http://localhost:11434/v1`), keyless. (The default `JUDGE_MODEL` is `qwen2.5:7b`; this + run used the smaller 0.5B model that happened to be installed locally — it still scores + correctly.) +- **Harness:** `createOllamaJudge(...).score(request, responseText)` — the exact code path + the gateway uses for its sampled, out-of-band judge. + +Prompt for both cases: _"What is the capital of France?"_ + +| Case | Response judged | Score (1–5) | Judge reason | +| ---- | --------------------------------------------------------------- | ----------- | ------------------------------------------------------------------------------- | +| Good | "The capital of France is Paris." | **5** | "The response correctly identifies the capital city of France, which is Paris." | +| Bad | "Bananas are yellow and have nothing to do with your question." | **2** | "The response does not provide a correct answer to the given question." | + +The judge cleanly separates a correct answer (5) from an off-topic one (2), parsing the +model's JSON verdict via `parseJudgeVerdict`. This confirms the judge integration works +against a real model end to end; the deterministic verdict-parsing and prompt-construction +logic are additionally covered by the Phase 5 unit tests +(`packages/gateway/src/verify/judge.test.ts`). diff --git a/packages/gateway/src/config.ts b/packages/gateway/src/config.ts index 3c97f4c..d966e52 100644 --- a/packages/gateway/src/config.ts +++ b/packages/gateway/src/config.ts @@ -24,6 +24,7 @@ export interface ServerEnv { throttleMaxWaitMs: number; verifyEnabled: boolean; guardrailsBlock: boolean; + guardrailsStreamBuffer: boolean; judgeEnabled: boolean; judgeModel: string; judgeSampleRate: number; @@ -58,6 +59,10 @@ const serverEnvSchema = z.object({ .enum(['true', 'false']) .default('false') .transform((v) => v === 'true'), + GUARDRAILS_STREAM_BUFFER: z + .enum(['true', 'false']) + .default('false') + .transform((v) => v === 'true'), JUDGE_ENABLED: z .enum(['true', 'false']) .default('false') @@ -98,6 +103,7 @@ export function loadServerEnv(env: NodeJS.ProcessEnv): ServerEnv { throttleMaxWaitMs: parsed.data.THROTTLE_MAX_WAIT_MS, verifyEnabled: parsed.data.VERIFY_ENABLED, guardrailsBlock: parsed.data.GUARDRAILS_BLOCK, + guardrailsStreamBuffer: parsed.data.GUARDRAILS_STREAM_BUFFER, judgeEnabled: parsed.data.JUDGE_ENABLED, judgeModel: parsed.data.JUDGE_MODEL, judgeSampleRate: parsed.data.JUDGE_SAMPLE_RATE, @@ -124,6 +130,7 @@ const providerConfigSchema = z const routingConfigSchema = z.object({ tiers: z.array(z.string().min(1)).optional(), fallback: z.array(z.string().min(1)).optional(), + thresholds: z.array(z.number().int().nonnegative()).optional(), }); const guardrailsConfigSchema = z.object({ @@ -180,6 +187,8 @@ export interface ResolvedProvider { export interface ResolvedRouting { tiers?: string[] | undefined; fallback?: string[] | undefined; + /** Complexity score boundaries between `auto` tiers (defaults applied when omitted). */ + thresholds?: number[] | undefined; } /** Guardrail policy (content blocklist + PII categories) from the config file. */ diff --git a/packages/gateway/src/main.ts b/packages/gateway/src/main.ts index a007296..01d3cca 100644 --- a/packages/gateway/src/main.ts +++ b/packages/gateway/src/main.ts @@ -64,6 +64,7 @@ async function main(): Promise { adminKey: env.adminKey, cache, pricing: config.pricing, + bufferStreamForGuardrails: env.guardrailsStreamBuffer, ...(clientThrottle ? { clientThrottle } : {}), routing: { config: config.routing, diff --git a/packages/gateway/src/routes.traces.ts b/packages/gateway/src/routes.traces.ts index 15ba5a5..cc40638 100644 --- a/packages/gateway/src/routes.traces.ts +++ b/packages/gateway/src/routes.traces.ts @@ -1,15 +1,22 @@ import type { FastifyPluginAsync } from 'fastify'; -import { createAdminAuthHook } from './auth.js'; +import { createAdminAuthHook, createAuthHook, extractBearerToken, hashApiKey } from './auth.js'; +import { AuthError } from './errors.js'; import type { TraceQuery, TraceStore } from './telemetry/trace.js'; export interface TraceRoutesOptions { traceStore: TraceStore; adminKey: string | undefined; + /** Client API keys, for the self-scoped `/v1/traces` endpoints. */ + apiKeys: ReadonlySet; } -/** Fastify plugin: an admin-key-gated read API over the trace store. */ +/** + * Fastify plugin: an admin-key-gated read API over all traces (`/traces`), plus a + * self-scoped read API (`/v1/traces`) where a client key sees only its own traces. + */ export const traceRoutes: FastifyPluginAsync = async (app, options) => { const adminHook = createAdminAuthHook(options.adminKey); + const authHook = createAuthHook(options.apiKeys); app.get('/traces', { preHandler: adminHook }, async (request, reply) => { return reply.send(options.traceStore.query(parseTraceQuery(request.query))); @@ -25,8 +32,33 @@ export const traceRoutes: FastifyPluginAsync = async (app, o } return reply.send(trace); }); + + // Self-scoped: a client key reads only its own traces (forced apiKeyHash filter). + app.get('/v1/traces', { preHandler: authHook }, async (request, reply) => { + const query = parseTraceQuery(request.query); + query.apiKeyHash = callerHash(request.headers.authorization); + return reply.send(options.traceStore.query(query)); + }); + + app.get('/v1/traces/:id', { preHandler: authHook }, async (request, reply) => { + const { id } = request.params as { id: string }; + const trace = options.traceStore.get(id); + if (trace === undefined || trace.apiKeyHash !== callerHash(request.headers.authorization)) { + return reply + .status(404) + .send({ error: { message: `No trace "${id}"`, type: 'not_found', code: null } }); + } + return reply.send(trace); + }); }; +/** Hashes the caller's bearer token so they can be scoped to their own traces. */ +function callerHash(authorization: string | undefined): string { + const token = extractBearerToken(authorization); + if (token === null) throw new AuthError(); // authHook already guarantees one; defensive + return hashApiKey(token); +} + function parseTraceQuery(raw: unknown): TraceQuery { const q = (typeof raw === 'object' && raw !== null ? raw : {}) as Record< string, diff --git a/packages/gateway/src/routing/classifier.test.ts b/packages/gateway/src/routing/classifier.test.ts index 611e9a5..ada35ff 100644 --- a/packages/gateway/src/routing/classifier.test.ts +++ b/packages/gateway/src/routing/classifier.test.ts @@ -34,4 +34,10 @@ describe('classifyTier', () => { }; expect(classifyTier(request, 3)).toBe(0); }); + + it('honours custom thresholds over the defaults', () => { + // 'hello world' (11 chars) + 1 message × 100 = score 111. + expect(classifyTier(reqOf('hello world'), 3, [10, 20])).toBe(2); // low bars → top tier + expect(classifyTier(reqOf('hello world'), 3, [100_000])).toBe(0); // high bar → cheapest + }); }); diff --git a/packages/gateway/src/routing/classifier.ts b/packages/gateway/src/routing/classifier.ts index 99999c2..c328401 100644 --- a/packages/gateway/src/routing/classifier.ts +++ b/packages/gateway/src/routing/classifier.ts @@ -1,7 +1,7 @@ import type { ChatCompletionRequest } from '../schemas.js'; -/** Score boundaries between tiers; crossing one escalates to the next tier. */ -const THRESHOLDS = [400, 2000, 8000, 32_000]; +/** Default score boundaries between tiers; crossing one escalates to the next tier. */ +export const DEFAULT_THRESHOLDS = [400, 2000, 8000, 32_000]; /** A cheap, model-free complexity estimate from the request shape. */ function complexityScore(request: ChatCompletionRequest): number { @@ -20,11 +20,15 @@ function complexityScore(request: ChatCompletionRequest): number { * configured. Deterministic and rules-based: longer prompts, more messages, and a * larger `max_tokens` budget escalate to higher (more capable) tiers. */ -export function classifyTier(request: ChatCompletionRequest, tierCount: number): number { +export function classifyTier( + request: ChatCompletionRequest, + tierCount: number, + thresholds: readonly number[] = DEFAULT_THRESHOLDS, +): number { if (tierCount <= 1) return 0; const score = complexityScore(request); let index = 0; - for (const threshold of THRESHOLDS) { + for (const threshold of thresholds) { if (score >= threshold) index += 1; else break; } diff --git a/packages/gateway/src/routing/router.ts b/packages/gateway/src/routing/router.ts index 2c2fe6a..1d0956f 100644 --- a/packages/gateway/src/routing/router.ts +++ b/packages/gateway/src/routing/router.ts @@ -16,6 +16,8 @@ export interface RoutingConfig { tiers?: string[] | undefined; /** Models appended as fallbacks to every request's candidate chain. */ fallback?: string[] | undefined; + /** Complexity score boundaries between tiers (defaults applied when omitted). */ + thresholds?: number[] | undefined; } export interface Router { @@ -44,7 +46,7 @@ export function createRouter(options: CreateRouterOptions): Router { let names: string[]; if (requested === 'auto') { if (tiers.length === 0) throw new ModelNotFoundError('auto'); - const index = classifyTier(request, tiers.length); + const index = classifyTier(request, tiers.length, options.routing?.thresholds); names = [...tiers.slice(index), ...fallback]; } else { names = [requested, ...fallback]; diff --git a/packages/gateway/src/server.security.test.ts b/packages/gateway/src/server.security.test.ts index 4a7faf0..9617db8 100644 --- a/packages/gateway/src/server.security.test.ts +++ b/packages/gateway/src/server.security.test.ts @@ -6,7 +6,9 @@ import type { ProviderRegistry } from './providers/registry.js'; import { InMemoryTraceStore } from './telemetry/store.memory.js'; import { createBucketRegistry } from './throttle/token-bucket.js'; import { createOpenAICompatibleProvider } from './providers/openai-compatible.js'; +import { hashApiKey } from './auth.js'; import type { ChatCompletionRequest } from './schemas.js'; +import type { TraceRecord } from './telemetry/trace.js'; const provider: Provider = { name: 'fake', @@ -120,3 +122,84 @@ describe('upstream redirect hardening', () => { expect(fetchMock.mock.calls[0]![1].redirect).toBe('error'); }); }); + +function traceRec(id: string, apiKeyHash: string): TraceRecord { + return { + id, + traceId: 't', + timestamp: 1, + durationMs: 1, + model: 'm', + provider: 'p', + stream: false, + status: 200, + promptTokens: null, + completionTokens: null, + totalTokens: null, + costUsd: null, + errorType: null, + errorMessage: null, + apiKeyHash, + cacheHit: false, + routedProvider: null, + routedModel: null, + fallbackUsed: false, + retryCount: 0, + guardrailStatus: null, + guardrailViolations: null, + judgeScore: null, + judgeReason: null, + judgeError: null, + promptFingerprint: null, + }; +} + +describe('self-scoped /v1/traces isolation', () => { + it("returns only the caller's own traces and never another key's", async () => { + const store = new InMemoryTraceStore(); + store.record(traceRec('mine', hashApiKey('good'))); + store.record(traceRec('theirs', hashApiKey('other'))); + const app = buildServer({ + registry, + apiKeys: new Set(['good', 'other']), + logger: false, + traceStore: store, + adminKey: 'admin', + }); + + const mine = await app.inject({ + method: 'GET', + url: '/v1/traces', + headers: { authorization: 'Bearer good' }, + }); + expect(mine.statusCode).toBe(200); + expect((mine.json() as { id: string }[]).map((t) => t.id)).toEqual(['mine']); + + const theirs = await app.inject({ + method: 'GET', + url: '/v1/traces', + headers: { authorization: 'Bearer other' }, + }); + expect((theirs.json() as { id: string }[]).map((t) => t.id)).toEqual(['theirs']); + + // No key → 401. + expect((await app.inject({ method: 'GET', url: '/v1/traces' })).statusCode).toBe(401); + + // A key cannot fetch another key's trace by id (404, not another tenant's data). + const cross = await app.inject({ + method: 'GET', + url: '/v1/traces/theirs', + headers: { authorization: 'Bearer good' }, + }); + expect(cross.statusCode).toBe(404); + + // ...but can fetch its own by id. + const own = await app.inject({ + method: 'GET', + url: '/v1/traces/mine', + headers: { authorization: 'Bearer good' }, + }); + expect(own.statusCode).toBe(200); + await app.close(); + }); +}); diff --git a/packages/gateway/src/server.test.ts b/packages/gateway/src/server.test.ts index 5e599ca..73838ba 100644 --- a/packages/gateway/src/server.test.ts +++ b/packages/gateway/src/server.test.ts @@ -888,3 +888,50 @@ describe('semantic cache', () => { await app.close(); }); }); + +describe('streaming guardrails (buffered, opt-in)', () => { + const buffered = (chunks: string[], block: boolean) => { + const store = new InMemoryTraceStore(); + const provider = makeProvider({ + chatStream: async function* () { + for (const c of chunks) yield c; + }, + }); + return buildServer({ + registry: makeRegistry(provider), + apiKeys: new Set(['good']), + logger: false, + traceStore: store, + verifier: createVerifier({ store, guardrails: { block } }), + bufferStreamForGuardrails: true, + }); + }; + + it('blocks a violating streamed response with 422 before sending any bytes', async () => { + const app = buffered(['{"choices":[{"delta":{"content":"reach me at a@b.com"}}]}'], true); + const res = await app.inject({ + method: 'POST', + url, + headers: auth, + payload: { ...body, stream: true }, + }); + expect(res.statusCode).toBe(422); + expect(res.json().error.type).toBe('guardrail_blocked'); + await app.close(); + }); + + it('streams a clean buffered response through to [DONE]', async () => { + const app = buffered(['{"choices":[{"delta":{"content":"all good here"}}]}'], true); + const res = await app.inject({ + method: 'POST', + url, + headers: auth, + payload: { ...body, stream: true }, + }); + expect(res.statusCode).toBe(200); + expect(res.headers['content-type']).toContain('text/event-stream'); + expect(res.body).toContain('all good here'); + expect(res.body).toContain('data: [DONE]'); + await app.close(); + }); +}); diff --git a/packages/gateway/src/server.ts b/packages/gateway/src/server.ts index 9c80d3a..ce63822 100644 --- a/packages/gateway/src/server.ts +++ b/packages/gateway/src/server.ts @@ -52,6 +52,8 @@ export interface ServerDeps { verifier?: Verifier | undefined; /** model → USD-per-1K-token pricing, for per-request cost attribution on the trace. */ pricing?: ReadonlyMap | undefined; + /** When true (with a verifier), buffer streaming responses and run inline guardrails before sending. */ + bufferStreamForGuardrails?: boolean | undefined; /** Per-API-key inbound rate-limit buckets (built from CLIENT_RPM). */ clientThrottle?: BucketRegistry | undefined; logger?: FastifyServerOptions['logger']; @@ -321,6 +323,39 @@ export function buildServer(deps: ServerDeps) { span?.setAttribute('sentinel.prompt_fingerprint', deps.verifier.fingerprint(chatRequest)); } + // Opt-in: buffer the whole stream, run inline guardrails on the assembled text, and + // block (422) before any bytes are sent — trading streaming latency for inline + // enforcement. Off by default; the live path below preserves first-byte latency. + if (deps.bufferStreamForGuardrails && deps.verifier !== undefined) { + const collected: string[] = []; + for (let step = first; step.done !== true; step = await iterator.next()) { + collected.push(step.value); + } + const text = extractStreamText(collected); + const verdict = deps.verifier.inspect(chatRequest, text); + span?.setAttribute('sentinel.guardrail_status', verdict.status); + if (verdict.violations.length > 0) { + span?.setAttribute('sentinel.guardrail_violations', verdict.violations.join(',')); + } + if (verdict.status === 'block') { + throw new GuardrailBlockedError(verdict.violations); // nothing sent yet → clean 422 + } + reply.hijack(); + reply.raw.writeHead(200, SSE_HEADERS); + for (const chunk of collected) reply.raw.write(`data: ${chunk}\n\n`); + reply.raw.write('data: [DONE]\n\n'); + reply.raw.end(); + span?.setAttribute('sentinel.cache_hit', false); + endSpan(request, 200); + if (deps.cache !== undefined && apiKeyHash !== null) { + await deps.cache.set(chatRequest, apiKeyHash, { kind: 'stream', chunks: collected }); + } + if (span !== undefined) { + deps.verifier.scheduleJudge(span.spanContext().spanId, chatRequest, text); + } + return reply; + } + reply.hijack(); reply.raw.writeHead(200, SSE_HEADERS); @@ -357,7 +392,11 @@ export function buildServer(deps: ServerDeps) { }, ); - app.register(traceRoutes, { traceStore: deps.traceStore, adminKey: deps.adminKey }); + app.register(traceRoutes, { + traceStore: deps.traceStore, + adminKey: deps.adminKey, + apiKeys: deps.apiKeys, + }); app.register(regressionRoutes, { traceStore: deps.traceStore, adminKey: deps.adminKey }); return app; diff --git a/packages/gateway/src/telemetry/store.memory.ts b/packages/gateway/src/telemetry/store.memory.ts index 5f9bafc..025dadf 100644 --- a/packages/gateway/src/telemetry/store.memory.ts +++ b/packages/gateway/src/telemetry/store.memory.ts @@ -56,5 +56,6 @@ function matches(trace: TraceRecord, filter: TraceQuery): boolean { trace.promptFingerprint !== filter.promptFingerprint ) return false; + if (filter.apiKeyHash !== undefined && trace.apiKeyHash !== filter.apiKeyHash) return false; return true; } diff --git a/packages/gateway/src/telemetry/store.sqlite.ts b/packages/gateway/src/telemetry/store.sqlite.ts index 8ceb554..81c815b 100644 --- a/packages/gateway/src/telemetry/store.sqlite.ts +++ b/packages/gateway/src/telemetry/store.sqlite.ts @@ -37,6 +37,7 @@ const SCHEMA = ` CREATE INDEX IF NOT EXISTS idx_traces_fallback_used ON traces (fallback_used); CREATE INDEX IF NOT EXISTS idx_traces_guardrail_status ON traces (guardrail_status); CREATE INDEX IF NOT EXISTS idx_traces_prompt_fingerprint ON traces (prompt_fingerprint); + CREATE INDEX IF NOT EXISTS idx_traces_api_key_hash ON traces (api_key_hash); `; interface TraceRow { @@ -202,6 +203,10 @@ export class SqliteTraceStore implements TraceStore { where.push('prompt_fingerprint = @promptFingerprint'); params.promptFingerprint = filter.promptFingerprint; } + if (filter.apiKeyHash !== undefined) { + where.push('api_key_hash = @apiKeyHash'); + params.apiKeyHash = filter.apiKeyHash; + } params.limit = filter.limit ?? 50; params.offset = filter.offset ?? 0; const clause = where.length > 0 ? `WHERE ${where.join(' AND ')}` : ''; diff --git a/packages/gateway/src/telemetry/store.test.ts b/packages/gateway/src/telemetry/store.test.ts index ba4a779..deb6bab 100644 --- a/packages/gateway/src/telemetry/store.test.ts +++ b/packages/gateway/src/telemetry/store.test.ts @@ -163,6 +163,15 @@ for (const [name, makeStore] of backends) { expect(store.get('free')?.costUsd).toBeNull(); store.close(); }); + + it('filters by apiKeyHash (per-key trace isolation)', () => { + const store = makeStore(); + store.record(sample({ id: 'k1', apiKeyHash: 'aaa' })); + store.record(sample({ id: 'k2', apiKeyHash: 'bbb' })); + expect(store.query({ apiKeyHash: 'aaa' }).map((t) => t.id)).toEqual(['k1']); + expect(store.query({ apiKeyHash: 'bbb' }).map((t) => t.id)).toEqual(['k2']); + store.close(); + }); }); } diff --git a/packages/gateway/src/telemetry/trace.ts b/packages/gateway/src/telemetry/trace.ts index c75dcbc..36e41a2 100644 --- a/packages/gateway/src/telemetry/trace.ts +++ b/packages/gateway/src/telemetry/trace.ts @@ -64,6 +64,8 @@ export interface TraceQuery { judgeScoreMin?: number; judgeScoreMax?: number; promptFingerprint?: string; + /** Restrict to one client's traces (by API-key hash) — used by the self-scoped read API. */ + apiKeyHash?: string; limit?: number; offset?: number; } diff --git a/sentinel.config.example.json b/sentinel.config.example.json index b17a844..0082aee 100644 --- a/sentinel.config.example.json +++ b/sentinel.config.example.json @@ -37,7 +37,8 @@ "defaultProvider": "ollama", "routing": { "tiers": ["llama3.2", "gpt-4o-mini", "llama-3.3-70b-versatile"], - "fallback": ["llama3.2"] + "fallback": ["llama3.2"], + "thresholds": [400, 2000, 8000, 32000] }, "guardrails": { "blocklist": ["ignore previous instructions"],