/**
 * AI Software Factory — Phase 11 automations E2E (TASK-0114).
 *
 * FULL STACK REQUIRED — this spec drives the REAL end-to-end path:
 *   browser (Next.js :3000) + API (:4000) + worker (tsx, real BullMQ
 *   `automations` worker + `factory:events` stream consumer + 5-min
 *   schedule reconciler) + Postgres + Redis.
 *
 *   pnpm docker:up && pnpm dev   (api :4000, web :3000, worker)
 *   DATABASE_URL=... pnpm test:e2e -- automation
 *
 * There is intentionally NO `webServer` block (the whole stack must be
 * running — see playwright.config.ts header).
 *
 * Scenarios (ROADMAP Phase 11 exit criteria):
 *   1. record.created → UPDATE_RECORD + CREATE_NOTIFICATION → SUCCEEDED
 *      with the property changed in < 10 s; the run + its FULL LOGS are
 *      visible in the UI (settings → Automations → row → Runs → expand).
 *   2. condition gate — a property condition (option NAME) skips the
 *      non-matching record (output.skipped = "conditions_not_met") and
 *      executes the matching one.
 *   3. schedule (cron) trigger fires → a run with triggerType "schedule"
 *      (the worker-side reconciler registers the repeatable; we invoke it
 *      directly via the worker source so the test does not wait for the
 *      5-minute cycle — the cron resolution still happens in real BullMQ).
 *   4. pause stops execution — PAUSED automations match nothing;
 *      re-activating resumes matching.
 *   5. failure → FAILED run with the error surfaced in the UI run history,
 *      and the UI Retry button re-runs the run. (The 8×-retry →
 *      DEAD_LETTER path is covered hermetically in the integration spec —
 *      no real network target in this environment produces a retryable
 *      transport error, so a full E2E DLQ cycle cannot be hit
 *      deterministically; recorded in the Phase 11 gap report.)
 *   6. self-trigger loop prevention through the REAL stream + consumer +
 *      worker: an UPDATE_RECORD emits record.updated, which re-matches the
 *      same automation → the second run is skipped with
 *      output.skipped = "self_trigger" and no third run follows.
 *   7. manual run (DRAFT automation, POST /automations/:id/run → 202)
 *      executes and its full per-action logs render in the UI.
 *
 * Data-testid contract (verified against apps/web, TASK-0114 UI):
 *   automation-list, automation-row-{id}, automation-toggle-{id}
 *   (role=switch), automation-runs-{id}, automation-run-{runId}
 *   (status text lowercase, "dead-letter" for DEAD_LETTER),
 *   automation-run-log-{runId}, automation-retry-run-{runId}.
 *
 * API note: databases are created through the page path
 * (POST /workspaces/:id/pages { type: "database" } → { id, databaseId }) —
 * the workspace-scoped database-create route does not exist in the API
 * (pre-existing gap, also breaks the pre-existing databases integration
 * suite; recorded in the Phase 11 gap report).
 */
import { spawn } from "node:child_process";
import { randomBytes } from "node:crypto";
import { writeFileSync, rmSync } from "node:fs";
import path from "node:path";
import type { APIRequestContext, Page } from "@playwright/test";
import { test, expect } from "../fixtures";

// ---------------------------------------------------------------------------
// Helpers (spec-local: tolerant of envelope shapes, exact on what is asserted)
// ---------------------------------------------------------------------------

type Json = Record<string, unknown>;
interface Run {
  id: string;
  status: string;
  triggerType: string | null;
  retryCount: number;
  error: { code?: string; message?: string } | null;
  createdAt: string;
  startedAt: string | null;
  finishedAt: string | null;
  output?: unknown;
}

const j = async (res: { json(): Promise<unknown> }): Promise<Json> =>
  (await res.json().catch(() => ({}))) as Json;

async function expectOk(res: { status(): number; text(): Promise<string> }, what: string): Promise<Json> {
  const status = res.status();
  if (status < 200 || status >= 300) throw new Error(`${what} → ${status}: ${await res.text()}`);
  return j(res);
}

async function createDatabasePage(ctx: APIRequestContext, wsId: string, title: string): Promise<{ pageId: string; databaseId: string }> {
  // The REAL database-creation surface: a page with type "database"
  // (page.service.ts creates the Database row + seeds the Title property +
  // embeds the database block). Returns the BARE page with databaseId.
  const res = await ctx.post(`/api/v1/workspaces/${wsId}/pages`, { data: { title, type: "database" } });
  const body = await expectOk(res, `create database page "${title}"`);
  const databaseId = body.databaseId;
  if (typeof databaseId !== "string" || !databaseId) throw new Error(`createDatabasePage: no databaseId in response: ${JSON.stringify(body)}`);
  return { pageId: String(body.id), databaseId };
}

async function addPropertyRaw(ctx: APIRequestContext, dbId: string, body: { name: string; type: string; config?: unknown }): Promise<void> {
  await expectOk(
    await ctx.post(`/api/v1/databases/${dbId}/properties`, { data: body }),
    `add property ${body.name}`
  );
}

/** Create a property then look up its id by name (the list envelope is tolerated). */
async function findPropertyId(ctx: APIRequestContext, dbId: string, name: string): Promise<string> {
  const res = await ctx.get(`/api/v1/databases/${dbId}/properties`);
  const body = await expectOk(res, `list properties of ${dbId}`);
  const list = (Array.isArray(body) ? body : (body.data ?? body.properties ?? [])) as Json[];
  const p = list.find((x) => x.name === name);
  if (!p?.id) throw new Error(`property "${name}" not found in ${dbId}: ${JSON.stringify(list.map((x) => x.name))}`);
  return String(p.id);
}

async function createRecordRaw(ctx: APIRequestContext, dbId: string, properties: Record<string, unknown>): Promise<Json> {
  const res = await ctx.post(`/api/v1/databases/${dbId}/records`, { data: { properties } });
  const body = await expectOk(res, `create record in ${dbId}`);
  const rec = (body.record ?? body) as Json;
  if (!rec.id) throw new Error(`createRecordRaw: no record id: ${JSON.stringify(body)}`);
  return rec;
}

async function getRecord(ctx: APIRequestContext, recordId: string): Promise<Json> {
  const res = await ctx.get(`/api/v1/records/${recordId}`);
  return expectOk(res, `get record ${recordId}`);
}

/** The SELECT value stored on a record property: the optionId (envelope-tolerant). */
function selectValue(record: Json, propertyId: string): string | null {
  const props = record.properties as Record<string, unknown> | undefined;
  const v = props?.[propertyId];
  if (v == null) return null;
  if (typeof v === "object" && "optionId" in (v as Json)) return String((v as Json).optionId);
  return String(v);
}

/**
 * On narrow (mobile) viewports the app shell keeps a fixed 256px sidebar and
 * the automations table (min-width ~460px) overflows UNDER it, so the
 * run-history rows render partially beneath the sidebar and their click point
 * is intercepted by the aside. The intended mobile operation is the topbar
 * "Toggle sidebar" button (collapses the aside to nothing, freeing the full
 * width) — perform it on narrow viewports so the run rows are clickable.
 * No-op on desktop (the sidebar stays expanded).
 *
 * NOTE: the app shell not auto-collapsing the sidebar on narrow viewports (and
 * letting wide tables overflow under it) is a PRE-EXISTING app-shell
 * mobile-layout limitation, not an automations bug — recorded in the Phase 11
 * gap report. This helper exercises the intended collapse rather than masking
 * the layout with a JS click.
 */
async function collapseSidebarIfNarrow(page: Page): Promise<void> {
  const vp = page.viewportSize();
  if (!vp || vp.width >= 768) return; // desktop — leave the sidebar as-is
  if ((await page.locator("aside").count()) === 0) return; // already collapsed
  await page.getByRole("button", { name: "Toggle sidebar" }).click();
  await page.waitForTimeout(300);
}

// --- automation API ---------------------------------------------------------

interface AutomationDetail {
  id: string;
  name: string;
  status: string;
  databaseId: string | null;
  triggers: string[];
  hasCondition: boolean;
  actions: string[];
}

async function createAutomation(ctx: APIRequestContext, wsId: string, input: Json): Promise<AutomationDetail> {
  const res = await ctx.post(`/api/v1/workspaces/${wsId}/automations`, { data: input });
  const body = await expectOk(res, `create automation "${input.name}"`);
  if (!body.id) throw new Error(`createAutomation: no id: ${JSON.stringify(body)}`);
  return body as unknown as AutomationDetail;
}

async function setTriggers(ctx: APIRequestContext, autoId: string, triggers: Json[]): Promise<void> {
  await expectOk(await ctx.put(`/api/v1/automations/${autoId}/triggers`, { data: { triggers } }), `set triggers on ${autoId}`);
}

async function setCondition(ctx: APIRequestContext, autoId: string, condition: Json | null): Promise<void> {
  await expectOk(await ctx.put(`/api/v1/automations/${autoId}/conditions`, { data: { condition } }), `set condition on ${autoId}`);
}

async function setActions(ctx: APIRequestContext, autoId: string, actions: Json[]): Promise<void> {
  await expectOk(await ctx.put(`/api/v1/automations/${autoId}/actions`, { data: { actions } }), `set actions on ${autoId}`);
}

async function setStatus(ctx: APIRequestContext, autoId: string, status: string): Promise<void> {
  await expectOk(await ctx.post(`/api/v1/automations/${autoId}/status`, { data: { status } }), `set status ${status} on ${autoId}`);
}

async function listRuns(ctx: APIRequestContext, autoId: string): Promise<Run[]> {
  const res = await ctx.get(`/api/v1/automations/${autoId}/runs?limit=200`);
  const body = await expectOk(res, `list runs of ${autoId}`);
  return (body.data as Run[]) ?? [];
}

/**
 * listRuns that is tolerant of 429 RATE_LIMITED: the API rate-limits per
 * user/min (RATE_LIMIT_API, default 100), and this suite polls the runs
 * endpoint in a tight loop — a 429 must not fail the test, it must just
 * wait out the window (Retry-After) and re-poll. Returns null on a 429 so
 * the caller can skip this poll cycle.
 */
async function listRunsTolerant(ctx: APIRequestContext, autoId: string): Promise<Run[] | null> {
  const res = await ctx.get(`/api/v1/automations/${autoId}/runs?limit=200`);
  if (res.status() === 429) {
    const retryAfter = Number(res.headers()["retry-after"] ?? 60);
    const waitMs = Math.min(Math.max(retryAfter, 1) * 1000, 65_000);
    await sleep(waitMs);
    return null;
  }
  const body = await expectOk(res, `list runs of ${autoId}`);
  return (body.data as Run[]) ?? [];
}

async function getRun(ctx: APIRequestContext, autoId: string, runId: string): Promise<Json> {
  const res = await ctx.get(`/api/v1/automations/${autoId}/runs/${runId}`);
  return expectOk(res, `get run ${runId}`);
}

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

/** Poll listRuns until pred() matches; returns the matching run (or the last state on timeout). */
async function waitForRun(
  ctx: APIRequestContext,
  autoId: string,
  pred: (runs: Run[]) => Run | undefined,
  opts: { timeoutMs?: number; intervalMs?: number; what?: string } = {}
): Promise<Run> {
  const { timeoutMs = 30_000, intervalMs = 1_000 } = opts;
  const t0 = Date.now();
  for (;;) {
    const runs = await listRunsTolerant(ctx, autoId);
    if (runs === null) continue; // 429 — waited out Retry-After, re-poll
    const m = pred(runs);
    if (m) return m;
    if (Date.now() - t0 > timeoutMs) {
      throw new Error(
        `${opts.what ?? "waitForRun"}: timeout after ${timeoutMs} ms; runs = ${JSON.stringify(runs.map((r) => ({ id: r.id.slice(-6), status: r.status })))}`
      );
    }
    await sleep(intervalMs);
  }
}

// --- schedule reconciler (worker source, via a child tsx process) ------------
//
// The worker registers SCHEDULE repeatables at boot + every 5 minutes. The
// test invokes the REAL reconciler so a cron `* * * * *` automation is
// registered immediately; the cron resolution + fire still happen inside the
// real BullMQ queue (no time manipulation).
//
// The reconciler lives in the worker package (TS source, no build for the
// monorepo test path). Playwright's strip-only TS transform cannot load it
// into the spec process (parameter properties etc.), so — exactly like the
// integration spec's SIGKILL/crash-recovery test — we run it in a CHILD tsx
// process whose script is written next to the spec (so @prisma/client,
// ioredis and bullmq resolve from the e2e package's node_modules). The child
// imports the worker source via file:// URL, runs ONE reconcile, prints
// `{added, removed}` and exits.
async function reconcileSchedules(): Promise<{ added: number; removed: number }> {
  const dsn = process.env.DATABASE_URL;
  const redisUrl = process.env.REDIS_URL ?? "redis://localhost:6379";
  if (!dsn) throw new Error("reconcileSchedules: DATABASE_URL is not set (the schedule E2E test needs the DB DSN).");
  // CJS spec (no "type":"module") — use __dirname (matches files.spec.ts).
  const here = __dirname;
  const workerSrc = path.resolve(here, "..", "..", "..", "apps", "worker", "src");
  const childScript = path.join(here, `.reconcile-child-${randomBytes(4).toString("hex")}.mts`);
  // tsx is a repo-root devDependency (root node_modules), 3 levels up from here.
  const tsxCli = path.resolve(here, "..", "..", "..", "node_modules", "tsx", "dist", "cli.mjs");
  writeFileSync(
    childScript,
    [
      "import { PrismaClient } from \"@prisma/client\";",
      "import { Redis } from \"ioredis\";",
      "const dir = process.env.WORKER_SRC_DIR!;",
      "const { reconcileScheduleAutomations } = await import(\"file://\" + dir + \"/processors/automations/schedule.ts\");",
      "const prisma = new PrismaClient({ datasources: { db: { url: process.env.DATABASE_URL! } } });",
      "const redis = new Redis(process.env.REDIS_URL!, { maxRetriesPerRequest: null });",
      "const r = await reconcileScheduleAutomations(redis, prisma);",
      // Marker-prefixed line: the worker source pulls in the pino-pretty
      // logger, whose ASYNC flush can interleave indented field lines AFTER
      // a synchronous console.log, so the last stdout line is unreliable.
      // The parent scans for the marker prefix instead.
      "console.log(\"RECONCILE_DONE \" + JSON.stringify(r));",
      "await redis.quit();",
      "await prisma.$disconnect();",
      ""
    ].join("\n"),
    "utf8"
  );
  try {
    const out: string[] = [];
    const err: string[] = [];
    await new Promise<void>((resolve, reject) => {
      const child = spawn(process.execPath, [tsxCli, childScript], {
        cwd: here,
        stdio: ["ignore", "pipe", "pipe"],
        env: { ...process.env, DATABASE_URL: dsn, REDIS_URL: redisUrl, WORKER_SRC_DIR: workerSrc }
      });
      const t = setTimeout(() => {
        child.kill("SIGKILL");
        reject(new Error(`reconcileSchedules: child timed out; stderr=${err.join("")}`));
      }, 30_000);
      child.stdout.on("data", (d) => out.push(String(d)));
      child.stderr.on("data", (d) => err.push(String(d)));
      child.on("close", (code) => {
        clearTimeout(t);
        if (code === 0) resolve();
        else reject(new Error(`reconcileSchedules: child exited ${code}; stderr=${err.join("")}`));
      });
    });
    const stdout = out.join("");
    const line = stdout.split("\n").find((l) => l.startsWith("RECONCILE_DONE "));
    if (!line) {
      throw new Error(`reconcileSchedules: no RECONCILE_DONE marker in child stdout; stdout=${JSON.stringify(stdout.slice(-500))}; stderr=${err.join("").slice(-500)}`);
    }
    return JSON.parse(line.slice("RECONCILE_DONE ".length)) as { added: number; removed: number };
  } finally {
    rmSync(childScript, { force: true });
  }
}

/** GET /notifications (per-user) — find one matching body. */
async function findNotification(ctx: APIRequestContext, bodyIncludes: string): Promise<Json | null> {
  const res = await ctx.get(`/api/v1/notifications?limit=100`);
  const body = await expectOk(res, "list notifications");
  const list = (Array.isArray(body) ? body : (body.data ?? [])) as Json[];
  return list.find((n) => typeof n.body === "string" && n.body.includes(bodyIncludes)) ?? null;
}

// ---------------------------------------------------------------------------
// 1. record.created → UPDATE_RECORD + CREATE_NOTIFICATION → SUCCEEDED < 10 s
// ---------------------------------------------------------------------------

test("record.created triggers UPDATE_RECORD + notification; property changed in < 10 s; run + logs visible in UI", async ({
  api,
  authPage,
  freshWorkspace
}) => {
  const ws = await freshWorkspace("Auto E2E A");
  const { databaseId } = await createDatabasePage(api, ws.id, "Tickets");
  await addPropertyRaw(api, databaseId, {
    name: "Stage",
    type: "SELECT",
    config: { options: [{ name: "A" }, { name: "B" }] }
  });
  const statusPropId = await findPropertyId(api, databaseId, "Stage");

  const auto = await createAutomation(api, ws.id, {
    name: "E2E: promote new tickets",
    databaseId,
    triggers: [{ type: "RECORD_CREATED", config: { databaseId } }],
    actions: [
      { type: "UPDATE_RECORD", config: { recordId: "from_trigger", propertyValues: { Stage: "B" } } },
      { type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "DATABASE_CHANGE", body: "E2E auto: ticket promoted to B" } }
    ]
  });
  await setStatus(api, auto.id, "ACTIVE");

  // The record that fires the automation (created with A; the automation sets B).
  const createdMs = Date.now();
  const rec = await createRecordRaw(api, databaseId, { [statusPropId]: "A" });
  const recordId = String(rec.id);

  const run = await waitForRun(
    api,
    auto.id,
    (runs) => runs.find((r) => r.status === "SUCCEEDED" || r.status === "FAILED" || r.status === "DEAD_LETTER"),
    { timeoutMs: 10_000, what: "terminal run within the 10 s exit criterion" }
  );
  expect(run.status, `run ${run.id} terminal status (error=${JSON.stringify(run.error)})`).toBe("SUCCEEDED");
  const elapsedMs = Date.now() - createdMs;
  expect(elapsedMs, `record-created → SUCCEEDED must be < 10 s (was ${elapsedMs} ms)`).toBeLessThan(10_000);

  // The effect: Status flipped to B on the same record.
  const after = await getRecord(api, recordId);
  expect(selectValue(after, statusPropId)).toBe("B");

  // The notification reached the automation's creator (the demo user owns the workspace).
  expect(await findNotification(api, "E2E auto: ticket promoted to B")).not.toBeNull();

  // --- UI: list → row → run history → full logs -----------------------------
  // Navigate by WORKSPACE **id** (not slug): the API's GET /workspaces/:id
  // resolves by id only (findUnique on id) and 403s on a slug, which would
  // bounce the [workspace] layout back to /w/default (matching the pre-existing
  // spec convention — files/databases.spec navigate with ws.id).
  await authPage.goto(`/w/${ws.id}/settings/automations`);
  await expect(authPage.getByTestId("automation-list")).toBeVisible({ timeout: 15_000 });
  await expect(authPage.getByTestId(`automation-row-${auto.id}`)).toBeVisible({ timeout: 15_000 });
  // Latest-run cell on the row shows "succeeded".
  await expect(authPage.getByTestId(`automation-row-${auto.id}`).getByText("succeeded")).toBeVisible();

  await collapseSidebarIfNarrow(authPage);
  await authPage.getByTestId(`automation-runs-${auto.id}`).click();
  await expect(authPage.getByTestId(`automation-run-${run.id}`)).toBeVisible({ timeout: 15_000 });
  await authPage.getByTestId(`automation-run-${run.id}`).click();
  const logPanel = authPage.getByTestId(`automation-run-log-${run.id}`);
  await expect(logPanel).toBeVisible({ timeout: 15_000 });
  // Full run logs: per-action lines + the terminal line.
  await expect(logPanel).toContainText("action 1/2 UPDATE_RECORD started");
  await expect(logPanel).toContainText("action 2/2 CREATE_NOTIFICATION started");
  await expect(logPanel).toContainText("run completed");
});

// ---------------------------------------------------------------------------
// 2. condition gate (option NAME semantics)
// ---------------------------------------------------------------------------

test("condition gate: non-matching record skipped, matching record executes", async ({ api, freshWorkspace }) => {
  const ws = await freshWorkspace("Auto E2E B");
  const { databaseId } = await createDatabasePage(api, ws.id, "Leads");
  await addPropertyRaw(api, databaseId, {
    name: "Stage",
    type: "SELECT",
    config: { options: [{ name: "A" }, { name: "B" }] }
  });
  const statusPropId = await findPropertyId(api, databaseId, "Stage");

  const auto = await createAutomation(api, ws.id, {
    name: "E2E: gate on Stage = B",
    databaseId,
    triggers: [{ type: "RECORD_CREATED", config: { databaseId } }],
    condition: { field: { ref: "property", name: "Stage" }, operator: "eq", value: "B" },
    actions: [{ type: "UPDATE_RECORD", config: { recordId: "from_trigger", propertyValues: { Stage: "A" } } }]
  });
  await setStatus(api, auto.id, "ACTIVE");

  // Non-matching (A) → a run is still created (the matcher matched the
  // trigger) but SUCCEEDS with output.skipped = "conditions_not_met".
  await createRecordRaw(api, databaseId, { [statusPropId]: "A" });
  const skipped = await waitForRun(api, auto.id, (runs) => runs.find((r) => r.status === "SUCCEEDED"), {
    timeoutMs: 15_000,
    what: "skipped run"
  });
  const skippedDetail = await getRun(api, auto.id, skipped.id);
  expect(skippedDetail.output, "output.skipped for the gated run").toEqual({ skipped: "conditions_not_met" });

  // Matching (B) → the action runs.
  await createRecordRaw(api, databaseId, { [statusPropId]: "B" });
  const ran = await waitForRun(api, auto.id, (runs) => runs.find((r) => r.id !== skipped.id && r.status === "SUCCEEDED"), {
    timeoutMs: 15_000,
    what: "executed run"
  });
  const ranDetail = await getRun(api, auto.id, ran.id);
  expect((ranDetail.output as Json)?.skipped, "the matching run must NOT be skipped").toBeUndefined();
  const actions = (ranDetail.output as Json)?.actions as Json[];
  expect(actions?.length).toBe(1);
  expect(actions?.[0]?.type).toBe("UPDATE_RECORD");
});

// ---------------------------------------------------------------------------
// 3. schedule (cron) trigger fires
// ---------------------------------------------------------------------------

test("schedule trigger: cron automation fires and creates a schedule run", async ({ api, freshWorkspace }) => {
  test.setTimeout(120_000);
  const ws = await freshWorkspace("Auto E2E C");
  const auto = await createAutomation(api, ws.id, {
    name: "E2E: hourly ping",
    triggers: [{ type: "SCHEDULE", config: { cron: "* * * * *" } }],
    actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "E2E cron tick" } }]
  });
  await setStatus(api, auto.id, "ACTIVE");

  // Register the repeatable via the real worker reconciler (normally boot /
  // every 5 min — not worth waiting in a test).
  const { added } = await reconcileSchedules();
  expect(added, "the reconciler must register the new repeatable").toBeGreaterThanOrEqual(1);

  const run = await waitForRun(
    api,
    auto.id,
    (runs) => runs.find((r) => r.triggerType === "schedule" && (r.status === "SUCCEEDED" || r.status === "FAILED" || r.status === "DEAD_LETTER")),
    { timeoutMs: 95_000, intervalMs: 2_000, what: "cron fire (≤ 60 s resolution + execution)" }
  );
  expect(run.status).toBe("SUCCEEDED");
  expect(await findNotification(api, "E2E cron tick")).not.toBeNull();

  // Cleanup: pause + reconcile again so the repeatable is removed (the
  // workspace teardown would otherwise leave a harmless orphan that skips
  // on the missing automation).
  await setStatus(api, auto.id, "PAUSED");
  await reconcileSchedules();
});

// ---------------------------------------------------------------------------
// 4. pause stops execution; re-activate resumes
// ---------------------------------------------------------------------------

test("PAUSED automation stops matching; re-activating resumes", async ({ api, freshWorkspace }) => {
  const ws = await freshWorkspace("Auto E2E D");
  const { databaseId } = await createDatabasePage(api, ws.id, "Items");
  const auto = await createAutomation(api, ws.id, {
    name: "E2E: pausable",
    databaseId,
    triggers: [{ type: "RECORD_CREATED", config: { databaseId } }],
    actions: [{ type: "CREATE_NOTIFICATION", config: { to: "automation_creator", type: "AUTOMATION", body: "E2E pause test tick" } }]
  });
  await setStatus(api, auto.id, "ACTIVE");

  // Active: the first record matches.
  await createRecordRaw(api, databaseId, {});
  const first = await waitForRun(api, auto.id, (runs) => runs.find((r) => r.status === "SUCCEEDED"), {
    timeoutMs: 15_000,
    what: "first (active) run"
  });

  // Paused: a second record must NOT create a new run (10 s of consumer time).
  await setStatus(api, auto.id, "PAUSED");
  await sleep(1_000); // let the status settle before the event is published
  await createRecordRaw(api, databaseId, {});
  await sleep(10_000);
  const whilePaused = await listRuns(api, auto.id);
  expect(whilePaused, "no new run while PAUSED").toHaveLength(1);
  expect(whilePaused[0]?.id).toBe(first.id);

  // Re-activated: a third record matches again.
  await setStatus(api, auto.id, "ACTIVE");
  await createRecordRaw(api, databaseId, {});
  await waitForRun(api, auto.id, (runs) => runs.find((r) => r.id !== first.id && r.status === "SUCCEEDED"), {
    timeoutMs: 15_000,
    what: "run after re-activation"
  });
});

// ---------------------------------------------------------------------------
// 5. failure → FAILED + UI error + retry re-runs
// ---------------------------------------------------------------------------

test("failing action (dead TEST-NET webhook) → FAILED run with error in UI; retry re-runs the run", async ({
  api,
  authPage,
  freshWorkspace
}) => {
  test.setTimeout(90_000);
  const ws = await freshWorkspace("Auto E2E E");
  const { databaseId } = await createDatabasePage(api, ws.id, "Outbox");
  const auto = await createAutomation(api, ws.id, {
    name: "E2E: failing webhook",
    databaseId,
    triggers: [{ type: "RECORD_CREATED", config: { databaseId } }],
    actions: [{ type: "SEND_WEBHOOK", config: { url: "https://192.0.2.1/hook" } }]
  });
  await setStatus(api, auto.id, "ACTIVE");

  // 192.0.2.1 is TEST-NET (public-range, passes the SSRF guard) but
  // unreachable: undici gives up after its 10 s connect timeout and the
  // pipeline classifies the transport error as fatal → FAILED, no retries.
  await createRecordRaw(api, databaseId, {});
  const run = await waitForRun(
    api,
    auto.id,
    (runs) => runs.find((r) => r.status === "FAILED" || r.status === "SUCCEEDED" || r.status === "DEAD_LETTER"),
    { timeoutMs: 45_000, what: "terminal (failed) run" }
  );
  expect(run.status, `terminal status (error=${JSON.stringify(run.error)})`).toBe("FAILED");
  expect(run.error, "the run error must be stored").not.toBeNull();

  // --- UI: the failed run + its error + the Retry button --------------------
  // Navigate by WORKSPACE **id** (not slug): the API's GET /workspaces/:id
  // resolves by id only (findUnique on id) and 403s on a slug, which would
  // bounce the [workspace] layout back to /w/default (matching the pre-existing
  // spec convention — files/databases.spec navigate with ws.id).
  await authPage.goto(`/w/${ws.id}/settings/automations`);
  await expect(authPage.getByTestId(`automation-row-${auto.id}`)).toBeVisible({ timeout: 15_000 });
  await collapseSidebarIfNarrow(authPage);
  await authPage.getByTestId(`automation-runs-${auto.id}`).click();
  await expect(authPage.getByTestId(`automation-run-${run.id}`)).toBeVisible({ timeout: 15_000 });
  await expect(authPage.getByTestId(`automation-run-${run.id}`).getByText("failed")).toBeVisible();
  await authPage.getByTestId(`automation-run-${run.id}`).click();
  const logPanel = authPage.getByTestId(`automation-run-log-${run.id}`);
  await expect(logPanel).toBeVisible({ timeout: 15_000 });
  await expect(logPanel).toContainText("failed");
  const retryButton = authPage.getByTestId(`automation-retry-run-${run.id}`);
  await expect(retryButton).toBeVisible();

  // Retry re-queues the SAME run row (retryCount reset) → it runs again.
  const before = (await getRun(api, auto.id, run.id)) as Json;
  const firstFinishedAt = before.finishedAt as string;
  await retryButton.click();
  // The run goes back to PENDING/RUNNING, then fails again (the target is
  // still dead) — with a FRESH startedAt (proof it actually re-ran).
  await waitForRun(
    api,
    auto.id,
    (runs) => runs.find((r) => r.id === run.id && r.status === "FAILED" && (r.finishedAt ?? "") > firstFinishedAt),
    { timeoutMs: 60_000, what: "re-run after UI retry" }
  );
});

// ---------------------------------------------------------------------------
// 6. self-trigger loop prevention through the real stream + worker
// ---------------------------------------------------------------------------

test("self-trigger: an automation's own record.updated re-match is skipped, no loop", async ({ api, freshWorkspace }) => {
  test.setTimeout(90_000);
  const ws = await freshWorkspace("Auto E2E F");
  const { databaseId } = await createDatabasePage(api, ws.id, "Loop");
  await addPropertyRaw(api, databaseId, {
    name: "Stage",
    type: "SELECT",
    config: { options: [{ name: "A" }, { name: "B" }] }
  });
  const statusPropId = await findPropertyId(api, databaseId, "Stage");

  const auto = await createAutomation(api, ws.id, {
    name: "E2E: self-trigger guard",
    databaseId,
    // NOTE: a BARE RECORD_UPDATED trigger (no `properties` filter). The
    // matcher only narrows on `changedProperties` when the trigger carries a
    // non-empty `properties` list; user-originated record.updated events carry
    // the property **ids** (query.service.ts `changedFields` → Object.keys of
    // the id-keyed patch), whereas the trigger filter / condition / executor
    // contract is **property names** — so a property-scoped trigger would fail
    // closed on a user edit. Without the filter, both the user edit and the
    // automation's own echo match, and the loop guard skips the echo
    // (self_trigger). The id/name mismatch for property-scoped user triggers
    // is a pre-existing contract gap (recorded in the Phase 11 gap report).
    triggers: [{ type: "RECORD_UPDATED", config: { databaseId } }],
    actions: [{ type: "UPDATE_RECORD", config: { recordId: "from_trigger", propertyValues: { Stage: "A" } } }]
  });
  await setStatus(api, auto.id, "ACTIVE");

  // Run 1 (depth 0): a USER edit publishes record.updated — the matcher
  // enqueues the run. The action flips Status back to A, which emits ANOTHER
  // record.updated — stamped with THIS automationId at depth 1. The consumer
  // re-matches the same automation (run 2), and the loop guard must skip it
  // with "self_trigger". No run 3 may follow (a skipped run executes no
  // actions → emits nothing).
  const rec = await createRecordRaw(api, databaseId, { [statusPropId]: "A" });
  await expectOk(
    await api.patch(`/api/v1/records/${String(rec.id)}`, { data: { properties: { [statusPropId]: "B" } } }),
    "user edit (B) that starts the chain"
  );
  // Wait until BOTH runs exist AND the newest (the echo) is terminal — a
  // two-run count alone would let us read the echo's output before it ran.
  const isTerminal = (r: Run) =>
    r.status === "SUCCEEDED" || r.status === "FAILED" || r.status === "DEAD_LETTER";
  await waitForRun(api, auto.id, (runs) => (runs.length >= 2 && isTerminal(runs[0]) ? runs[0] : undefined), {
    timeoutMs: 30_000,
    what: "the second run (the re-matched record.updated echo) reaching terminal"
  });
  const runsNow = await listRuns(api, auto.id);
  expect(runsNow.length, "exactly 2 runs: the update ran, its own echo was skipped").toBe(2);
  // The echo run is the newest one (createdAt desc); it must be the skip.
  const skipped = await getRun(api, auto.id, runsNow[0].id);
  expect(skipped.output, "self_trigger skip output").toEqual({ skipped: "self_trigger" });

  // Let the consumer prove there is no third run.
  await sleep(8_000);
  expect((await listRuns(api, auto.id)).length, "no third run (loop prevented)").toBe(2);
});

// ---------------------------------------------------------------------------
// 7. manual run (DRAFT) + full logs in the UI
// ---------------------------------------------------------------------------

test("manual run on a DRAFT automation executes; full per-action logs render in the UI", async ({
  api,
  authPage,
  freshWorkspace
}) => {
  const ws = await freshWorkspace("Auto E2E G");
  const { databaseId } = await createDatabasePage(api, ws.id, "Manual");
  await addPropertyRaw(api, databaseId, {
    name: "Stage",
    type: "SELECT",
    config: { options: [{ name: "A" }, { name: "B" }] }
  });
  const statusPropId = await findPropertyId(api, databaseId, "Stage");
  const seed = await createRecordRaw(api, databaseId, { [statusPropId]: "A" });

  // DRAFT (created, never activated) — manual runs are allowed on drafts.
  const auto = await createAutomation(api, ws.id, {
    name: "E2E: manual draft",
    databaseId,
    actions: [{ type: "UPDATE_RECORD", config: { recordId: String(seed.id), propertyValues: { Stage: "B" } } }]
  });
  expect(auto.status).toBe("DRAFT");

  const res = await api.post(`/api/v1/automations/${auto.id}/run`, { data: {} });
  expect(res.status(), "manual run → 202").toBe(202);
  const body = await j(res);
  const runId = String(body.runId);
  expect(runId).toBeTruthy();

  const run = await waitForRun(
    api,
    auto.id,
    (runs) => runs.find((r) => r.id === runId && (r.status === "SUCCEEDED" || r.status === "FAILED")),
    { timeoutMs: 30_000, what: "manual run terminal" }
  );
  expect(run.status).toBe("SUCCEEDED");
  expect(selectValue(await getRecord(api, String(seed.id)), statusPropId)).toBe("B");

  // UI: the DRAFT row exists, its run history shows the manual run + logs.
  // Navigate by WORKSPACE **id** (not slug): the API's GET /workspaces/:id
  // resolves by id only (findUnique on id) and 403s on a slug, which would
  // bounce the [workspace] layout back to /w/default (matching the pre-existing
  // spec convention — files/databases.spec navigate with ws.id).
  await authPage.goto(`/w/${ws.id}/settings/automations`);
  await expect(authPage.getByTestId(`automation-row-${auto.id}`)).toBeVisible({ timeout: 15_000 });
  await collapseSidebarIfNarrow(authPage);
  await authPage.getByTestId(`automation-runs-${auto.id}`).click();
  await expect(authPage.getByTestId(`automation-run-${runId}`)).toBeVisible({ timeout: 15_000 });
  await authPage.getByTestId(`automation-run-${runId}`).click();
  const logPanel = authPage.getByTestId(`automation-run-log-${runId}`);
  await expect(logPanel).toBeVisible({ timeout: 15_000 });
  await expect(logPanel).toContainText("action 1/1 UPDATE_RECORD started");
  await expect(logPanel).toContainText("action 1/1 UPDATE_RECORD done");
  await expect(logPanel).toContainText("run completed");
});