Add internal tool ingest tests for API routes

- Introduced a new test suite for the internal tool's ingest routes, validating the handling of malformed JSON payloads for `log-mcp-call` and `log-ai-query`.
- Ensured that invalid JSON structures are correctly rejected with appropriate error messages.
- Verified that well-formed JSON arrays are accepted, enhancing the robustness of the ingest functionality.
- Updated route utilities to include a new validation function for JSON array strings, improving error handling across the API.

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<boolean> {
+  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<boolean> {
+  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<boolean> {
+  return await isReachable(`${backendApiUrl()}/api/v1/projects/${INTERNAL_TOOL_PROJECT_ID}/.well-known/jwks.json`);
+}
+
+async function mintBackendAssertion(): Promise<string> {
+  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,
+      }
+    `);
+  });
+});

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<SessionGcScheduleRow>,
+      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);
+});

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(),

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(),

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." });
+  }
+});

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] };
 }