automation-pipeline.integration.test.ts

automation-pipeline.integration.test.tstests/integration/automation-pipeline.integration.test.tsraw ← integration41,624 B← Back to the app
1/**
2 * AI Software Factory — Integration: Automation pipeline (Phase 11, TASK-0113)
3 *
4 * The run pipeline (`handleRunAutomation`) against the REAL stack:
5 * Testcontainers Postgres/Redis (shared container helpers). No MinIO /
6 * OpenSearch / ClamAV needed — the scenarios use hermetic actions
7 * (UPDATE_RECORD, CREATE_NOTIFICATION) plus the registry's test-injection
8 * seam (`registerTestExecutor`) for the retry→dead-letter path. The seam is
9 * the only production-code addition in Phase 11 for testability (Node's
10 * undici fetch errors carry no classifiable errno, so a real HTTPS endpoint
11 * cannot be made to fail deterministically; see docs/architecture/automation.md).
12 *
13 * Scenarios (docs/tasks/phase-11-automation.md TASK-0113 + §9 crash-safety):
14 *   1.  happy path        → SUCCEEDED, property written, notification created,
15 *                           automation.executed emitted, run logged
16 *   2.  condition gate    → false: SUCCEEDED {skipped:"conditions_not_met"};
17 *                           true: actions run (SELECT compared by option NAME)
18 *   3.  self-trigger      → SUCCEEDED {skipped:"self_trigger"} (no actions)
19 *   4.  depth limit       → SUCCEEDED {skipped:"depth_limit"} at depth 5
20 *                           (self_trigger wins when both apply)
21 *   5.  fatal error       → FAILED (no retry, no dead-letter)
22 *   6.  retryable ×8      → 8 RETRYING hops (PENDING, retryCount 1..8) then
23 *                           DEAD_LETTER + admin notification +
24 *                           automation.dead_lettered audit + automation.executed
25 *                           + automations:dead_letter:{ws} counter
26 *   7.  60-action guard   → FAILED RUN_ACTION_LIMIT (corrupt-row defense)
27 *   8.  watchdog          → stale RUNNING (below cap) → PENDING +retryCount;
28 *                           stale RUNNING (at cap) → FAILED WORKER_CRASHED +
29 *                           notification + audit + counter; stale PENDING →
30 *                           re-enqueued without retryCount bump
31 *   9.  crash recovery    → a REAL BullMQ worker child SIGKILL'd mid-run
32 *                           leaves the row RUNNING; the watchdog reclaims it;
33 *                           the re-driven run succeeds
34 *  10.  end-to-end stream → record.created xadd → consumer (group
35 *                           automations-matcher) → matchAndEnqueue → run row →
36 *                           real BullMQ worker → SUCCEEDED (the
37 *                           record-created→status-change worker-side half)
38 *
39 * Retry hops are driven by calling `handleRunAutomation` directly per attempt
40 * (each `scheduleRetry` leaves the row PENDING with retryCount+1); the real
41 * BullMQ backoff delays (1 s … 128 s) are NOT awaited — the re-enqueued
42 * delayed jobs are harmless (no worker picks them up in time; the queue is
43 * obliterated in afterAll).
44 */
45import { afterAll, beforeAll, describe, expect, it } from "vitest";
46import { spawn } from "node:child_process";
47import { randomBytes } from "node:crypto";
48import { writeFileSync, rmSync } from "node:fs";
49import os from "node:os";
50import path from "node:path";
51import { fileURLToPath, pathToFileURL } from "node:url";
52import { Prisma, PrismaClient } from "@prisma/client";
53import { Redis } from "ioredis";
54import { getDatabaseUrl, getRedisUrl } from "./helpers/containers";
55import { deploySchema } from "./helpers/api";
56 
57const HERE = path.dirname(fileURLToPath(import.meta.url));
58const WORKER_SRC = path.resolve(HERE, "..", "..", "apps", "worker", "src");
59const REGISTRY_DIR = path.join(WORKER_SRC, "processors", "automations");
60 
61function fileUrl(p: string): string {
62  return `file://${p.startsWith("/") ? p : `/${p}`}`;
63}
64 
65// Worker pipeline + seams from SOURCE (the worker ships no build for the
66// monorepo test path — same source-import convention as files-pipeline).
67const { handleRunAutomation } = await import(
68  /* webpackIgnore: true */
69  fileUrl(path.join(REGISTRY_DIR, "pipeline.ts"))
70);
71const { enqueueRunJob } = await import(
72  /* webpackIgnore: true */
73  fileUrl(path.join(REGISTRY_DIR, "matcher.ts"))
74);
75const { runWatchdog } = await import(
76  /* webpackIgnore: true */
77  fileUrl(path.join(REGISTRY_DIR, "watchdog.ts"))
78);
79const { registerTestExecutor, clearTestExecutors } = await import(
80  /* webpackIgnore: true */
81  fileUrl(path.join(REGISTRY_DIR, "executors", "registry.ts"))
82);
83const { FatalError, RetryableError } = await import(
84  /* webpackIgnore: true */
85  fileUrl(path.join(REGISTRY_DIR, "executors", "classify.ts"))
86);
87const processor = await import(
88  /* webpackIgnore: true */
89  fileUrl(path.join(WORKER_SRC, "processors", "automations.processor.ts"))
90);
91 
92// ── infra gate ──────────────────────────────────────────────────────────────
93 
94let dbUrl = "";
95let redisUrl = "";
96let infraReady = false;
97try {
98  [dbUrl, redisUrl] = await Promise.all([getDatabaseUrl(), getRedisUrl()]);
99  infraReady = true;
100} catch {
101  infraReady = false; // no Docker daemon (or container image unavailable)
102}
103 
104// ── state ───────────────────────────────────────────────────────────────────
105 
106let prisma: PrismaClient;
107let redis: Redis;
108let env: import("../../apps/worker/src/config").WorkerEnv;
109 
110function baseEnv(): import("../../apps/worker/src/config").WorkerEnv {
111  return {
112    NODE_ENV: "test",
113    REDIS_URL: redisUrl,
114    DATABASE_URL: dbUrl,
115    S3_ENDPOINT: "http://localhost:9000",
116    S3_REGION: "us-east-1",
117    S3_BUCKET_FILES: "factory-files",
118    S3_BUCKET_THUMBS: "factory-thumbs",
119    S3_BUCKET: "factory-files",
120    S3_ACCESS_KEY: "minioadmin",
121    S3_SECRET_KEY: "minioadmin",
122    S3_PUBLIC_BASE: "http://localhost:9001",
123    SEARCH_URL: "http://localhost:9200",
124    SEARCH_USERNAME: "admin",
125    SEARCH_PASSWORD: "changeme",
126    SEARCH_VECTOR_DIMENSION: 1536,
127    SEARCH_KNN_M: 16,
128    SEARCH_KNN_EF_CONSTRUCTION: 100,
129    MASTER_ENCRYPTION_KEY: "",
130    AI_API_KEY: "",
131    AI_PROVIDER: "openai",
132    AI_BASE_URL: "",
133    AI_MODEL: "gpt-4o-mini",
134    AI_EMBEDDING_PROVIDER: "",
135    AI_EMBEDDING_MODEL: "text-embedding-3-small",
136    SMTP_HOST: "", // no mail — the Notification row is the assertion
137    SMTP_PORT: 587,
138    SMTP_USER: "",
139    SMTP_PASS: "",
140    SMTP_FROM: "noreply@factory.local",
141    SMTP_STARTTLS: true,
142    CLAMAV_HOST: "",
143    CLAMAV_PORT: 3310,
144    CLAMD_SOCKET: "",
145    FILES_WORKER_CONCURRENCY: 4,
146    CLAMAV_TIMEOUT_MS: 30_000,
147    FILES_TEMP_DIR: os.tmpdir(),
148    FILES_TEMP_MAX_AGE_MS: 600_000,
149    FILES_TEMP_FREE_MIN_BYTES: 1,
150    PUBLIC_URL: "http://localhost:3000",
151    WORKER_CONCURRENCY: 10,
152    WORKER_ID: "pipeline-it",
153    WORKER_RETENTION_EXECUTE: false,
154    HEALTH_PORT: 4003,
155    AUTO_RUN_MAX_RETRIES: 8,
156    AUTOMATIONS_WORKER_CONCURRENCY: 1
157  } as import("../../apps/worker/src/config").WorkerEnv;
158}
159 
160// ── fixtures ────────────────────────────────────────────────────────────────
161 
162interface Fixture {
163  ownerId: string;
164  workspaceId: string;
165  databaseId: string;
166  recordId: string;
167}
168 
169/** User (OWNER member) + workspace + database + record with a TITLE and a
170 *  SELECT "Status" property (options A/B). The SELECT value is stored as the
171 *  Phase-5 envelope { optionId }; the condition evaluator normalizes it to
172 *  the option NAME. The OWNER membership is required by the worker's
173 *  permission walk (role baseline) AND is the DLQ notification recipient. */
174async function seedFixture(): Promise<Fixture> {
175  const ownerId = `user_ap_${randomBytes(6).toString("hex")}`;
176  const workspaceId = `wsp_ap_${randomBytes(6).toString("hex")}`;
177  const databaseId = `db_ap_${randomBytes(6).toString("hex")}`;
178  const recordId = `rec_ap_${randomBytes(6).toString("hex")}`;
179  const titlePropId = `prop_t_${randomBytes(4).toString("hex")}`;
180  const statusPropId = `prop_s_${randomBytes(4).toString("hex")}`;
181  await prisma.user.create({ data: { id: ownerId, email: `${ownerId}@factory.local`, name: "Pipeline Owner" } });
182  await prisma.workspace.create({
183    data: { id: workspaceId, name: `WS ${workspaceId}`, slug: `slug-${workspaceId}`, ownerId, billingPlan: "paid" }
184  });
185  await prisma.workspaceMember.create({ data: { workspaceId, userId: ownerId, role: "OWNER" } });
186  await prisma.database.create({ data: { id: databaseId, workspaceId, title: "Pipeline DB" } });
187  await prisma.databaseProperty.create({
188    data: { id: titlePropId, databaseId, name: "Name", type: "TITLE", isTitle: true }
189  });
190  await prisma.databaseProperty.create({
191    data: {
192      id: statusPropId,
193      databaseId,
194      name: "Status",
195      type: "SELECT",
196      config: { options: [{ id: "opt-a", name: "A" }, { id: "opt-b", name: "B" }] } as Prisma.InputJsonValue
197    }
198  });
199  await prisma.databaseRecord.create({ data: { id: recordId, workspaceId, databaseId, createdBy: ownerId } });
200  await prisma.recordProperty.create({
201    data: { recordId, propertyId: statusPropId, value: { optionId: "opt-a" } as Prisma.InputJsonValue }
202  });
203  return { ownerId, workspaceId, databaseId, recordId };
204}
205 
206async function makeAutomation(opts: {
207  fixture: Fixture;
208  name: string;
209  actions: Array<{ type: string; config: Record<string, unknown> }>;
210  condition?: unknown;
211  triggerType?: string;
212  triggerConfig?: Record<string, unknown>;
213}): Promise<string> {
214  const id = `auto_ap_${randomBytes(6).toString("hex")}`;
215  const { fixture } = opts;
216  await prisma.automation.create({
217    data: {
218      id,
219      workspaceId: fixture.workspaceId,
220      name: opts.name,
221      status: "ACTIVE",
222      databaseId: fixture.databaseId,
223      createdById: fixture.ownerId,
224      triggers: {
225        create: [
226          {
227            type: (opts.triggerType ?? "RECORD_CREATED") as never,
228            config: { databaseId: fixture.databaseId, ...(opts.triggerConfig ?? {}) } as Prisma.InputJsonValue,
229            position: 0
230          }
231        ]
232      },
233      ...(opts.condition !== undefined
234        ? { conditions: { create: [{ expression: opts.condition as Prisma.InputJsonValue, position: 0 }] } }
235        : {})
236    }
237  });
238  for (let i = 0; i < opts.actions.length; i++) {
239    const a = opts.actions[i]!;
240    await prisma.automationAction.create({
241      data: {
242        automationId: id,
243        type: a.type as never,
244        config: a.config as Prisma.InputJsonValue,
245        position: i,
246        retries: 3,
247        timeoutMs: 30_000
248      }
249    });
250  }
251  return id;
252}
253 
254/** A PENDING run row for a direct pipeline drive (the matcher's shape). */
255async function makeRun(automationId: string, fixture: Fixture, triggerEvent: Record<string, unknown>): Promise<string> {
256  const runId = `run_ap_${randomBytes(6).toString("hex")}`;
257  await prisma.automationRun.create({
258    data: {
259      id: runId,
260      workspaceId: fixture.workspaceId,
261      automationId,
262      status: "PENDING",
263      triggerEvent: triggerEvent as Prisma.InputJsonValue,
264      input: { triggerEvent, actionCount: 0 } as Prisma.InputJsonValue,
265      retryCount: 0
266    }
267  });
268  return runId;
269}
270 
271function jobData(runId: string, automationId: string, workspaceId: string, triggerEvent: Record<string, unknown>) {
272  return { runId, automationId, workspaceId, triggerEvent, eventId: `evt_${randomBytes(4).toString("hex")}` };
273}
274 
275function recordCreatedEvent(fixture: Fixture): Record<string, unknown> {
276  const payload = {
277    workspaceId: fixture.workspaceId,
278    databaseId: fixture.databaseId,
279    recordId: fixture.recordId,
280    changedProperties: ["*"],
281    createdBy: fixture.ownerId
282  };
283  // buildTriggerEvent's shape: { type, payload, ...flattened payload }.
284  return { type: "record.created", payload, ...payload };
285}
286 
287const deps = () => ({ prisma, redis, env });
288 
289/** All `factory:events` entries carrying one runId (decoded payloads). */
290async function runEvents(runId: string): Promise<Array<Record<string, unknown>>> {
291  const raw = await redis.xrange("factory:events", "-", "+");
292  const out: Array<Record<string, unknown>> = [];
293  for (const [, fields] of raw) {
294    const kv: Record<string, string> = {};
295    for (let i = 0; i < fields.length; i += 2) kv[fields[i]!] = fields[i + 1]!;
296    if (kv.payload) {
297      try {
298        const p = JSON.parse(kv.payload) as Record<string, unknown>;
299        if (p.runId === runId) out.push({ event: kv.event, ...p });
300      } catch {
301        /* not JSON — ignore */
302      }
303    }
304  }
305  return out;
306}
307 
308const logCount = (runId: string) => prisma.automationLog.count({ where: { runId } });
309 
310async function statusOf(runId: string): Promise<string> {
311  const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
312  return row.status;
313}
314 
315async function waitFor<T>(
316  fn: () => Promise<T>,
317  pred: (v: T) => boolean,
318  timeoutMs = 20_000,
319  what = "condition"
320): Promise<T> {
321  const t0 = Date.now();
322  for (;;) {
323    const v = await fn();
324    if (pred(v)) return v;
325    if (Date.now() - t0 > timeoutMs) throw new Error(`timed out waiting for ${what}`);
326    await new Promise((r) => setTimeout(r, 250));
327  }
328}
329 
330// ── suite ───────────────────────────────────────────────────────────────────
331 
332describe.skipIf(!infraReady)(
333  "Automation pipeline integration (TASK-0113 — worker, real stack)",
334  () => {
335    beforeAll(async () => {
336      await deploySchema(dbUrl);
337      prisma = new PrismaClient({ datasources: { db: { url: dbUrl } } });
338      await prisma.$connect();
339      redis = new Redis(redisUrl, { maxRetriesPerRequest: null });
340      env = baseEnv();
341    }, 180_000);
342 
343    afterAll(async () => {
344      clearTestExecutors();
345      // Wipe the automations queue (delayed retry jobs left by the retry
346      // scenario are still parked with real backoff delays).
347      try {
348        const { Queue } = await import("bullmq");
349        const q = new Queue("automations", { connection: redis });
350        await q.obliterate({ force: true });
351        await q.close();
352      } catch {
353        /* best-effort */
354      }
355      await redis?.quit().catch(() => {});
356      await prisma?.$disconnect().catch(() => {});
357    });
358 
359    // ── 1. happy path (direct driver) ────────────────────────────────────
360 
361    it("record.created → UPDATE_RECORD + CREATE_NOTIFICATION → SUCCEEDED (property written, notification, automation.executed, logged)", async () => {
362      const f = await seedFixture();
363      const automationId = await makeAutomation({
364        fixture: f,
365        name: "Happy path",
366        actions: [
367          { type: "UPDATE_RECORD", config: { recordId: "from_trigger", propertyValues: { Status: "opt-b" } } },
368          { type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "Pipeline test notification" } }
369        ]
370      });
371      const te = recordCreatedEvent(f);
372      const runId = await makeRun(automationId, f, te);
373 
374      const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
375      expect(status).toBe("SUCCEEDED");
376 
377      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
378      expect(row.retryCount).toBe(0);
379      expect(row.startedAt).not.toBeNull();
380      expect(row.finishedAt).not.toBeNull();
381      const output = row.output as Record<string, unknown>;
382      const actions = output.actions as Array<Record<string, unknown>>;
383      expect(actions).toHaveLength(2);
384      expect(actions[0]!.status).toBe("succeeded");
385      expect(actions[1]!.status).toBe("succeeded");
386 
387      // The property write landed as the Phase-5 envelope.
388      const props = await prisma.recordProperty.findMany({
389        where: { recordId: f.recordId },
390        include: { property: { select: { name: true } } }
391      });
392      const statusProp = props.find((p) => p.property.name === "Status");
393      expect((statusProp?.value as Record<string, unknown>)?.optionId).toBe("opt-b");
394 
395      // The notification was created for the automation's creator (owner).
396      const note = await prisma.notification.findFirst({ where: { workspaceId: f.workspaceId, userId: f.ownerId } });
397      expect(note).not.toBeNull();
398      expect(note!.body).toBe("Pipeline test notification");
399 
400      // Depth-0 terminal events carry automation.executed.
401      const evts = await runEvents(runId);
402      const executed = evts.find((e) => e.event === "automation.executed") as { status?: string } | undefined;
403      expect(executed).toBeDefined();
404      expect(executed?.status).toBe("SUCCEEDED");
405 
406      // Every run is logged (started + actions + completed).
407      expect(await logCount(runId)).toBeGreaterThanOrEqual(4);
408    });
409 
410    // ── 2. condition gate ─────────────────────────────────────────────────
411 
412    it("condition gate: SELECT property compared by option NAME — false skips, true runs", async () => {
413      const f = await seedFixture();
414 
415      // Record Status = A. Condition "Status eq B" → false.
416      const gateFalse = await makeAutomation({
417        fixture: f,
418        name: "Gate false",
419        condition: { field: { ref: "property", name: "Status" }, operator: "eq", value: "B" },
420        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "should not fire" } }]
421      });
422      const teF = recordCreatedEvent(f);
423      const runF = await makeRun(gateFalse, f, teF);
424      const sF = await handleRunAutomation(deps(), { data: jobData(runF, gateFalse, f.workspaceId, teF) });
425      expect(sF).toBe("SUCCEEDED");
426      const rowF = await prisma.automationRun.findUniqueOrThrow({ where: { id: runF } });
427      expect((rowF.output as Record<string, unknown>).skipped).toBe("conditions_not_met");
428      expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(0);
429      expect(await logCount(runF)).toBeGreaterThanOrEqual(1);
430 
431      // Same record, condition "Status eq A" → true → the action runs.
432      const gateTrue = await makeAutomation({
433        fixture: f,
434        name: "Gate true",
435        condition: { field: { ref: "property", name: "Status" }, operator: "eq", value: "A" },
436        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "fired" } }]
437      });
438      const teT = recordCreatedEvent(f);
439      const runT = await makeRun(gateTrue, f, teT);
440      const sT = await handleRunAutomation(deps(), { data: jobData(runT, gateTrue, f.workspaceId, teT) });
441      expect(sT).toBe("SUCCEEDED");
442      const rowT = await prisma.automationRun.findUniqueOrThrow({ where: { id: runT } });
443      const outT = rowT.output as Record<string, unknown>;
444      expect(outT.skipped).toBeUndefined();
445      expect((outT.actions as unknown[])).toHaveLength(1);
446      expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(1);
447    });
448 
449    // ── 3. loop prevention ────────────────────────────────────────────────
450 
451    it("self-trigger block: the automation's own event is skipped (SUCCEEDED {skipped:'self_trigger'})", async () => {
452      const f = await seedFixture();
453      const automationId = await makeAutomation({
454        fixture: f,
455        name: "Self trigger",
456        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "must not fire" } }]
457      });
458      const te = {
459        ...recordCreatedEvent(f),
460        automationContext: { runId: "run_prev", automationId, depth: 1 }
461      };
462      const runId = await makeRun(automationId, f, te);
463      const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
464      expect(status).toBe("SUCCEEDED");
465      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
466      expect((row.output as Record<string, unknown>).skipped).toBe("self_trigger");
467      expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(0);
468      expect(await logCount(runId)).toBeGreaterThanOrEqual(1);
469      // automation.executed is depth-0 ONLY (§8 event-storm guard): this run
470      // was triggered by the automation's own depth-1 event, so NO executed
471      // event is emitted for it (the root run's event already covered the chain).
472      expect((await runEvents(runId)).find((e) => e.event === "automation.executed")).toBeUndefined();
473    });
474 
475    it("depth limit: events at depth ≥ 5 are skipped (SUCCEEDED {skipped:'depth_limit'}); self-trigger wins", async () => {
476      const f = await seedFixture();
477      const automationId = await makeAutomation({
478        fixture: f,
479        name: "Depth limit",
480        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "must not fire" } }]
481      });
482      const te = {
483        ...recordCreatedEvent(f),
484        automationContext: { runId: "run_prev", automationId: "auto_someone_else", depth: 5 }
485      };
486      const runId = await makeRun(automationId, f, te);
487      const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
488      expect(status).toBe("SUCCEEDED");
489      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
490      expect((row.output as Record<string, unknown>).skipped).toBe("depth_limit");
491      expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(0);
492 
493      // Self-trigger beats depth limit (same event, both true).
494      const teSelf = {
495        ...recordCreatedEvent(f),
496        automationContext: { runId: "run_prev", automationId, depth: 5 }
497      };
498      const runId2 = await makeRun(automationId, f, teSelf);
499      const status2 = await handleRunAutomation(deps(), { data: jobData(runId2, automationId, f.workspaceId, teSelf) });
500      expect(status2).toBe("SUCCEEDED");
501      const row2 = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId2 } });
502      expect((row2.output as Record<string, unknown>).skipped).toBe("self_trigger");
503    });
504 
505    // ── 5. fatal error ────────────────────────────────────────────────────
506 
507    it("fatal error → FAILED immediately (no retry, no dead-letter, automation.executed FAILED)", async () => {
508      const f = await seedFixture();
509      const automationId = await makeAutomation({
510        fixture: f,
511        name: "Fatal",
512        actions: [{ type: "CREATE_PAGE", config: { title: "x" } }]
513      });
514      registerTestExecutor("CREATE_PAGE", {
515        type: "CREATE_PAGE",
516        execute: async () => {
517          throw new FatalError("BOOM", "deterministic fatal");
518        }
519      });
520      try {
521        const te = recordCreatedEvent(f);
522        const runId = await makeRun(automationId, f, te);
523        const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
524        expect(status).toBe("FAILED");
525        const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
526        expect(row.retryCount).toBe(0);
527        const err = row.error as Record<string, unknown>;
528        expect(err.code).toBe("BOOM");
529        expect(await redis.get(`automations:dead_letter:${f.workspaceId}`)).toBeNull();
530        const evts = await runEvents(runId);
531        const executed = evts.find((e) => e.event === "automation.executed") as { status?: string; error?: { code?: string } } | undefined;
532        expect(executed?.status).toBe("FAILED");
533        expect(executed?.error?.code).toBe("BOOM");
534        expect(await logCount(runId)).toBeGreaterThanOrEqual(2);
535      } finally {
536        clearTestExecutors();
537      }
538    });
539 
540    // ── 6. retry → dead letter (the seam-driven hermetic DLQ path) ───────
541 
542    it("retryable ×8 → 8 RETRYING hops (PENDING, retryCount 1..8) then DEAD_LETTER + notification + audit + event + counter", async () => {
543      const f = await seedFixture();
544      const automationId = await makeAutomation({
545        fixture: f,
546        name: "DLQ",
547        actions: [{ type: "CREATE_PAGE", config: { title: "x" } }]
548      });
549      registerTestExecutor("CREATE_PAGE", {
550        type: "CREATE_PAGE",
551        execute: async () => {
552          throw new RetryableError("HTTP_500", "deterministic retryable failure");
553        }
554      });
555      try {
556        const te = recordCreatedEvent(f);
557        const runId = await makeRun(automationId, f, te);
558 
559        // Hop 0 (initial, retryCount 0) + retries 1..7 → RETRYING.
560        for (let i = 0; i < 8; i++) {
561          const s = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
562          expect(s).toBe("RETRYING");
563          const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
564          expect(row.status).toBe("PENDING");
565          expect(row.retryCount).toBe(i + 1);
566        }
567 
568        // 9th drive (retryCount 8 = cap) → DEAD_LETTER.
569        const finalStatus = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
570        expect(finalStatus).toBe("DEAD_LETTER");
571 
572        const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
573        const err = row.error as Record<string, unknown>;
574        expect(err.code).toBe("HTTP_500");
575        expect(err.attempts).toBe(8);
576        expect(row.finishedAt).not.toBeNull();
577 
578        // Admin notification (workspace owner): "dead-lettered" is in the
579        // title; the body carries the final error.
580        const note = await prisma.notification.findFirst({ where: { workspaceId: f.workspaceId, type: "AUTOMATION" } });
581        expect(note).not.toBeNull();
582        expect(note!.title).toContain("dead-lettered");
583        expect(note!.body).toContain("HTTP_500");
584 
585        // Audit (system actor).
586        const audit = await prisma.auditLog.findFirst({ where: { workspaceId: f.workspaceId, action: "automation.dead_lettered", resourceId: runId } });
587        expect(audit).not.toBeNull();
588        expect(audit!.actorType).toBe("SYSTEM");
589 
590        // Counter + terminal event.
591        expect(await redis.get(`automations:dead_letter:${f.workspaceId}`)).toBe("1");
592        const evts = await runEvents(runId);
593        const executed = evts.find((e) => e.event === "automation.executed") as { status?: string; error?: { code?: string } } | undefined;
594        expect(executed?.status).toBe("DEAD_LETTER");
595        expect(executed?.error?.code).toBe("HTTP_500");
596 
597        // Every hop left a log.
598        expect(await logCount(runId)).toBeGreaterThanOrEqual(9);
599        expect(await prisma.automationLog.findFirst({ where: { runId, message: { contains: "dead-lettered" } } })).not.toBeNull();
600      } finally {
601        clearTestExecutors();
602      }
603    });
604 
605    // ── 7. 50-action guard ────────────────────────────────────────────────
606 
607    it("corrupt 60-action automation → FAILED RUN_ACTION_LIMIT (no actions run)", async () => {
608      const f = await seedFixture();
609      const automationId = await makeAutomation({
610        fixture: f,
611        name: "Action limit",
612        actions: Array.from({ length: 60 }, (_, i) => ({
613          type: "CREATE_NOTIFICATION",
614          config: { to: "automation_creator", type: "AUTOMATION", body: `n${i}` }
615        }))
616      });
617      const te = recordCreatedEvent(f);
618      const runId = await makeRun(automationId, f, te);
619      const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
620      expect(status).toBe("FAILED");
621      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
622      expect((row.error as Record<string, unknown>).code).toBe("RUN_ACTION_LIMIT");
623      expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(0);
624      expect(await logCount(runId)).toBeGreaterThanOrEqual(2);
625    });
626 
627    // ── 8. watchdog ───────────────────────────────────────────────────────
628 
629    it("watchdog: stale RUNNING (below cap) → PENDING + retryCount+1 (re-enqueued)", async () => {
630      const f = await seedFixture();
631      const automationId = await makeAutomation({
632        fixture: f,
633        name: "WD below cap",
634        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "x" } }]
635      });
636      const te = recordCreatedEvent(f);
637      const runId = await makeRun(automationId, f, te);
638      await prisma.automationRun.update({
639        where: { id: runId },
640        data: { status: "RUNNING", startedAt: new Date(Date.now() - 20 * 60_000) }
641      });
642 
643      const result = await runWatchdog(prisma, redis, env);
644      expect(result.reclaimed).toBeGreaterThanOrEqual(1);
645 
646      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
647      expect(row.status).toBe("PENDING");
648      expect(row.retryCount).toBe(1);
649      expect(
650        await prisma.automationLog.findFirst({ where: { runId, message: { contains: "reclaimed stale RUNNING" } } })
651      ).not.toBeNull();
652    });
653 
654    it("watchdog: stale RUNNING (at cap) → FAILED WORKER_CRASHED + notification + audit + counter", async () => {
655      const f = await seedFixture();
656      const automationId = await makeAutomation({
657        fixture: f,
658        name: "WD at cap",
659        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "x" } }]
660      });
661      const te = recordCreatedEvent(f);
662      const runId = await makeRun(automationId, f, te);
663      await prisma.automationRun.update({
664        where: { id: runId },
665        data: { status: "RUNNING", startedAt: new Date(Date.now() - 20 * 60_000), retryCount: 8 }
666      });
667 
668      const result = await runWatchdog(prisma, redis, env);
669      expect(result.deadLettered).toBeGreaterThanOrEqual(1);
670 
671      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
672      expect(row.status).toBe("FAILED");
673      expect((row.error as Record<string, unknown>).code).toBe("WORKER_CRASHED");
674      const note = await prisma.notification.findFirst({ where: { workspaceId: f.workspaceId, type: "AUTOMATION" } });
675      expect(note).not.toBeNull();
676      expect(note!.title).toContain("worker crash");
677      expect(
678        await prisma.auditLog.findFirst({ where: { workspaceId: f.workspaceId, action: "automation.worker_crashed" } })
679      ).not.toBeNull();
680      expect(await redis.get(`automations:dead_letter:${f.workspaceId}`)).toBe("1");
681    });
682 
683    it("watchdog: stale PENDING → re-enqueued WITHOUT retryCount bump", async () => {
684      const f = await seedFixture();
685      const automationId = await makeAutomation({
686        fixture: f,
687        name: "WD pending",
688        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "x" } }]
689      });
690      const te = recordCreatedEvent(f);
691      const runId = await makeRun(automationId, f, te);
692      await prisma.automationRun.update({
693        where: { id: runId },
694        data: { createdAt: new Date(Date.now() - 20 * 60_000) }
695      });
696 
697      const result = await runWatchdog(prisma, redis, env);
698      expect(result.reclaimed).toBeGreaterThanOrEqual(1);
699 
700      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
701      expect(row.status).toBe("PENDING");
702      expect(row.retryCount).toBe(0);
703      expect(
704        await prisma.automationLog.findFirst({ where: { runId, message: { contains: "re-enqueued stale PENDING" } } })
705      ).not.toBeNull();
706    });
707 
708    // ── 9. crash recovery (SIGKILL a real worker child mid-run) ───────────
709 
710    it("SIGKILL mid-run → row stuck RUNNING → watchdog reclaims → re-driven run SUCCEEDS", async () => {
711      const f = await seedFixture();
712      const automationId = await makeAutomation({
713        fixture: f,
714        name: "Crash recovery",
715        actions: [{ type: "CREATE_PAGE", config: { title: "crash page" } }]
716      });
717      const te = recordCreatedEvent(f);
718      const runId = await makeRun(automationId, f, te);
719 
720      // Child: a real BullMQ worker on the SAME queue whose executor never
721      // resolves (the wedged run). It logs "RUNNING" once the pipeline has
722      // flipped the row, then parks (an idle interval keeps it alive).
723      // Written into tests/integration so @prisma/client / ioredis / bullmq
724      // resolve from this package's node_modules.
725      const childScript = path.join(HERE, `.crash-child-${randomBytes(4).toString("hex")}.mts`);
726      writeFileSync(
727        childScript,
728        [
729          "import { PrismaClient } from \"@prisma/client\";",
730          "import { Redis } from \"ioredis\";",
731          "import { Worker } from \"bullmq\";",
732          "const dir = process.env.WORKER_SRC_DIR!;",
733          "const { handleRunAutomation } = await import(\"file://\" + dir + \"/processors/automations/pipeline.ts\");",
734          "const { registerTestExecutor } = await import(\"file://\" + dir + \"/processors/automations/executors/registry.ts\");",
735          "registerTestExecutor(\"CREATE_PAGE\", { type: \"CREATE_PAGE\", execute: async () => { await new Promise(() => {}); return { output: {} }; } });",
736          "const prisma = new PrismaClient({ datasources: { db: { url: process.env.DATABASE_URL! } } });",
737          "const redis = new Redis(process.env.REDIS_URL!, { maxRetriesPerRequest: null });",
738          "const w = new Worker(\"automations\", async (job) => { const s = await handleRunAutomation({ prisma, redis, env: { AUTO_RUN_MAX_RETRIES: 8 } as never }, job); return { status: s }; }, { connection: redis, concurrency: 1, lockDuration: 60_000, stalledInterval: 30_000 });",
739          "await w.waitUntilReady();",
740          "setInterval(() => {}, 1 << 30);",
741          ""
742        ].join("\n"),
743        "utf8"
744      );
745 
746      // Enqueue the run the same way the matcher does — the child's real
747      // BullMQ worker picks it up.
748      await enqueueRunJob(redis, { runId, automationId, workspaceId: f.workspaceId, triggerEvent: te, eventId: `crash_${randomBytes(4).toString("hex")}` });
749 
750      const tsxCli = path.resolve(HERE, "..", "..", "node_modules", "tsx", "dist", "cli.mjs");
751      const child = spawn(process.execPath, [tsxCli, childScript, runId], {
752        cwd: HERE,
753        stdio: ["ignore", "pipe", "pipe"],
754        env: { ...process.env, DATABASE_URL: dbUrl, REDIS_URL: redisUrl, WORKER_SRC_DIR: WORKER_SRC }
755      });
756      let errOut = "";
757      child.stderr.on("data", (d) => (errOut += String(d)));
758 
759      try {
760        // Wait for the child's worker to flip the row RUNNING.
761        await waitFor(
762          () => prisma.automationRun.findUnique({ where: { id: runId }, select: { status: true } }),
763          (r) => (r as { status: string } | null)?.status === "RUNNING",
764          60_000,
765          "row to reach RUNNING via the child worker"
766        );
767        const rowBefore = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
768        expect(rowBefore.startedAt).not.toBeNull();
769 
770        // SIGKILL — no shutdown hook, no lock release (the BullMQ lock
771        // expires later; the watchdog is the cross-restart backstop).
772        child.kill("SIGKILL");
773        await new Promise((r) => {
774          const t = setTimeout(() => r(undefined), 5_000);
775          child.once("exit", () => {
776            clearTimeout(t);
777            r(undefined);
778          });
779        });
780 
781        // The row is still RUNNING (the crash signature). Age it past the
782        // 15-min stale threshold (simulating the crash being discovered).
783        await prisma.automationRun.update({
784          where: { id: runId },
785          data: { startedAt: new Date(Date.now() - 20 * 60_000) }
786        });
787        await runWatchdog(prisma, redis, env);
788        const rowReclaimed = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
789        expect(rowReclaimed.status).toBe("PENDING");
790        expect(rowReclaimed.retryCount).toBe(1);
791 
792        // Re-drive the reclaimed run (a healthy worker re-executes from
793        // action 1) → SUCCEEDED.
794        registerTestExecutor("CREATE_PAGE", {
795          type: "CREATE_PAGE",
796          execute: async () => ({ output: { recovered: true } })
797        });
798        try {
799          const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
800          expect(status).toBe("SUCCEEDED");
801          const outRow = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
802          expect((outRow.output as Record<string, unknown>).actions).toBeDefined();
803        } finally {
804          clearTestExecutors();
805        }
806      } catch (err) {
807        child.kill("SIGKILL");
808        throw new Error(`crash test failed (child err: ${errOut.slice(-500) || "<empty>"}): ${(err as Error).message}`);
809      } finally {
810        rmSync(childScript, { force: true });
811      }
812    }, 150_000);
813 
814    // ── 10. end-to-end: stream → consumer → matcher → queue → worker ──────
815 
816    it("record.created on factory:events → consumer → matchAndEnqueue → run row → BullMQ worker → SUCCEEDED", async () => {
817      const f = await seedFixture();
818      const automationId = await makeAutomation({
819        fixture: f,
820        name: "E2E stream",
821        actions: [
822          { type: "UPDATE_RECORD", config: { recordId: "from_trigger", propertyValues: { Status: "opt-b" } } },
823          { type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "stream fired" } }
824        ]
825      });
826 
827      // Dedicated connections (StreamConsumer BLOCK + BullMQ semantics need
828      // their own clients — the worker's own index.ts follows the same rule).
829      const consumerRedis = new Redis(redisUrl, { maxRetriesPerRequest: null });
830      const workerRedis = new Redis(redisUrl, { maxRetriesPerRequest: null });
831      const consumer = processor.createAutomationsConsumer({ redis: consumerRedis, prisma, consumerId: "it-consumer" });
832      const worker = processor.createAutomationsWorker({ connection: workerRedis, prisma, env, concurrency: 1 });
833 
834      try {
835        await consumer.start();
836        await worker.waitUntilReady();
837 
838        // The consumer group is created from "0" (whole stream); earlier
839        // scenarios' record.updated events carry no matching RECORD_CREATED
840        // trigger, so only THIS entry fires the automation.
841        const entryId = await redis.xadd(
842          "factory:events",
843          "*",
844          "event",
845          "record.created",
846          "payload",
847          JSON.stringify({
848            workspaceId: f.workspaceId,
849            databaseId: f.databaseId,
850            recordId: f.recordId,
851            changedProperties: ["*"],
852            createdBy: f.ownerId
853          }),
854          "ts",
855          new Date().toISOString()
856        );
857        expect(typeof entryId).toBe("string");
858 
859        // Matcher decision counter (observability, §7).
860        await waitFor(
861          () => redis.get("automations:matched:record_created"),
862          (v) => v !== null && Number(v) >= 1,
863          15_000,
864          "matched counter"
865        );
866 
867        // The run row exists and the worker settles it.
868        const run = await waitFor(
869          () =>
870            prisma.automationRun.findFirst({
871              where: { automationId, workspaceId: f.workspaceId },
872              orderBy: { createdAt: "desc" }
873            }),
874          (r) => r !== null,
875          15_000,
876          "run row"
877        );
878        const runId = (run as { id: string }).id;
879        await waitFor(
880          () => statusOf(runId),
881          (s) => s === "SUCCEEDED" || s === "FAILED" || s === "DEAD_LETTER",
882          20_000,
883          "terminal run status"
884        );
885        expect(await statusOf(runId)).toBe("SUCCEEDED");
886 
887        // The full effect: property changed to B + notification + executed event.
888        const props = await prisma.recordProperty.findMany({
889          where: { recordId: f.recordId },
890          include: { property: { select: { name: true } } }
891        });
892        const statusProp = props.find((p) => p.property.name === "Status");
893        expect((statusProp?.value as Record<string, unknown>)?.optionId).toBe("opt-b");
894        expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(1);
895        const evts = await runEvents(runId);
896        expect(evts.find((e) => e.event === "automation.executed")).toBeDefined();
897        expect(await logCount(runId)).toBeGreaterThanOrEqual(4);
898      } finally {
899        await worker.close().catch(() => {});
900        await consumer.stop().catch(() => {});
901        await consumerRedis.quit().catch(() => {});
902        await workerRedis.quit().catch(() => {});
903      }
904    }, 60_000);
905  }
906);
907