fix: preserve authoritative decisions during worker recovery

apps/control-plane/src/app.ts

@@ -75,32 +75,15 @@ export async function buildControlPlane(config: ControlPlaneConfig): Promise<Fas
   const eventsStore = database.events;
 
   await ensureAdmin(usersStore);
-  const storedApps = await appsStore.read();
-  const normalizedApps = storedApps.flatMap((record) => {
-    const parsed = AppSchema.safeParse(record);
-    return parsed.success ? [parsed.data] : [];
+  await appsStore.update((stored) => {
+    const valid = stored.flatMap((record) => { const parsed = AppSchema.safeParse(record); return parsed.success ? [parsed.data] : []; });
+    return valid.length ? valid : [createDefaultApp()];
   });
-  const validApps = normalizedApps.length ? normalizedApps : [createDefaultApp()];
-  if (JSON.stringify(validApps) !== JSON.stringify(storedApps)) await appsStore.write(validApps);
-  const storedKeys = await keysStore.read();
-  if (storedKeys.some((key) => !key.appId)) {
-    await keysStore.write(storedKeys.map((key) => ({ ...key, appId: key.appId ?? "default" })));
-  }
-  const storedProfiles = await profilesStore.read();
-  const normalizedProfiles = storedProfiles.flatMap((profile) => {
-    const parsed = ProfileSchema.safeParse(profile);
-    return parsed.success ? [parsed.data] : [];
+  await keysStore.update((stored) => stored.map((key) => ({ ...key, appId: key.appId ?? "default" })));
+  await settingsStore.update((stored) => {
+    const parsed = ProviderSettingsSchema.safeParse(stored);
+    return parsed.success ? parsed.data : { ...createDefaultProviderSettings(), endpoint: config.typesafeEndpoint, model: config.typesafeModel };
   });
-  if (normalizedProfiles.length !== storedProfiles.length || JSON.stringify(normalizedProfiles) !== JSON.stringify(storedProfiles)) {
-    await profilesStore.write(normalizedProfiles.length ? normalizedProfiles : [createDefaultProfile()]);
-  }
-  const storedSettings = await settingsStore.read();
-  const normalizedSettings = ProviderSettingsSchema.safeParse(storedSettings);
-  if (normalizedSettings.success && JSON.stringify(normalizedSettings.data) !== JSON.stringify(storedSettings)) {
-    await settingsStore.write(normalizedSettings.data);
-  } else if (!normalizedSettings.success) {
-    await settingsStore.write({ ...createDefaultProviderSettings(), endpoint: config.typesafeEndpoint, model: config.typesafeModel });
-  }
 
   const requireSession = accessGuard(database);
 

apps/control-plane/test/reviews.test.ts

@@ -1,19 +1,31 @@
 import assert from "node:assert/strict";
 import test from "node:test";
 import { randomUUID } from "node:crypto";
-import { openDatabase } from "@pyro/storage";
+import { encryptText, openDatabase } from "@pyro/storage";
+import { DeliveryWorker, signWebhook } from "@pyro/integrations";
 import { buildControlPlane } from "../src/app.js";
 test("review resolution is attributed, immutable, concurrency-safe and does not change a decision", async (t) => {
  const config = { host: "127.0.0.1", port: 0, databaseUrl: `memory://reviews-${randomUUID()}`, adminPassword: "correct-horse-battery-staple", controlPlaneSecret: "control-plane-test-secret", gatewayInternalUrl: "http://127.0.0.1:1", gatewayApiKey: "test-key", typesafeEndpoint: "https://api.typesafe.ai/v1/systemone", typesafeModel: "jev-latest" };
  const app = await buildControlPlane(config); t.after(() => app.close()); const db = await openDatabase(config.databaseUrl);
  const login = await app.inject({ method: "POST", url: "/api/auth/login", payload: { password: config.adminPassword } }); const cookie = login.headers["set-cookie"]!.split(";")[0]!;
  const event = { id: "review-one", appId: "default", createdAt: new Date().toISOString(), profileId: "default", action: "review" as const, verdict: "suspicious" as const, risk: .7, confidence: .7, reason: "review", detectors: [], model: "local", provider: "local-rules", latencyMs: 0, queueMs: 0, inputHash: "hash" };
  await db.events.append(event);
+ const integrationId = randomUUID(), signingSecret = "qa-webhook-signing-secret";
+ await db.document("integrations", () => []).write([{ id: integrationId, name: "Review callback", type: "webhook", enabled: true, reviewResolutions: true, actions: ["review"], minimumRisk: 0, profileIds: [], appIds: ["default"], allowPrivateNetwork: true, destination: encryptText("http://127.0.0.1/events", config.controlPlaneSecret), signingSecret: encryptText(signingSecret, config.controlPlaneSecret), createdAt: event.createdAt, updatedAt: event.createdAt }] as never[]);
  const edit = (disposition: string) => app.inject({ method: "PUT", url: "/api/reviews/review-one", headers: { cookie }, payload: { expectedRevision: 0, disposition, comment: "Redacted feedback" } });
  const responses = await Promise.all([edit("true_positive"), edit("false_positive")]);
  assert.deepEqual(responses.map((r) => r.statusCode).sort(), [200, 409]);
  assert.deepEqual(await db.events.findById(event.id), event);
  const review = (await app.inject({ method: "GET", url: "/api/reviews/review-one", headers: { cookie } })).json().review;
+ for (let i = 0; i < 40 && !(await db.deliveries.list(integrationId)).length; i++) await new Promise((r) => setTimeout(r, 50));
+ const callbacks = await db.deliveries.list(integrationId);
+ assert.equal(callbacks.length, 1); assert.equal(callbacks[0]!.payload.type, "review.resolved");
+ assert.equal(callbacks[0]!.payload.data.action, "review");
+ const transport = new DeliveryWorker(db, config.controlPlaneSecret, async (_url, body, headers) => {
+   assert.equal(headers["x-pyro-signature"], signWebhook(signingSecret, headers["x-pyro-timestamp"]!, body));
+   assert.equal(JSON.parse(body).data.review.revision, 1); return { status: 204 };
+ });
+ await transport.tick(); assert.equal((await db.deliveries.list(integrationId))[0]!.status, "delivered");
  assert.equal(review.revision, 1); assert.equal(review.status, "resolved"); assert.equal(review.resolvedBy, login.json().user.id); assert.equal(review.comments.length, 1);
  const viewer = (await app.inject({ method: "POST", url: "/api/team", headers: { cookie }, payload: { username: "viewer", role: "viewer", appIds: ["default"] } })).json();
  const session = await app.inject({ method: "POST", url: "/api/auth/login", payload: { username: "viewer", password: viewer.password } });

apps/gateway/src/app.ts

@@ -105,45 +105,16 @@ export async function buildGateway(config: GatewayConfig): Promise<FastifyInstan
   const keys = new CachedDocument(keysStore);
   const secrets = new CachedDocument(secretsStore);
 
-  const storedProfiles = await profilesStore.read();
-  const normalizedProfiles = storedProfiles.flatMap((profile) => {
-    const parsed = ProfileSchema.safeParse(profile);
-    return parsed.success ? [{ ...profile, ...parsed.data }] : [];
+  await appsStore.update((stored) => {
+    const valid = stored.flatMap((record) => { const parsed = AppSchema.safeParse(record); return parsed.success ? [parsed.data] : []; });
+    return valid.length ? valid : [createDefaultApp()];
   });
-  if (normalizedProfiles.length !== storedProfiles.length || JSON.stringify(normalizedProfiles) !== JSON.stringify(storedProfiles)) {
-    await profilesStore.write(normalizedProfiles.length ? normalizedProfiles : [createDefaultProfile()]);
-  }
-  const storedApps = await appsStore.read();
-  const normalizedApps = storedApps.flatMap((record) => {
-    const parsed = AppSchema.safeParse(record);
-    return parsed.success ? [parsed.data] : [];
+  await settingsStore.update((stored) => { const parsed = ProviderSettingsSchema.safeParse(stored); return parsed.success ? parsed.data : createDefaultProviderSettings(); });
+  await keysStore.update((stored) => {
+    const normalized = stored.map((key) => ({ ...key, appId: key.appId ?? "default" }));
+    if (normalized.length || !config.bootstrapApiKey) return normalized;
+    return [{ id: randomUUID(), name: "Local development", prefix: config.bootstrapApiKey.slice(0, 8), hash: hash(config.bootstrapApiKey), appId: "default", createdAt: new Date().toISOString() }];
   });
-  const validApps = normalizedApps.length ? normalizedApps : [createDefaultApp()];
-  if (JSON.stringify(validApps) !== JSON.stringify(storedApps)) await appsStore.write(validApps);
-  const storedSettings = await settingsStore.read();
-  const normalizedSettings = ProviderSettingsSchema.safeParse(storedSettings);
-  if (normalizedSettings.success && JSON.stringify(normalizedSettings.data) !== JSON.stringify(storedSettings)) {
-    await settingsStore.write(normalizedSettings.data);
-  } else if (!normalizedSettings.success) {
-    await settingsStore.write(createDefaultProviderSettings());
-  }
-  const storedKeys = await keysStore.read();
-  if (storedKeys.some((key) => !key.appId)) {
-    await keysStore.write(storedKeys.map((key) => ({ ...key, appId: key.appId ?? "default" })));
-  }
-  if ((await keysStore.read()).length === 0 && config.bootstrapApiKey) {
-    await keysStore.write([
-      {
-        id: randomUUID(),
-        name: "Local development",
-        prefix: config.bootstrapApiKey.slice(0, 8),
-        hash: hash(config.bootstrapApiKey),
-        appId: "default",
-        createdAt: new Date().toISOString(),
-      },
-    ]);
-    keys.invalidate();
-  }
 
   const queue = new ConcurrentQueue(config.queueConcurrency, config.queueMaxDepth);
   const metrics = new GatewayMetrics();
@@ -228,13 +199,15 @@ export async function buildGateway(config: GatewayConfig): Promise<FastifyInstan
     const allProfiles = await profiles.read();
     const shadowProfiles = (profile.shadowProfileIds ?? [])
       .filter((profileId) => profileId !== profile.id)
-      .map((profileId) => allProfiles.find((candidate) => candidate.id === profileId))
+      .map((profileId) => { const record = allProfiles.find((candidate) => candidate.id === profileId); return record ? revisionOf(record, firewallApp.profileRevisions?.[profileId]) : undefined; })
       .filter((candidate): candidate is Profile => Boolean(candidate))
       .slice(0, 3);
     const shadows = await Promise.all(shadowProfiles.map(async (shadowProfile) => {
       const result = await evaluate(`${id}:shadow:${shadowProfile.id}`, envelope, shadowProfile, 0, traceId, firewallApp);
       return {
         profileId: shadowProfile.id,
+        policyRevision: shadowProfile.revision,
+        policyHash: shadowProfile.contentHash ?? policyHash(shadowProfile),
         verdict: result.decision.verdict,
         action: result.decision.action,
         risk: result.decision.risk,
@@ -263,8 +236,12 @@ export async function buildGateway(config: GatewayConfig): Promise<FastifyInstan
     app.log.info({ decisionId: id, requestId, traceId, profile: profile.id, action: decision.action, risk: decision.risk, latencyMs: decision.latencyMs }, "classification completed");
     metrics.record(decision, Boolean(failure));
     metrics.updateQueue(queue.snapshot());
-    eventBus.emit("decision", event);
-    return decision;
+    // A recovered lease can overlap an upstream call from a paused worker.
+    // The first committed event remains authoritative for every replay.
+    const committed = (await eventsStore.findById(id))!;
+    eventBus.emit("decision", committed);
+    const { inputPreview, inputHash, inputBytes, appRulesSnapshot, localRuleId: storedRule, apiKeyId, apiKeyName, appName, error, ...result } = committed;
+    return result;
   };
 
   const enqueue = (
@@ -279,7 +256,7 @@ export async function buildGateway(config: GatewayConfig): Promise<FastifyInstan
 
   app.get("/", async () => ({
     name: "Pyro gateway",
-    version: "0.2.0",
+    version: "0.3.0-beta.1",
     endpoints: ["POST /v1/classify", "POST /v1/jobs", "GET /v1/jobs/:id", "WS /v1/events"],
   }));
 
@@ -312,7 +289,7 @@ export async function buildGateway(config: GatewayConfig): Promise<FastifyInstan
     const firewallApp = request.firewallApp!;
     return (await profiles.read())
       .filter((profile) => !profile.archived)
-      .filter((profile) => firewallApp.allowedProfileIds.length === 0 || firewallApp.allowedProfileIds.includes(profile.id))
+      .filter((profile) => profileAllowed(profile.id, request.apiKey!, firewallApp))
       .map(({ id, name, description }) => ({ id, name, description }));
   });
 

packages/storage/src/event-replay.test.ts

@@ -0,0 +1,17 @@
+import assert from "node:assert/strict";
+import { randomUUID } from "node:crypto";
+import test from "node:test";
+import type { ClassificationEvent, Delivery } from "@pyro/contracts";
+import { openDatabase } from "./index.js";
+for (const url of [`memory://replay-${randomUUID()}`, ...(process.env.TEST_DATABASE_URL ? [process.env.TEST_DATABASE_URL] : [])]) {
+  test(`first event and its webhook outbox are authoritative: ${url.startsWith("memory") ? "memory" : "PostgreSQL"}`, async (t) => {
+    const db = await openDatabase(url); t.after(() => db.close());
+    const id = randomUUID(), integrationId = randomUUID(), at = new Date().toISOString();
+    const event: ClassificationEvent = { id, appId: id, profileId: "default", createdAt: at, action: "allow", verdict: "safe", risk: 0, confidence: 1, reason: "first", detectors: [], model: "local", provider: "local-rules", latencyMs: 0, queueMs: 0, inputHash: "hash" };
+    const delivery: Delivery = { id: randomUUID(), integrationId, eventId: id, createdAt: at, status: "pending", attempts: 0, nextAttemptAt: at, payload: { id, type: "decision.created", createdAt: at, data: { id, action: "allow" } } };
+    await db.events.append(event, [delivery]);
+    await db.events.append({ ...event, action: "block", reason: "late worker" }, [{ ...delivery, id: randomUUID() }]);
+    assert.deepEqual(await db.events.findById(id), event);
+    assert.equal((await db.deliveries.list(integrationId)).length, 1);
+  });
+}

packages/storage/src/index.ts

@@ -311,15 +311,15 @@ class PostgresEvents implements EventStore {
     const client = await this.pool.connect();
     try {
       await client.query("BEGIN");
-    await client.query(
+    const inserted = await client.query(
       `INSERT INTO pyro_events
         (id, created_at, app_id, profile_id, verdict, action, risk, provider, api_key_id, labels, payload)
        VALUES ($1, $2::timestamptz, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11::jsonb)
        ON CONFLICT (id) DO NOTHING`,
       [event.id, event.createdAt, event.appId ?? null, event.profileId, event.verdict, event.action, event.risk,
         event.provider, event.apiKeyId ?? null, JSON.stringify(event.labels ?? {}), JSON.stringify(event)],
     );
-      for (const delivery of deliveries) await insertDelivery(client, delivery);
+      if (inserted.rowCount) for (const delivery of deliveries) await insertDelivery(client, delivery);
       await client.query("COMMIT");
     } catch (error) { await client.query("ROLLBACK"); throw error; }
     finally { client.release(); }
@@ -689,7 +689,8 @@ class MemoryDatabase implements Database {
   };
   readonly events: EventStore = {
     append: async (event, deliveries = []) => {
-      if (!this.eventRows.some((item) => item.id === event.id)) this.eventRows.push(structuredClone(event));
+      if (this.eventRows.some((item) => item.id === event.id)) return;
+      this.eventRows.push(structuredClone(event));
       for (const delivery of deliveries) await this.deliveries.enqueue(delivery);
     },
     readRecent: async (limit = 100) => this.eventRows.slice(-Math.max(1, limit)).reverse().map((event) => structuredClone(event)),