feat: add protected event ingest foundation agent:clawd

README.md

@@ -30,7 +30,9 @@ SignalOps is a React dashboard for operating AI image-generation products. It is
 - Recharts
 - Lucide React
 
-## Run Locally
+## Run locally
+
+Requires Node.js 24 or newer and pnpm 10.24.0, matching CI and the native TypeScript contract checks.
 
 ```bash
 pnpm install
@@ -39,6 +41,14 @@ pnpm dev
 
 The dev server is pinned to [http://localhost:3020](http://localhost:3020) to avoid colliding with other local portfolio/product apps.
 
+## Event API boundaries
+
+- `POST /api/events/validate` is public and verification-only. It normalizes and redacts events but stores nothing.
+- `POST /api/events` is a protected development/test ingest seam. Set `SIGNALOPS_INGEST_TOKEN` and send that exact value using the Bearer authorization scheme.
+- Accepted local events always use the validator's bounded JSON, batch, normalization, and redaction contract. Exact retries return the same receipt reference and do not duplicate retained events.
+- The local memory sink retains at most 1,000 events and can evict the oldest retained event. Receipts expose stored, duplicate, evicted, and retained counts so this behavior is not presented as durability.
+- Production intentionally returns `503 storage_not_ready` after authentication because this slice does not configure a durable store. The endpoint is not production-ready until a separate durable sink is implemented and explicitly wired.
+
 ## Demo script
 
 For a fast portfolio review, the dashboard opens with a **Guided incident replay** rail directly under the header. It turns the surface into a self-explaining demo — a first-run reviewer can follow one incident end to end in well under 90 seconds.

package.json

@@ -2,6 +2,10 @@
   "name": "signalops",
   "version": "0.1.0",
   "private": true,
+  "engines": {
+    "node": ">=24"
+  },
+  "packageManager": "pnpm@10.24.0",
   "scripts": {
     "dev": "next dev -p 3020",
     "dev:any": "next dev",
@@ -10,7 +14,9 @@
     "lint": "eslint",
     "typecheck": "tsc --noEmit --pretty false",
     "test:home-navigation": "node scripts/test-home-navigation.mjs",
-    "check": "pnpm lint && pnpm typecheck && pnpm test:home-navigation && pnpm build"
+    "test:events": "node scripts/test-signalops-events.mjs",
+    "test:ingest": "node scripts/test-signalops-ingest.mjs",
+    "check": "pnpm lint && pnpm typecheck && pnpm test:home-navigation && pnpm test:events && pnpm test:ingest && pnpm build"
   },
   "dependencies": {
     "@base-ui/react": "^1.5.0",

scripts/test-signalops-ingest.mjs

@@ -0,0 +1,266 @@
+import assert from "node:assert/strict";
+
+import { SIGNALOPS_EVENT_LIMITS } from "../src/lib/signalops/events.ts";
+import {
+  authenticateSignalEventIngest,
+  isSignalEventIngestConfigured,
+} from "../src/lib/signalops/ingest-auth.ts";
+import {
+  createMemorySignalEventSink,
+  getSignalEventSinkForRuntime,
+} from "../src/lib/signalops/ingest-sink.ts";
+import { handleSignalEventIngest } from "../src/lib/signalops/ingest.ts";
+
+const configuredEnv = { SIGNALOPS_INGEST_TOKEN: "test-token-with-enough-entropy" };
+
+assert.equal(isSignalEventIngestConfigured(configuredEnv), true);
+assert.equal(isSignalEventIngestConfigured({}), false);
+
+const unauthorizedCases = [
+  new Headers(),
+  new Headers({ authorization: "Basic dGVzdDp0ZXN0" }),
+  new Headers({ authorization: "Bearer" }),
+  new Headers({ authorization: "Bearer wrong-token" }),
+  new Headers({ authorization: "Bearer  test-token-with-enough-entropy" }),
+];
+
+for (const headers of unauthorizedCases) {
+  assert.equal(authenticateSignalEventIngest(headers, configuredEnv), false);
+}
+
+assert.equal(
+  authenticateSignalEventIngest(
+    new Headers({ authorization: "Bearer test-token-with-enough-entropy" }),
+    configuredEnv,
+  ),
+  true,
+);
+
+const sink = createMemorySignalEventSink({ capacity: 2 });
+const eventOne = { eventId: "evt_one", occurredAt: "2026-07-19T00:00:00.000Z" };
+const eventTwo = { eventId: "evt_two", occurredAt: "2026-07-19T00:01:00.000Z" };
+const eventThree = { eventId: "evt_three", occurredAt: "2026-07-19T00:02:00.000Z" };
+
+const firstWrite = await sink.store([eventOne, eventTwo]);
+assert.deepEqual(firstWrite.storedEventIds, ["evt_one", "evt_two"]);
+assert.equal(firstWrite.duplicateEvents, 0);
+assert.equal(firstWrite.evictedEvents, 0);
+assert.equal(firstWrite.retainedEvents, 2);
+
+const boundedWrite = await sink.store([eventTwo, eventThree]);
+assert.deepEqual(boundedWrite.duplicateEventIds, ["evt_two"]);
+assert.deepEqual(boundedWrite.storedEventIds, ["evt_three"]);
+assert.equal(boundedWrite.evictedEvents, 1);
+assert.deepEqual(
+  sink.snapshot().map((event) => event.eventId),
+  ["evt_two", "evt_three"],
+);
+
+sink.reset();
+assert.deepEqual(sink.snapshot(), []);
+
+assert.equal(getSignalEventSinkForRuntime({ NODE_ENV: "production" }).adapter, "unconfigured");
+assert.equal(getSignalEventSinkForRuntime({ VERCEL: "1" }).adapter, "unconfigured");
+
+const repeatedBatchSink = createMemorySignalEventSink();
+const repeatedBatchWrite = await repeatedBatchSink.store([eventOne, eventOne]);
+assert.equal(repeatedBatchWrite.storedEvents, 1);
+assert.equal(repeatedBatchWrite.duplicateEvents, 1);
+assert.equal(repeatedBatchWrite.retainedEvents, 1);
+
+function ingestRequest(body, authorization) {
+  return new Request("http://localhost/api/events", {
+    method: "POST",
+    headers: {
+      "content-type": "application/json",
+      ...(authorization ? { authorization } : {}),
+    },
+    body,
+  });
+}
+
+const fixedRequestId = () => "req_test_ingest";
+const validBody = JSON.stringify({
+  type: "generation.completed",
+  occurredAt: "2026-07-19T00:00:00.000Z",
+  generationId: "gen_ingest_001",
+  providerId: "fal",
+  modelId: "flux-kontext-pro",
+  user: "private@example.com",
+  prompt: "private prompt",
+});
+
+const missingConfigResponse = await handleSignalEventIngest(ingestRequest(validBody), {
+  env: {},
+  sink: createMemorySignalEventSink(),
+  requestIdFactory: fixedRequestId,
+});
+assert.equal(missingConfigResponse.status, 503);
+assert.equal(missingConfigResponse.body.code, "ingest_not_configured");
+
+const unauthorizedResponses = await Promise.all(
+  [undefined, "Basic bad", "Bearer", "Bearer wrong-token"].map((authorization) =>
+    handleSignalEventIngest(ingestRequest(validBody, authorization), {
+      env: configuredEnv,
+      sink: createMemorySignalEventSink(),
+      requestIdFactory: fixedRequestId,
+    }),
+  ),
+);
+for (const response of unauthorizedResponses) {
+  assert.equal(response.status, 401);
+  assert.deepEqual(response.body, {
+    ok: false,
+    code: "unauthorized",
+    error: "Unauthorized.",
+    requestId: "req_test_ingest",
+  });
+}
+
+const productionSink = createMemorySignalEventSink();
+const productionResponse = await handleSignalEventIngest(
+  ingestRequest(validBody, "Bearer test-token-with-enough-entropy"),
+  {
+    env: { ...configuredEnv, NODE_ENV: "production" },
+    sink: productionSink,
+    requestIdFactory: fixedRequestId,
+  },
+);
+assert.equal(productionResponse.status, 503);
+assert.equal(productionResponse.body.code, "storage_not_ready");
+assert.deepEqual(productionSink.snapshot(), []);
+
+const ingestSink = createMemorySignalEventSink({ capacity: 4 });
+const ingestOptions = {
+  env: configuredEnv,
+  sink: ingestSink,
+  requestIdFactory: fixedRequestId,
+  currentTimeMs: Date.parse("2026-07-19T00:05:00.000Z"),
+};
+const authorizedResponse = await handleSignalEventIngest(
+  ingestRequest(validBody, "Bearer test-token-with-enough-entropy"),
+  ingestOptions,
+);
+assert.equal(authorizedResponse.status, 200);
+assert.equal(authorizedResponse.body.ok, true);
+assert.equal(authorizedResponse.body.partial, false);
+assert.equal(authorizedResponse.body.receipt.acceptedEvents, 1);
+assert.equal(authorizedResponse.body.receipt.rejectedEvents, 0);
+assert.equal(authorizedResponse.body.receipt.storedEvents, 1);
+assert.equal(authorizedResponse.body.receipt.duplicateEvents, 0);
+assert.match(authorizedResponse.body.receipt.receiptId, /^rcpt_[0-9a-f]{16}$/);
+assert.match(authorizedResponse.body.receipt.eventRefs[0], /^evtref_[0-9a-f]{16}$/);
+assert.equal(ingestSink.snapshot()[0].user, "[redacted]");
+assert.equal(ingestSink.snapshot()[0].prompt, "[redacted]");
+
+const duplicateResponse = await handleSignalEventIngest(
+  ingestRequest(validBody, "Bearer test-token-with-enough-entropy"),
+  ingestOptions,
+);
+assert.equal(duplicateResponse.status, 200);
+assert.equal(duplicateResponse.body.receipt.storedEvents, 0);
+assert.equal(duplicateResponse.body.receipt.duplicateEvents, 1);
+assert.equal(
+  duplicateResponse.body.receipt.receiptId,
+  authorizedResponse.body.receipt.receiptId,
+);
+assert.equal(ingestSink.snapshot().length, 1);
+
+const conflictingResponse = await handleSignalEventIngest(
+  ingestRequest(
+    JSON.stringify({ ...JSON.parse(validBody), cost: 99 }),
+    "Bearer test-token-with-enough-entropy",
+  ),
+  ingestOptions,
+);
+assert.equal(conflictingResponse.status, 409);
+assert.equal(conflictingResponse.body.code, "idempotency_conflict");
+assert.equal(ingestSink.snapshot().length, 1);
+assert.equal(ingestSink.snapshot()[0].cost, undefined);
+
+const atomicConflictSink = createMemorySignalEventSink();
+await atomicConflictSink.store([{ eventId: "evt_existing", cost: 1 }]);
+const atomicConflict = await atomicConflictSink.store([
+  { eventId: "evt_new", cost: 2 },
+  { eventId: "evt_existing", cost: 3 },
+]);
+assert.equal(atomicConflict.conflictEvents, 1);
+assert.equal(atomicConflict.storedEvents, 0);
+assert.deepEqual(
+  atomicConflictSink.snapshot().map((event) => event.eventId),
+  ["evt_existing"],
+);
+
+const partialSink = createMemorySignalEventSink();
+const partialResponse = await handleSignalEventIngest(
+  ingestRequest(
+    JSON.stringify({
+      events: [
+        JSON.parse(validBody),
+        { ...JSON.parse(validBody), eventId: "invalid-row", durationMs: -1 },
+      ],
+    }),
+    "Bearer test-token-with-enough-entropy",
+  ),
+  { ...ingestOptions, sink: partialSink },
+);
+assert.equal(partialResponse.status, 200);
+assert.equal(partialResponse.body.partial, true);
+assert.equal(partialResponse.body.receipt.acceptedEvents, 1);
+assert.equal(partialResponse.body.receipt.rejectedEvents, 1);
+assert.equal(partialResponse.body.receipt.storedEvents, 1);
+assert.equal(partialSink.snapshot().length, 1);
+
+const invalidJsonResponse = await handleSignalEventIngest(
+  ingestRequest("{not-json", "Bearer test-token-with-enough-entropy"),
+  { ...ingestOptions, sink: createMemorySignalEventSink() },
+);
+assert.equal(invalidJsonResponse.status, 400);
+assert.equal(invalidJsonResponse.body.code, "invalid_json");
+
+const oversizedSink = createMemorySignalEventSink();
+const oversizedResponse = await handleSignalEventIngest(
+  new Request("http://localhost/api/events", {
+    method: "POST",
+    headers: {
+      authorization: "Bearer test-token-with-enough-entropy",
+      "content-type": "application/json",
+      "content-length": String(SIGNALOPS_EVENT_LIMITS.maxBodyBytes + 1),
+    },
+    body: "{}",
+  }),
+  { ...ingestOptions, sink: oversizedSink },
+);
+assert.equal(oversizedResponse.status, 413);
+assert.equal(oversizedResponse.body.code, "payload_too_large");
+assert.deepEqual(oversizedSink.snapshot(), []);
+
+const rejectedSink = createMemorySignalEventSink();
+const rejectedResponse = await handleSignalEventIngest(
+  ingestRequest(
+    JSON.stringify({ ...JSON.parse(validBody), durationMs: -1 }),
+    "Bearer test-token-with-enough-entropy",
+  ),
+  { ...ingestOptions, sink: rejectedSink },
+);
+assert.equal(rejectedResponse.status, 422);
+assert.equal(rejectedResponse.body.code, "invalid_event");
+assert.deepEqual(rejectedSink.snapshot(), []);
+
+const throwingSink = {
+  adapter: "memory",
+  durable: false,
+  capacity: 1,
+  async store() {
+    throw new Error("secret backend detail");
+  },
+};
+const unexpectedResponse = await handleSignalEventIngest(
+  ingestRequest(validBody, "Bearer test-token-with-enough-entropy"),
+  { ...ingestOptions, sink: throwingSink },
+);
+assert.equal(unexpectedResponse.status, 500);
+assert.equal(unexpectedResponse.body.code, "internal_error");
+assert.doesNotMatch(JSON.stringify(unexpectedResponse.body), /secret backend detail/);
+
+console.log("signalops protected ingest auth, storage, and receipt checks passed");

src/app/api/events/route.ts

@@ -0,0 +1,17 @@
+import { NextResponse } from "next/server";
+
+import { getSignalEventSinkForRuntime } from "@/lib/signalops/ingest-sink";
+import { handleSignalEventIngest } from "@/lib/signalops/ingest";
+
+export const runtime = "nodejs";
+
+export async function POST(request: Request) {
+  const response = await handleSignalEventIngest(request, {
+    sink: getSignalEventSinkForRuntime(),
+  });
+
+  return NextResponse.json(response.body, {
+    status: response.status,
+    headers: { "cache-control": "no-store" },
+  });
+}

src/lib/signalops/ingest-auth.ts

@@ -0,0 +1,28 @@
+import { createHash, timingSafeEqual } from "node:crypto";
+
+type RuntimeEnv = Record<string, string | undefined>;
+
+function digestSecret(value: string) {
+  return createHash("sha256").update(value, "utf8").digest();
+}
+
+function extractBearerToken(headers: Headers) {
+  const authorization = headers.get("authorization") ?? "";
+  const match = /^Bearer ([^\s]+)$/.exec(authorization);
+  return match?.[1] ?? "";
+}
+
+export function isSignalEventIngestConfigured(env: RuntimeEnv = process.env) {
+  return Boolean(env.SIGNALOPS_INGEST_TOKEN?.trim());
+}
+
+export function authenticateSignalEventIngest(
+  headers: Headers,
+  env: RuntimeEnv = process.env,
+) {
+  const expectedToken = env.SIGNALOPS_INGEST_TOKEN ?? "";
+  const suppliedToken = extractBearerToken(headers);
+  const secretsMatch = timingSafeEqual(digestSecret(suppliedToken), digestSecret(expectedToken));
+
+  return Boolean(suppliedToken) && isSignalEventIngestConfigured(env) && secretsMatch;
+}