/**
 * AI Software Factory — Integration: Automation pipeline (Phase 11, TASK-0113)
 *
 * The run pipeline (`handleRunAutomation`) against the REAL stack:
 * Testcontainers Postgres/Redis (shared container helpers). No MinIO /
 * OpenSearch / ClamAV needed — the scenarios use hermetic actions
 * (UPDATE_RECORD, CREATE_NOTIFICATION) plus the registry's test-injection
 * seam (`registerTestExecutor`) for the retry→dead-letter path. The seam is
 * the only production-code addition in Phase 11 for testability (Node's
 * undici fetch errors carry no classifiable errno, so a real HTTPS endpoint
 * cannot be made to fail deterministically; see docs/architecture/automation.md).
 *
 * Scenarios (docs/tasks/phase-11-automation.md TASK-0113 + §9 crash-safety):
 *   1.  happy path        → SUCCEEDED, property written, notification created,
 *                           automation.executed emitted, run logged
 *   2.  condition gate    → false: SUCCEEDED {skipped:"conditions_not_met"};
 *                           true: actions run (SELECT compared by option NAME)
 *   3.  self-trigger      → SUCCEEDED {skipped:"self_trigger"} (no actions)
 *   4.  depth limit       → SUCCEEDED {skipped:"depth_limit"} at depth 5
 *                           (self_trigger wins when both apply)
 *   5.  fatal error       → FAILED (no retry, no dead-letter)
 *   6.  retryable ×8      → 8 RETRYING hops (PENDING, retryCount 1..8) then
 *                           DEAD_LETTER + admin notification +
 *                           automation.dead_lettered audit + automation.executed
 *                           + automations:dead_letter:{ws} counter
 *   7.  60-action guard   → FAILED RUN_ACTION_LIMIT (corrupt-row defense)
 *   8.  watchdog          → stale RUNNING (below cap) → PENDING +retryCount;
 *                           stale RUNNING (at cap) → FAILED WORKER_CRASHED +
 *                           notification + audit + counter; stale PENDING →
 *                           re-enqueued without retryCount bump
 *   9.  crash recovery    → a REAL BullMQ worker child SIGKILL'd mid-run
 *                           leaves the row RUNNING; the watchdog reclaims it;
 *                           the re-driven run succeeds
 *  10.  end-to-end stream → record.created xadd → consumer (group
 *                           automations-matcher) → matchAndEnqueue → run row →
 *                           real BullMQ worker → SUCCEEDED (the
 *                           record-created→status-change worker-side half)
 *
 * Retry hops are driven by calling `handleRunAutomation` directly per attempt
 * (each `scheduleRetry` leaves the row PENDING with retryCount+1); the real
 * BullMQ backoff delays (1 s … 128 s) are NOT awaited — the re-enqueued
 * delayed jobs are harmless (no worker picks them up in time; the queue is
 * obliterated in afterAll).
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { spawn } from "node:child_process";
import { randomBytes } from "node:crypto";
import { writeFileSync, rmSync } from "node:fs";
import os from "node:os";
import path from "node:path";
import { fileURLToPath, pathToFileURL } from "node:url";
import { Prisma, PrismaClient } from "@prisma/client";
import { Redis } from "ioredis";
import { getDatabaseUrl, getRedisUrl } from "./helpers/containers";
import { deploySchema } from "./helpers/api";

const HERE = path.dirname(fileURLToPath(import.meta.url));
const WORKER_SRC = path.resolve(HERE, "..", "..", "apps", "worker", "src");
const REGISTRY_DIR = path.join(WORKER_SRC, "processors", "automations");

function fileUrl(p: string): string {
  return `file://${p.startsWith("/") ? p : `/${p}`}`;
}

// Worker pipeline + seams from SOURCE (the worker ships no build for the
// monorepo test path — same source-import convention as files-pipeline).
const { handleRunAutomation } = await import(
  /* webpackIgnore: true */
  fileUrl(path.join(REGISTRY_DIR, "pipeline.ts"))
);
const { enqueueRunJob } = await import(
  /* webpackIgnore: true */
  fileUrl(path.join(REGISTRY_DIR, "matcher.ts"))
);
const { runWatchdog } = await import(
  /* webpackIgnore: true */
  fileUrl(path.join(REGISTRY_DIR, "watchdog.ts"))
);
const { registerTestExecutor, clearTestExecutors } = await import(
  /* webpackIgnore: true */
  fileUrl(path.join(REGISTRY_DIR, "executors", "registry.ts"))
);
const { FatalError, RetryableError } = await import(
  /* webpackIgnore: true */
  fileUrl(path.join(REGISTRY_DIR, "executors", "classify.ts"))
);
const processor = await import(
  /* webpackIgnore: true */
  fileUrl(path.join(WORKER_SRC, "processors", "automations.processor.ts"))
);

// ── infra gate ──────────────────────────────────────────────────────────────

let dbUrl = "";
let redisUrl = "";
let infraReady = false;
try {
  [dbUrl, redisUrl] = await Promise.all([getDatabaseUrl(), getRedisUrl()]);
  infraReady = true;
} catch {
  infraReady = false; // no Docker daemon (or container image unavailable)
}

// ── state ───────────────────────────────────────────────────────────────────

let prisma: PrismaClient;
let redis: Redis;
let env: import("../../apps/worker/src/config").WorkerEnv;

function baseEnv(): import("../../apps/worker/src/config").WorkerEnv {
  return {
    NODE_ENV: "test",
    REDIS_URL: redisUrl,
    DATABASE_URL: dbUrl,
    S3_ENDPOINT: "http://localhost:9000",
    S3_REGION: "us-east-1",
    S3_BUCKET_FILES: "factory-files",
    S3_BUCKET_THUMBS: "factory-thumbs",
    S3_BUCKET: "factory-files",
    S3_ACCESS_KEY: "minioadmin",
    S3_SECRET_KEY: "minioadmin",
    S3_PUBLIC_BASE: "http://localhost:9001",
    SEARCH_URL: "http://localhost:9200",
    SEARCH_USERNAME: "admin",
    SEARCH_PASSWORD: "changeme",
    SEARCH_VECTOR_DIMENSION: 1536,
    SEARCH_KNN_M: 16,
    SEARCH_KNN_EF_CONSTRUCTION: 100,
    MASTER_ENCRYPTION_KEY: "",
    AI_API_KEY: "",
    AI_PROVIDER: "openai",
    AI_BASE_URL: "",
    AI_MODEL: "gpt-4o-mini",
    AI_EMBEDDING_PROVIDER: "",
    AI_EMBEDDING_MODEL: "text-embedding-3-small",
    SMTP_HOST: "", // no mail — the Notification row is the assertion
    SMTP_PORT: 587,
    SMTP_USER: "",
    SMTP_PASS: "",
    SMTP_FROM: "noreply@factory.local",
    SMTP_STARTTLS: true,
    CLAMAV_HOST: "",
    CLAMAV_PORT: 3310,
    CLAMD_SOCKET: "",
    FILES_WORKER_CONCURRENCY: 4,
    CLAMAV_TIMEOUT_MS: 30_000,
    FILES_TEMP_DIR: os.tmpdir(),
    FILES_TEMP_MAX_AGE_MS: 600_000,
    FILES_TEMP_FREE_MIN_BYTES: 1,
    PUBLIC_URL: "http://localhost:3000",
    WORKER_CONCURRENCY: 10,
    WORKER_ID: "pipeline-it",
    WORKER_RETENTION_EXECUTE: false,
    HEALTH_PORT: 4003,
    AUTO_RUN_MAX_RETRIES: 8,
    AUTOMATIONS_WORKER_CONCURRENCY: 1
  } as import("../../apps/worker/src/config").WorkerEnv;
}

// ── fixtures ────────────────────────────────────────────────────────────────

interface Fixture {
  ownerId: string;
  workspaceId: string;
  databaseId: string;
  recordId: string;
}

/** User (OWNER member) + workspace + database + record with a TITLE and a
 *  SELECT "Status" property (options A/B). The SELECT value is stored as the
 *  Phase-5 envelope { optionId }; the condition evaluator normalizes it to
 *  the option NAME. The OWNER membership is required by the worker's
 *  permission walk (role baseline) AND is the DLQ notification recipient. */
async function seedFixture(): Promise<Fixture> {
  const ownerId = `user_ap_${randomBytes(6).toString("hex")}`;
  const workspaceId = `wsp_ap_${randomBytes(6).toString("hex")}`;
  const databaseId = `db_ap_${randomBytes(6).toString("hex")}`;
  const recordId = `rec_ap_${randomBytes(6).toString("hex")}`;
  const titlePropId = `prop_t_${randomBytes(4).toString("hex")}`;
  const statusPropId = `prop_s_${randomBytes(4).toString("hex")}`;
  await prisma.user.create({ data: { id: ownerId, email: `${ownerId}@factory.local`, name: "Pipeline Owner" } });
  await prisma.workspace.create({
    data: { id: workspaceId, name: `WS ${workspaceId}`, slug: `slug-${workspaceId}`, ownerId, billingPlan: "paid" }
  });
  await prisma.workspaceMember.create({ data: { workspaceId, userId: ownerId, role: "OWNER" } });
  await prisma.database.create({ data: { id: databaseId, workspaceId, title: "Pipeline DB" } });
  await prisma.databaseProperty.create({
    data: { id: titlePropId, databaseId, name: "Name", type: "TITLE", isTitle: true }
  });
  await prisma.databaseProperty.create({
    data: {
      id: statusPropId,
      databaseId,
      name: "Status",
      type: "SELECT",
      config: { options: [{ id: "opt-a", name: "A" }, { id: "opt-b", name: "B" }] } as Prisma.InputJsonValue
    }
  });
  await prisma.databaseRecord.create({ data: { id: recordId, workspaceId, databaseId, createdBy: ownerId } });
  await prisma.recordProperty.create({
    data: { recordId, propertyId: statusPropId, value: { optionId: "opt-a" } as Prisma.InputJsonValue }
  });
  return { ownerId, workspaceId, databaseId, recordId };
}

async function makeAutomation(opts: {
  fixture: Fixture;
  name: string;
  actions: Array<{ type: string; config: Record<string, unknown> }>;
  condition?: unknown;
  triggerType?: string;
  triggerConfig?: Record<string, unknown>;
}): Promise<string> {
  const id = `auto_ap_${randomBytes(6).toString("hex")}`;
  const { fixture } = opts;
  await prisma.automation.create({
    data: {
      id,
      workspaceId: fixture.workspaceId,
      name: opts.name,
      status: "ACTIVE",
      databaseId: fixture.databaseId,
      createdById: fixture.ownerId,
      triggers: {
        create: [
          {
            type: (opts.triggerType ?? "RECORD_CREATED") as never,
            config: { databaseId: fixture.databaseId, ...(opts.triggerConfig ?? {}) } as Prisma.InputJsonValue,
            position: 0
          }
        ]
      },
      ...(opts.condition !== undefined
        ? { conditions: { create: [{ expression: opts.condition as Prisma.InputJsonValue, position: 0 }] } }
        : {})
    }
  });
  for (let i = 0; i < opts.actions.length; i++) {
    const a = opts.actions[i]!;
    await prisma.automationAction.create({
      data: {
        automationId: id,
        type: a.type as never,
        config: a.config as Prisma.InputJsonValue,
        position: i,
        retries: 3,
        timeoutMs: 30_000
      }
    });
  }
  return id;
}

/** A PENDING run row for a direct pipeline drive (the matcher's shape). */
async function makeRun(automationId: string, fixture: Fixture, triggerEvent: Record<string, unknown>): Promise<string> {
  const runId = `run_ap_${randomBytes(6).toString("hex")}`;
  await prisma.automationRun.create({
    data: {
      id: runId,
      workspaceId: fixture.workspaceId,
      automationId,
      status: "PENDING",
      triggerEvent: triggerEvent as Prisma.InputJsonValue,
      input: { triggerEvent, actionCount: 0 } as Prisma.InputJsonValue,
      retryCount: 0
    }
  });
  return runId;
}

function jobData(runId: string, automationId: string, workspaceId: string, triggerEvent: Record<string, unknown>) {
  return { runId, automationId, workspaceId, triggerEvent, eventId: `evt_${randomBytes(4).toString("hex")}` };
}

function recordCreatedEvent(fixture: Fixture): Record<string, unknown> {
  const payload = {
    workspaceId: fixture.workspaceId,
    databaseId: fixture.databaseId,
    recordId: fixture.recordId,
    changedProperties: ["*"],
    createdBy: fixture.ownerId
  };
  // buildTriggerEvent's shape: { type, payload, ...flattened payload }.
  return { type: "record.created", payload, ...payload };
}

const deps = () => ({ prisma, redis, env });

/** All `factory:events` entries carrying one runId (decoded payloads). */
async function runEvents(runId: string): Promise<Array<Record<string, unknown>>> {
  const raw = await redis.xrange("factory:events", "-", "+");
  const out: Array<Record<string, unknown>> = [];
  for (const [, fields] of raw) {
    const kv: Record<string, string> = {};
    for (let i = 0; i < fields.length; i += 2) kv[fields[i]!] = fields[i + 1]!;
    if (kv.payload) {
      try {
        const p = JSON.parse(kv.payload) as Record<string, unknown>;
        if (p.runId === runId) out.push({ event: kv.event, ...p });
      } catch {
        /* not JSON — ignore */
      }
    }
  }
  return out;
}

const logCount = (runId: string) => prisma.automationLog.count({ where: { runId } });

async function statusOf(runId: string): Promise<string> {
  const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
  return row.status;
}

async function waitFor<T>(
  fn: () => Promise<T>,
  pred: (v: T) => boolean,
  timeoutMs = 20_000,
  what = "condition"
): Promise<T> {
  const t0 = Date.now();
  for (;;) {
    const v = await fn();
    if (pred(v)) return v;
    if (Date.now() - t0 > timeoutMs) throw new Error(`timed out waiting for ${what}`);
    await new Promise((r) => setTimeout(r, 250));
  }
}

// ── suite ───────────────────────────────────────────────────────────────────

describe.skipIf(!infraReady)(
  "Automation pipeline integration (TASK-0113 — worker, real stack)",
  () => {
    beforeAll(async () => {
      await deploySchema(dbUrl);
      prisma = new PrismaClient({ datasources: { db: { url: dbUrl } } });
      await prisma.$connect();
      redis = new Redis(redisUrl, { maxRetriesPerRequest: null });
      env = baseEnv();
    }, 180_000);

    afterAll(async () => {
      clearTestExecutors();
      // Wipe the automations queue (delayed retry jobs left by the retry
      // scenario are still parked with real backoff delays).
      try {
        const { Queue } = await import("bullmq");
        const q = new Queue("automations", { connection: redis });
        await q.obliterate({ force: true });
        await q.close();
      } catch {
        /* best-effort */
      }
      await redis?.quit().catch(() => {});
      await prisma?.$disconnect().catch(() => {});
    });

    // ── 1. happy path (direct driver) ────────────────────────────────────

    it("record.created → UPDATE_RECORD + CREATE_NOTIFICATION → SUCCEEDED (property written, notification, automation.executed, logged)", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "Happy path",
        actions: [
          { type: "UPDATE_RECORD", config: { recordId: "from_trigger", propertyValues: { Status: "opt-b" } } },
          { type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "Pipeline test notification" } }
        ]
      });
      const te = recordCreatedEvent(f);
      const runId = await makeRun(automationId, f, te);

      const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
      expect(status).toBe("SUCCEEDED");

      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
      expect(row.retryCount).toBe(0);
      expect(row.startedAt).not.toBeNull();
      expect(row.finishedAt).not.toBeNull();
      const output = row.output as Record<string, unknown>;
      const actions = output.actions as Array<Record<string, unknown>>;
      expect(actions).toHaveLength(2);
      expect(actions[0]!.status).toBe("succeeded");
      expect(actions[1]!.status).toBe("succeeded");

      // The property write landed as the Phase-5 envelope.
      const props = await prisma.recordProperty.findMany({
        where: { recordId: f.recordId },
        include: { property: { select: { name: true } } }
      });
      const statusProp = props.find((p) => p.property.name === "Status");
      expect((statusProp?.value as Record<string, unknown>)?.optionId).toBe("opt-b");

      // The notification was created for the automation's creator (owner).
      const note = await prisma.notification.findFirst({ where: { workspaceId: f.workspaceId, userId: f.ownerId } });
      expect(note).not.toBeNull();
      expect(note!.body).toBe("Pipeline test notification");

      // Depth-0 terminal events carry automation.executed.
      const evts = await runEvents(runId);
      const executed = evts.find((e) => e.event === "automation.executed") as { status?: string } | undefined;
      expect(executed).toBeDefined();
      expect(executed?.status).toBe("SUCCEEDED");

      // Every run is logged (started + actions + completed).
      expect(await logCount(runId)).toBeGreaterThanOrEqual(4);
    });

    // ── 2. condition gate ─────────────────────────────────────────────────

    it("condition gate: SELECT property compared by option NAME — false skips, true runs", async () => {
      const f = await seedFixture();

      // Record Status = A. Condition "Status eq B" → false.
      const gateFalse = await makeAutomation({
        fixture: f,
        name: "Gate false",
        condition: { field: { ref: "property", name: "Status" }, operator: "eq", value: "B" },
        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "should not fire" } }]
      });
      const teF = recordCreatedEvent(f);
      const runF = await makeRun(gateFalse, f, teF);
      const sF = await handleRunAutomation(deps(), { data: jobData(runF, gateFalse, f.workspaceId, teF) });
      expect(sF).toBe("SUCCEEDED");
      const rowF = await prisma.automationRun.findUniqueOrThrow({ where: { id: runF } });
      expect((rowF.output as Record<string, unknown>).skipped).toBe("conditions_not_met");
      expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(0);
      expect(await logCount(runF)).toBeGreaterThanOrEqual(1);

      // Same record, condition "Status eq A" → true → the action runs.
      const gateTrue = await makeAutomation({
        fixture: f,
        name: "Gate true",
        condition: { field: { ref: "property", name: "Status" }, operator: "eq", value: "A" },
        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "fired" } }]
      });
      const teT = recordCreatedEvent(f);
      const runT = await makeRun(gateTrue, f, teT);
      const sT = await handleRunAutomation(deps(), { data: jobData(runT, gateTrue, f.workspaceId, teT) });
      expect(sT).toBe("SUCCEEDED");
      const rowT = await prisma.automationRun.findUniqueOrThrow({ where: { id: runT } });
      const outT = rowT.output as Record<string, unknown>;
      expect(outT.skipped).toBeUndefined();
      expect((outT.actions as unknown[])).toHaveLength(1);
      expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(1);
    });

    // ── 3. loop prevention ────────────────────────────────────────────────

    it("self-trigger block: the automation's own event is skipped (SUCCEEDED {skipped:'self_trigger'})", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "Self trigger",
        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "must not fire" } }]
      });
      const te = {
        ...recordCreatedEvent(f),
        automationContext: { runId: "run_prev", automationId, depth: 1 }
      };
      const runId = await makeRun(automationId, f, te);
      const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
      expect(status).toBe("SUCCEEDED");
      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
      expect((row.output as Record<string, unknown>).skipped).toBe("self_trigger");
      expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(0);
      expect(await logCount(runId)).toBeGreaterThanOrEqual(1);
      // automation.executed is depth-0 ONLY (§8 event-storm guard): this run
      // was triggered by the automation's own depth-1 event, so NO executed
      // event is emitted for it (the root run's event already covered the chain).
      expect((await runEvents(runId)).find((e) => e.event === "automation.executed")).toBeUndefined();
    });

    it("depth limit: events at depth ≥ 5 are skipped (SUCCEEDED {skipped:'depth_limit'}); self-trigger wins", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "Depth limit",
        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "must not fire" } }]
      });
      const te = {
        ...recordCreatedEvent(f),
        automationContext: { runId: "run_prev", automationId: "auto_someone_else", depth: 5 }
      };
      const runId = await makeRun(automationId, f, te);
      const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
      expect(status).toBe("SUCCEEDED");
      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
      expect((row.output as Record<string, unknown>).skipped).toBe("depth_limit");
      expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(0);

      // Self-trigger beats depth limit (same event, both true).
      const teSelf = {
        ...recordCreatedEvent(f),
        automationContext: { runId: "run_prev", automationId, depth: 5 }
      };
      const runId2 = await makeRun(automationId, f, teSelf);
      const status2 = await handleRunAutomation(deps(), { data: jobData(runId2, automationId, f.workspaceId, teSelf) });
      expect(status2).toBe("SUCCEEDED");
      const row2 = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId2 } });
      expect((row2.output as Record<string, unknown>).skipped).toBe("self_trigger");
    });

    // ── 5. fatal error ────────────────────────────────────────────────────

    it("fatal error → FAILED immediately (no retry, no dead-letter, automation.executed FAILED)", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "Fatal",
        actions: [{ type: "CREATE_PAGE", config: { title: "x" } }]
      });
      registerTestExecutor("CREATE_PAGE", {
        type: "CREATE_PAGE",
        execute: async () => {
          throw new FatalError("BOOM", "deterministic fatal");
        }
      });
      try {
        const te = recordCreatedEvent(f);
        const runId = await makeRun(automationId, f, te);
        const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
        expect(status).toBe("FAILED");
        const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
        expect(row.retryCount).toBe(0);
        const err = row.error as Record<string, unknown>;
        expect(err.code).toBe("BOOM");
        expect(await redis.get(`automations:dead_letter:${f.workspaceId}`)).toBeNull();
        const evts = await runEvents(runId);
        const executed = evts.find((e) => e.event === "automation.executed") as { status?: string; error?: { code?: string } } | undefined;
        expect(executed?.status).toBe("FAILED");
        expect(executed?.error?.code).toBe("BOOM");
        expect(await logCount(runId)).toBeGreaterThanOrEqual(2);
      } finally {
        clearTestExecutors();
      }
    });

    // ── 6. retry → dead letter (the seam-driven hermetic DLQ path) ───────

    it("retryable ×8 → 8 RETRYING hops (PENDING, retryCount 1..8) then DEAD_LETTER + notification + audit + event + counter", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "DLQ",
        actions: [{ type: "CREATE_PAGE", config: { title: "x" } }]
      });
      registerTestExecutor("CREATE_PAGE", {
        type: "CREATE_PAGE",
        execute: async () => {
          throw new RetryableError("HTTP_500", "deterministic retryable failure");
        }
      });
      try {
        const te = recordCreatedEvent(f);
        const runId = await makeRun(automationId, f, te);

        // Hop 0 (initial, retryCount 0) + retries 1..7 → RETRYING.
        for (let i = 0; i < 8; i++) {
          const s = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
          expect(s).toBe("RETRYING");
          const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
          expect(row.status).toBe("PENDING");
          expect(row.retryCount).toBe(i + 1);
        }

        // 9th drive (retryCount 8 = cap) → DEAD_LETTER.
        const finalStatus = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
        expect(finalStatus).toBe("DEAD_LETTER");

        const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
        const err = row.error as Record<string, unknown>;
        expect(err.code).toBe("HTTP_500");
        expect(err.attempts).toBe(8);
        expect(row.finishedAt).not.toBeNull();

        // Admin notification (workspace owner): "dead-lettered" is in the
        // title; the body carries the final error.
        const note = await prisma.notification.findFirst({ where: { workspaceId: f.workspaceId, type: "AUTOMATION" } });
        expect(note).not.toBeNull();
        expect(note!.title).toContain("dead-lettered");
        expect(note!.body).toContain("HTTP_500");

        // Audit (system actor).
        const audit = await prisma.auditLog.findFirst({ where: { workspaceId: f.workspaceId, action: "automation.dead_lettered", resourceId: runId } });
        expect(audit).not.toBeNull();
        expect(audit!.actorType).toBe("SYSTEM");

        // Counter + terminal event.
        expect(await redis.get(`automations:dead_letter:${f.workspaceId}`)).toBe("1");
        const evts = await runEvents(runId);
        const executed = evts.find((e) => e.event === "automation.executed") as { status?: string; error?: { code?: string } } | undefined;
        expect(executed?.status).toBe("DEAD_LETTER");
        expect(executed?.error?.code).toBe("HTTP_500");

        // Every hop left a log.
        expect(await logCount(runId)).toBeGreaterThanOrEqual(9);
        expect(await prisma.automationLog.findFirst({ where: { runId, message: { contains: "dead-lettered" } } })).not.toBeNull();
      } finally {
        clearTestExecutors();
      }
    });

    // ── 7. 50-action guard ────────────────────────────────────────────────

    it("corrupt 60-action automation → FAILED RUN_ACTION_LIMIT (no actions run)", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "Action limit",
        actions: Array.from({ length: 60 }, (_, i) => ({
          type: "CREATE_NOTIFICATION",
          config: { to: "automation_creator", type: "AUTOMATION", body: `n${i}` }
        }))
      });
      const te = recordCreatedEvent(f);
      const runId = await makeRun(automationId, f, te);
      const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
      expect(status).toBe("FAILED");
      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
      expect((row.error as Record<string, unknown>).code).toBe("RUN_ACTION_LIMIT");
      expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(0);
      expect(await logCount(runId)).toBeGreaterThanOrEqual(2);
    });

    // ── 8. watchdog ───────────────────────────────────────────────────────

    it("watchdog: stale RUNNING (below cap) → PENDING + retryCount+1 (re-enqueued)", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "WD below cap",
        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "x" } }]
      });
      const te = recordCreatedEvent(f);
      const runId = await makeRun(automationId, f, te);
      await prisma.automationRun.update({
        where: { id: runId },
        data: { status: "RUNNING", startedAt: new Date(Date.now() - 20 * 60_000) }
      });

      const result = await runWatchdog(prisma, redis, env);
      expect(result.reclaimed).toBeGreaterThanOrEqual(1);

      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
      expect(row.status).toBe("PENDING");
      expect(row.retryCount).toBe(1);
      expect(
        await prisma.automationLog.findFirst({ where: { runId, message: { contains: "reclaimed stale RUNNING" } } })
      ).not.toBeNull();
    });

    it("watchdog: stale RUNNING (at cap) → FAILED WORKER_CRASHED + notification + audit + counter", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "WD at cap",
        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "x" } }]
      });
      const te = recordCreatedEvent(f);
      const runId = await makeRun(automationId, f, te);
      await prisma.automationRun.update({
        where: { id: runId },
        data: { status: "RUNNING", startedAt: new Date(Date.now() - 20 * 60_000), retryCount: 8 }
      });

      const result = await runWatchdog(prisma, redis, env);
      expect(result.deadLettered).toBeGreaterThanOrEqual(1);

      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
      expect(row.status).toBe("FAILED");
      expect((row.error as Record<string, unknown>).code).toBe("WORKER_CRASHED");
      const note = await prisma.notification.findFirst({ where: { workspaceId: f.workspaceId, type: "AUTOMATION" } });
      expect(note).not.toBeNull();
      expect(note!.title).toContain("worker crash");
      expect(
        await prisma.auditLog.findFirst({ where: { workspaceId: f.workspaceId, action: "automation.worker_crashed" } })
      ).not.toBeNull();
      expect(await redis.get(`automations:dead_letter:${f.workspaceId}`)).toBe("1");
    });

    it("watchdog: stale PENDING → re-enqueued WITHOUT retryCount bump", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "WD pending",
        actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "x" } }]
      });
      const te = recordCreatedEvent(f);
      const runId = await makeRun(automationId, f, te);
      await prisma.automationRun.update({
        where: { id: runId },
        data: { createdAt: new Date(Date.now() - 20 * 60_000) }
      });

      const result = await runWatchdog(prisma, redis, env);
      expect(result.reclaimed).toBeGreaterThanOrEqual(1);

      const row = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
      expect(row.status).toBe("PENDING");
      expect(row.retryCount).toBe(0);
      expect(
        await prisma.automationLog.findFirst({ where: { runId, message: { contains: "re-enqueued stale PENDING" } } })
      ).not.toBeNull();
    });

    // ── 9. crash recovery (SIGKILL a real worker child mid-run) ───────────

    it("SIGKILL mid-run → row stuck RUNNING → watchdog reclaims → re-driven run SUCCEEDS", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "Crash recovery",
        actions: [{ type: "CREATE_PAGE", config: { title: "crash page" } }]
      });
      const te = recordCreatedEvent(f);
      const runId = await makeRun(automationId, f, te);

      // Child: a real BullMQ worker on the SAME queue whose executor never
      // resolves (the wedged run). It logs "RUNNING" once the pipeline has
      // flipped the row, then parks (an idle interval keeps it alive).
      // Written into tests/integration so @prisma/client / ioredis / bullmq
      // resolve from this package's node_modules.
      const childScript = path.join(HERE, `.crash-child-${randomBytes(4).toString("hex")}.mts`);
      writeFileSync(
        childScript,
        [
          "import { PrismaClient } from \"@prisma/client\";",
          "import { Redis } from \"ioredis\";",
          "import { Worker } from \"bullmq\";",
          "const dir = process.env.WORKER_SRC_DIR!;",
          "const { handleRunAutomation } = await import(\"file://\" + dir + \"/processors/automations/pipeline.ts\");",
          "const { registerTestExecutor } = await import(\"file://\" + dir + \"/processors/automations/executors/registry.ts\");",
          "registerTestExecutor(\"CREATE_PAGE\", { type: \"CREATE_PAGE\", execute: async () => { await new Promise(() => {}); return { output: {} }; } });",
          "const prisma = new PrismaClient({ datasources: { db: { url: process.env.DATABASE_URL! } } });",
          "const redis = new Redis(process.env.REDIS_URL!, { maxRetriesPerRequest: null });",
          "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 });",
          "await w.waitUntilReady();",
          "setInterval(() => {}, 1 << 30);",
          ""
        ].join("\n"),
        "utf8"
      );

      // Enqueue the run the same way the matcher does — the child's real
      // BullMQ worker picks it up.
      await enqueueRunJob(redis, { runId, automationId, workspaceId: f.workspaceId, triggerEvent: te, eventId: `crash_${randomBytes(4).toString("hex")}` });

      const tsxCli = path.resolve(HERE, "..", "..", "node_modules", "tsx", "dist", "cli.mjs");
      const child = spawn(process.execPath, [tsxCli, childScript, runId], {
        cwd: HERE,
        stdio: ["ignore", "pipe", "pipe"],
        env: { ...process.env, DATABASE_URL: dbUrl, REDIS_URL: redisUrl, WORKER_SRC_DIR: WORKER_SRC }
      });
      let errOut = "";
      child.stderr.on("data", (d) => (errOut += String(d)));

      try {
        // Wait for the child's worker to flip the row RUNNING.
        await waitFor(
          () => prisma.automationRun.findUnique({ where: { id: runId }, select: { status: true } }),
          (r) => (r as { status: string } | null)?.status === "RUNNING",
          60_000,
          "row to reach RUNNING via the child worker"
        );
        const rowBefore = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
        expect(rowBefore.startedAt).not.toBeNull();

        // SIGKILL — no shutdown hook, no lock release (the BullMQ lock
        // expires later; the watchdog is the cross-restart backstop).
        child.kill("SIGKILL");
        await new Promise((r) => {
          const t = setTimeout(() => r(undefined), 5_000);
          child.once("exit", () => {
            clearTimeout(t);
            r(undefined);
          });
        });

        // The row is still RUNNING (the crash signature). Age it past the
        // 15-min stale threshold (simulating the crash being discovered).
        await prisma.automationRun.update({
          where: { id: runId },
          data: { startedAt: new Date(Date.now() - 20 * 60_000) }
        });
        await runWatchdog(prisma, redis, env);
        const rowReclaimed = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
        expect(rowReclaimed.status).toBe("PENDING");
        expect(rowReclaimed.retryCount).toBe(1);

        // Re-drive the reclaimed run (a healthy worker re-executes from
        // action 1) → SUCCEEDED.
        registerTestExecutor("CREATE_PAGE", {
          type: "CREATE_PAGE",
          execute: async () => ({ output: { recovered: true } })
        });
        try {
          const status = await handleRunAutomation(deps(), { data: jobData(runId, automationId, f.workspaceId, te) });
          expect(status).toBe("SUCCEEDED");
          const outRow = await prisma.automationRun.findUniqueOrThrow({ where: { id: runId } });
          expect((outRow.output as Record<string, unknown>).actions).toBeDefined();
        } finally {
          clearTestExecutors();
        }
      } catch (err) {
        child.kill("SIGKILL");
        throw new Error(`crash test failed (child err: ${errOut.slice(-500) || "<empty>"}): ${(err as Error).message}`);
      } finally {
        rmSync(childScript, { force: true });
      }
    }, 150_000);

    // ── 10. end-to-end: stream → consumer → matcher → queue → worker ──────

    it("record.created on factory:events → consumer → matchAndEnqueue → run row → BullMQ worker → SUCCEEDED", async () => {
      const f = await seedFixture();
      const automationId = await makeAutomation({
        fixture: f,
        name: "E2E stream",
        actions: [
          { type: "UPDATE_RECORD", config: { recordId: "from_trigger", propertyValues: { Status: "opt-b" } } },
          { type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "stream fired" } }
        ]
      });

      // Dedicated connections (StreamConsumer BLOCK + BullMQ semantics need
      // their own clients — the worker's own index.ts follows the same rule).
      const consumerRedis = new Redis(redisUrl, { maxRetriesPerRequest: null });
      const workerRedis = new Redis(redisUrl, { maxRetriesPerRequest: null });
      const consumer = processor.createAutomationsConsumer({ redis: consumerRedis, prisma, consumerId: "it-consumer" });
      const worker = processor.createAutomationsWorker({ connection: workerRedis, prisma, env, concurrency: 1 });

      try {
        await consumer.start();
        await worker.waitUntilReady();

        // The consumer group is created from "0" (whole stream); earlier
        // scenarios' record.updated events carry no matching RECORD_CREATED
        // trigger, so only THIS entry fires the automation.
        const entryId = await redis.xadd(
          "factory:events",
          "*",
          "event",
          "record.created",
          "payload",
          JSON.stringify({
            workspaceId: f.workspaceId,
            databaseId: f.databaseId,
            recordId: f.recordId,
            changedProperties: ["*"],
            createdBy: f.ownerId
          }),
          "ts",
          new Date().toISOString()
        );
        expect(typeof entryId).toBe("string");

        // Matcher decision counter (observability, §7).
        await waitFor(
          () => redis.get("automations:matched:record_created"),
          (v) => v !== null && Number(v) >= 1,
          15_000,
          "matched counter"
        );

        // The run row exists and the worker settles it.
        const run = await waitFor(
          () =>
            prisma.automationRun.findFirst({
              where: { automationId, workspaceId: f.workspaceId },
              orderBy: { createdAt: "desc" }
            }),
          (r) => r !== null,
          15_000,
          "run row"
        );
        const runId = (run as { id: string }).id;
        await waitFor(
          () => statusOf(runId),
          (s) => s === "SUCCEEDED" || s === "FAILED" || s === "DEAD_LETTER",
          20_000,
          "terminal run status"
        );
        expect(await statusOf(runId)).toBe("SUCCEEDED");

        // The full effect: property changed to B + notification + executed event.
        const props = await prisma.recordProperty.findMany({
          where: { recordId: f.recordId },
          include: { property: { select: { name: true } } }
        });
        const statusProp = props.find((p) => p.property.name === "Status");
        expect((statusProp?.value as Record<string, unknown>)?.optionId).toBe("opt-b");
        expect(await prisma.notification.count({ where: { workspaceId: f.workspaceId } })).toBe(1);
        const evts = await runEvents(runId);
        expect(evts.find((e) => e.event === "automation.executed")).toBeDefined();
        expect(await logCount(runId)).toBeGreaterThanOrEqual(4);
      } finally {
        await worker.close().catch(() => {});
        await consumer.stop().catch(() => {});
        await consumerRedis.quit().catch(() => {});
        await workerRedis.quit().catch(() => {});
      }
    }, 60_000);
  }
);