diff --git a/apps/e2e/tests/spacetimedb/internal-tool-ingest.test.ts b/apps/e2e/tests/spacetimedb/internal-tool-ingest.test.ts new file mode 100644 index 000000000..a2e78c4e3 --- /dev/null +++ b/apps/e2e/tests/spacetimedb/internal-tool-ingest.test.ts @@ -0,0 +1,220 @@ +// Tests the internal tool's /api/backend/* ingest routes over HTTP — the layer +// between the Stack Auth backend loggers and SpacetimeDB. Focus: the +// JSON-string payload fields (innerToolCallsJson, stepsJson, ...) must be +// rejected at ingest when malformed, because the review UIs JSON.parse them +// with an empty-list fallback — a corrupted value stored here would silently +// render as "no tool calls" instead of failing anywhere visible. +// +// Auth: these routes require a backend assertion — a JWT signed with the Stack +// Auth project keys derived from STACK_SERVER_SECRET. We sign one exactly like +// apps/backend/src/lib/ai/internal-tool-client.ts does, using the committed +// dev secret (same one apps/backend/.env.development uses; dev-only, not a +// real secret) — mirroring how helpers.ts signs member tokens with the +// committed dev JWK. + +import { signJWT } from "@hexclave/shared/dist/utils/jwt"; +import { afterEach, beforeEach, describe } from "vitest"; +import { it } from "../helpers"; +import { createCleanupScope, isSpacetimedbReachable, type CleanupScope } from "./helpers"; + +process.env.STACK_SERVER_SECRET ??= "23-wuNpik0gIW4mruTz25rbIvhuuvZFrLOLtL7J4tyo"; + +const ASSERTION_SUBJECT = "__internal_tool_backend__"; +const ASSERTION_TOKEN_USE = "internal-tool-backend"; +const INTERNAL_TOOL_PROJECT_ID = "internal"; + +function portPrefix(): string { + return process.env.NEXT_PUBLIC_HEXCLAVE_PORT_PREFIX ?? "81"; +} + +function internalToolBase(): string { + return (process.env.HEXCLAVE_INTERNAL_TOOL_URL ?? `http://localhost:${portPrefix()}41`).replace(/\/+$/, ""); +} + +function backendApiUrl(): string { + return (process.env.HEXCLAVE_BACKEND_BASE_URL ?? `http://localhost:${portPrefix()}02`).replace(/\/+$/, ""); +} + +async function isReachable(url: string): Promise { + try { + const res = await fetch(url, { signal: AbortSignal.timeout(2000) }); + return res.ok; + } catch (err) { + const isAbort = err instanceof DOMException && (err.name === "AbortError" || err.name === "TimeoutError"); + const isNetwork = err instanceof TypeError; + if (isAbort || isNetwork) return false; + throw err; + } +} + +async function isInternalToolReachable(): Promise { + return await isReachable(`${internalToolBase()}/.well-known/openid-configuration`); +} + +// The ingest routes verify our assertion against the backend's public JWKS +// endpoint, so a running internal tool alone isn't enough — without the +// backend every request 401s, which would read as a test failure rather than +// an unavailable environment. Probe the exact JWKS URL the tool fetches. +async function isBackendJwksReachable(): Promise { + return await isReachable(`${backendApiUrl()}/api/v1/projects/${INTERNAL_TOOL_PROJECT_ID}/.well-known/jwks.json`); +} + +async function mintBackendAssertion(): Promise { + return await signJWT({ + issuer: `${backendApiUrl()}/api/v1/projects/${INTERNAL_TOOL_PROJECT_ID}`, + audience: INTERNAL_TOOL_PROJECT_ID, + expirationTime: "5m", + payload: { + sub: ASSERTION_SUBJECT, + token_use: ASSERTION_TOKEN_USE, + }, + }); +} + +async function postIngest(path: string, body: unknown): Promise<{ status: number, body: string }> { + const res = await fetch(`${internalToolBase()}${path}`, { + method: "POST", + headers: { + "Authorization": `Bearer ${await mintBackendAssertion()}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(body), + }); + return { status: res.status, body: await res.text() }; +} + +function uniqueMarker(prefix: string): string { + return `${prefix}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; +} + +function validMcpCallBody(question: string) { + return { + correlationId: crypto.randomUUID(), + toolName: "e2e-ingest-tool", + reason: "e2e ingest validation", + userPrompt: "prompt", + question, + response: "response", + stepCount: 0, + innerToolCallsJson: "[]", + durationMs: 0, + modelId: "e2e-model", + }; +} + +function validAiQueryBody(correlationId: string) { + return { + correlationId, + mode: "generate", + systemPromptId: "e2e", + quality: "unknown", + speed: "unknown", + modelId: "e2e-model", + isAuthenticated: false, + requestedToolsJson: "[]", + messagesJson: "[]", + stepsJson: "[]", + finalText: "", + stepCount: 0, + durationMs: 0, + }; +} + +const canRun = await isInternalToolReachable() && await isBackendJwksReachable() && await isSpacetimedbReachable(); + +describe.skipIf(!canRun)("internal tool ingest validation", () => { + let scope: CleanupScope; + beforeEach(() => { + scope = createCleanupScope(); + }); + afterEach(async () => { + await scope.cleanup(); + }); + + it("log-mcp-call rejects innerToolCallsJson that is not valid JSON", async ({ expect }) => { + const res = await postIngest("/api/backend/log-mcp-call", { + ...validMcpCallBody(uniqueMarker("ingest-invalid-json")), + innerToolCallsJson: "not json at all {", + }); + expect(res.status).toBe(400); + expect(JSON.parse(res.body)).toMatchInlineSnapshot(` + { + "error": "Invalid request body.", + "issues": [ + { + "message": "Must be a valid JSON string.", + "path": "innerToolCallsJson", + }, + ], + } + `); + }); + + it("log-mcp-call rejects innerToolCallsJson that is valid JSON but not an array", async ({ expect }) => { + const res = await postIngest("/api/backend/log-mcp-call", { + ...validMcpCallBody(uniqueMarker("ingest-non-array")), + innerToolCallsJson: '{"toolName":"x"}', + }); + expect(res.status).toBe(400); + expect(JSON.parse(res.body)).toMatchInlineSnapshot(` + { + "error": "Invalid request body.", + "issues": [ + { + "message": "Must be a JSON array.", + "path": "innerToolCallsJson", + }, + ], + } + `); + }); + + it("log-mcp-call accepts a well-formed JSON array payload", async ({ expect }) => { + const marker = uniqueMarker("ingest-valid"); + scope.trackMcpQuestion(marker); + const res = await postIngest("/api/backend/log-mcp-call", validMcpCallBody(marker)); + expect(res.status).toBe(200); + expect(JSON.parse(res.body)).toMatchInlineSnapshot(` + { + "success": true, + } + `); + }); + + it("log-ai-query rejects malformed JSON payload fields", async ({ expect }) => { + const correlationId = crypto.randomUUID(); + const res = await postIngest("/api/backend/log-ai-query", { + ...validAiQueryBody(correlationId), + stepsJson: "[unterminated", + messagesJson: "42", + }); + expect(res.status).toBe(400); + expect(JSON.parse(res.body)).toMatchInlineSnapshot(` + { + "error": "Invalid request body.", + "issues": [ + { + "message": "Must be a JSON array.", + "path": "messagesJson", + }, + { + "message": "Must be a valid JSON string.", + "path": "stepsJson", + }, + ], + } + `); + }); + + it("log-ai-query accepts well-formed JSON array payloads", async ({ expect }) => { + const correlationId = crypto.randomUUID(); + scope.trackAiQueryCorrelationId(correlationId); + const res = await postIngest("/api/backend/log-ai-query", validAiQueryBody(correlationId)); + expect(res.status).toBe(200); + expect(JSON.parse(res.body)).toMatchInlineSnapshot(` + { + "success": true, + } + `); + }); +}); diff --git a/apps/internal-tool/spacetimedb/src/index.ts b/apps/internal-tool/spacetimedb/src/index.ts index 66837f069..874716ba5 100644 --- a/apps/internal-tool/spacetimedb/src/index.ts +++ b/apps/internal-tool/spacetimedb/src/index.ts @@ -1,5 +1,5 @@ import { schema, t, table, SenderError } from 'spacetimedb/server'; -import { Timestamp, type Identity } from 'spacetimedb'; +import { ScheduleAt, Timestamp, type Identity } from 'spacetimedb'; // Injected at publish time by scripts/spacetime-auth-config.mjs (non-secret). // SpacetimeDB validates the JWT signature via OIDC discovery on the token's @@ -14,6 +14,7 @@ const EXPECTED_AUDIENCE = '__SPACETIMEDB_EXPECTED_AUDIENCE__'; // Fallback session lifetime when a JWT carries no `exp` claim. Stack Auth // access tokens always carry one (~10min), so this is defensive only. const FALLBACK_SESSION_TTL_MICROS = 15n * 60n * 1000n * 1000n; +const SESSION_GC_INTERVAL_MICROS = 60n * 1000n * 1000n; type SenderAuthLike = Readonly<{ isInternal: boolean, @@ -87,6 +88,11 @@ function removeExpiredSessions(ctx: { } } +// Presence of a (non-expired) session row IS the authorization. This can't +// check `expiresAt` itself because view callbacks have no timestamp; instead, +// expired rows are actively deleted — on every connect/`touch_session` via +// `removeExpiredSessions`, and at most SESSION_GC_INTERVAL_MICROS later by the +// scheduled `session_gc` reducer — which re-evaluates the dependent views. function hasMemberSession(ctx: { sender: Identity, db: { @@ -181,7 +187,9 @@ const aiQueryLog = table( // each connecting identity to the grants its token carried. Self-enrolled // only — there is no reducer that can write to this table on behalf of // someone else. Rows expire with the token's `exp` and are garbage-collected -// opportunistically on subsequent connects. +// opportunistically on subsequent connects, plus periodically by the +// scheduled `session_gc` reducer so expired rows can't outlive their token +// just because nobody reconnects. const sessions = table( { name: 'sessions', public: false }, { @@ -192,6 +200,21 @@ const sessions = table( } ); +// Drives the periodic `session_gc` sweep (see SESSION_GC_INTERVAL_MICROS). +// Holds exactly one repeating-interval row, inserted by +// `ensureSessionGcScheduled`. The thunk's `any` return type breaks a type +// cycle the SDK can't express: the table references the reducer via +// `scheduled`, while the reducer's arg type references this table's rowType. +// A wrong reference would still fail loudly at publish time — the SDK +// validates that `scheduled` resolves to an exported reducer. +const sessionGcSchedule = table( + { name: 'session_gc_schedule', public: false, scheduled: (): any => session_gc }, + { + id: t.u64().primaryKey().autoInc(), + scheduledAt: t.scheduleAt(), + } +); + const qaEntries = table( { name: 'qa_entries', public: false }, { @@ -211,12 +234,35 @@ const qaEntries = table( } ); -const spacetimedb = schema({ mcpCallLog, aiQueryLog, sessions, qaEntries }); +const spacetimedb = schema({ mcpCallLog, aiQueryLog, sessions, sessionGcSchedule, qaEntries }); export default spacetimedb; type SessionRow = typeof sessions.rowType.type; +type SessionGcScheduleRow = typeof sessionGcSchedule.rowType.type; -type SessionCtx = { +type SessionGcScheduleCtx = { + db: { + sessionGcSchedule: { + iter: () => Iterable, + insert: (row: SessionGcScheduleRow) => void, + }, + }, +}; + +// Idempotently arms the repeating GC schedule. Called from `init` (fresh +// databases) AND from `upsertSessionFromJwt` (every connect/`touch_session`): +// `init` does not re-run when an existing database gets a module update, so +// already-deployed databases only pick up the schedule row through the +// connect path. +function ensureSessionGcScheduled(ctx: SessionGcScheduleCtx): void { + for (const _row of ctx.db.sessionGcSchedule.iter()) return; + ctx.db.sessionGcSchedule.insert({ + id: 0n, + scheduledAt: ScheduleAt.interval(SESSION_GC_INTERVAL_MICROS), + }); +} + +type SessionCtx = SessionGcScheduleCtx & { sender: Identity, timestamp: Timestamp, senderAuth: SenderAuthLike, @@ -234,6 +280,7 @@ type SessionCtx = { }; function upsertSessionFromJwt(ctx: SessionCtx): void { + ensureSessionGcScheduled(ctx); removeExpiredSessions(ctx); const jwt = ctx.senderAuth.jwt; // Tokenless clients may connect — they can only see `published_qa`. @@ -274,6 +321,19 @@ export const touch_session = spacetimedb.reducer({}, (ctx, _args) => { upsertSessionFromJwt(ctx); }); +// Scheduled sweep of expired session rows (see sessionGcSchedule). Without it, +// a long-lived WebSocket subscriber whose token expired would keep passing +// `hasMemberSession` indefinitely, since the opportunistic GC in +// `upsertSessionFromJwt` only runs when someone (re)connects. Scheduled +// reducers are private in SpacetimeDB v2 (only the scheduler and the database +// owner can invoke them), and the sweep is idempotent anyway. +export const session_gc = spacetimedb.reducer( + { row: sessionGcSchedule.rowType }, + (ctx, _args) => { + removeExpiredSessions(ctx); + } +); + export const myVisibleMcpCallLog = spacetimedb.view( { name: 'my_visible_mcp_call_log', public: true }, t.array(mcpCallLog.rowType), @@ -733,4 +793,6 @@ export const delete_ai_query_log = spacetimedb.reducer( } ); -export const init = spacetimedb.init(_ctx => {}); +export const init = spacetimedb.init(ctx => { + ensureSessionGcScheduled(ctx); +}); diff --git a/apps/internal-tool/src/app/api/backend/log-ai-query/route.ts b/apps/internal-tool/src/app/api/backend/log-ai-query/route.ts index 4a82e47af..d6d9b6f22 100644 --- a/apps/internal-tool/src/app/api/backend/log-ai-query/route.ts +++ b/apps/internal-tool/src/app/api/backend/log-ai-query/route.ts @@ -1,5 +1,5 @@ import { requireBackendAssertion } from "@/lib/server/backend-auth"; -import { handleApiError, readJsonBody, successResponse } from "@/lib/server/route-utils"; +import { handleApiError, readJsonBody, successResponse, zJsonArrayString } from "@/lib/server/route-utils"; import { callReducerStrict, opt } from "@/lib/server/spacetimedb-client"; import { getServiceSpacetimeToken } from "@/lib/server/spacetimedb-token"; import { z } from "zod"; @@ -14,9 +14,9 @@ const bodySchema = z.object({ isAuthenticated: z.boolean(), projectId: z.string().optional(), userId: z.string().optional(), - requestedToolsJson: z.string(), - messagesJson: z.string(), - stepsJson: z.string(), + requestedToolsJson: zJsonArrayString, + messagesJson: zJsonArrayString, + stepsJson: zJsonArrayString, finalText: z.string(), inputTokens: z.number().int().nonnegative().optional(), outputTokens: z.number().int().nonnegative().optional(), diff --git a/apps/internal-tool/src/app/api/backend/log-mcp-call/route.ts b/apps/internal-tool/src/app/api/backend/log-mcp-call/route.ts index 59b3fda3e..c9b8df141 100644 --- a/apps/internal-tool/src/app/api/backend/log-mcp-call/route.ts +++ b/apps/internal-tool/src/app/api/backend/log-mcp-call/route.ts @@ -1,5 +1,5 @@ import { requireBackendAssertion } from "@/lib/server/backend-auth"; -import { handleApiError, readJsonBody, successResponse } from "@/lib/server/route-utils"; +import { handleApiError, readJsonBody, successResponse, zJsonArrayString } from "@/lib/server/route-utils"; import { callReducerStrict, opt } from "@/lib/server/spacetimedb-client"; import { getServiceSpacetimeToken } from "@/lib/server/spacetimedb-token"; import { z } from "zod"; @@ -13,7 +13,7 @@ const bodySchema = z.object({ question: z.string(), response: z.string(), stepCount: z.number().int().nonnegative(), - innerToolCallsJson: z.string(), + innerToolCallsJson: zJsonArrayString, durationMs: z.number().int().nonnegative(), modelId: z.string(), errorMessage: z.string().optional(), diff --git a/apps/internal-tool/src/lib/server/route-utils.ts b/apps/internal-tool/src/lib/server/route-utils.ts index 4b03cd163..93536fd1a 100644 --- a/apps/internal-tool/src/lib/server/route-utils.ts +++ b/apps/internal-tool/src/lib/server/route-utils.ts @@ -47,3 +47,16 @@ export function handleApiError(scope: string, err: unknown): Response { export function successResponse(): Response { return Response.json({ success: true }); } + +export const zJsonArrayString = z.string().superRefine((value, ctx) => { + let parsed: unknown; + try { + parsed = JSON.parse(value); + } catch { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "Must be a valid JSON string." }); + return; + } + if (!Array.isArray(parsed)) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "Must be a JSON array." }); + } +}); diff --git a/apps/internal-tool/src/lib/server/spacetimedb-token.ts b/apps/internal-tool/src/lib/server/spacetimedb-token.ts index 3a951b7b6..37bfc86cd 100644 --- a/apps/internal-tool/src/lib/server/spacetimedb-token.ts +++ b/apps/internal-tool/src/lib/server/spacetimedb-token.ts @@ -58,7 +58,13 @@ export async function signSpacetimeToken(options: { subject: string, expiresIn?: /** The public half of the signing key, served by the JWKS route. */ export function publicJwks(): { keys: jose.JWK[] } { - const { d: _privateScalar, ...publicJwk } = privateJwk(); + const jwk = privateJwk(); + if (jwk.kty !== "EC" || jwk.crv !== "P-256" || jwk.x == null || jwk.y == null) { + throw new HexclaveAssertionError("HEXCLAVE_SPACETIMEDB_SIGNING_KEY_JWK must be an EC P-256 JWK (kty, crv, x, y)."); + } + const publicJwk: jose.JWK = { kty: jwk.kty, crv: jwk.crv, x: jwk.x, y: jwk.y }; + if (jwk.kid != null) publicJwk.kid = jwk.kid; + if (jwk.alg != null) publicJwk.alg = jwk.alg; return { keys: [publicJwk] }; }