import { createHash } from "node:crypto";
import { afterEach, beforeAll, afterAll, beforeEach, describe, it, expect, vi } from "vitest";
import { GenericContainer, Wait, type StartedTestContainer } from "testcontainers";
import { Client } from "pg";
import { createRequire } from "node:module";
import { fileURLToPath } from "node:url";
import { execFileSync } from "node:child_process";
import { z } from "zod";
import type { BatchResult, ReverseEtlContext, JsonObject, WriteBatch } from "@jitsu/protocols/reverse-etl";
import { ReverseRunConfig } from "@jitsu/warehouse-query/src/runtime";
import { Database, openPersistence } from "./persistence";
import { execute, type ExecuteOptions } from "./execute";
import type { RuntimeAdapter } from "./adapters";
import { createAdapterRegistry } from "./adapters";
import { Tasks } from "./tasks";
import { googleAudienceStateStream } from "@jitsu/destination-functions/src/functions/google-ads/audience/state";
import { ReverseEtlManualReconciliationError } from "@jitsu/destination-functions/src/reverse-etl/failure";
import { recordKey } from "@jitsu/destination-functions/src/reverse-etl/identity";
import * as http from "node:http";
import { createHmac } from "node:crypto";
import type { AddressInfo } from "node:net";
import { createWarehouseReader } from "@jitsu/warehouse-query";
import { createGuardedRequest } from "@jitsu/destination-functions/src/functions/lib/guarded-request";
import { defaultDeliveryDeps } from "@jitsu/destination-functions/src/functions/webhook/deliver";
import { createWebhookRuntime } from "@jitsu/destination-functions/src/functions/webhook/runtime";

import { MemoryObjects, persisted } from "./artifacts/test-support";
import { gunzipSync } from "node:zlib";
import { Artifacts } from "./artifacts/store";
const objects = new MemoryObjects();
const storage = { objectStorage: { store: objects, signal: new AbortController().signal } };
const durable = () => persisted(admin, objects);
afterEach(() => vi.useRealTimers());

describe("Meta Reverse ETL runner integration", () => {
  const metaResponse = (body: unknown) =>
    new Response(JSON.stringify(body)) as Awaited<ReturnType<typeof globalThis.fetch>>;
  it("logs a missing website user-agent mapping without leaking row data or submitting events", async () => {
    const f = fixture();
    f.input.config.destination = { destinationType: "facebook-conversions", accessToken: "private-meta-token" };
    f.input.config.model.cursor = undefined;
    f.input.config.options = {
      ...f.input.config.options,
      stream: "conversions",
      mode: "upsert",
      mapping: { email: "id", eventSourceUrl: "url" },
      streamOptions: { pixelId: "789", actionSource: "website", eventName: "Lead" },
    };
    f.input.adapters = createAdapterRegistry(vi.fn());
    const rows = [{ id: "private@example.com", url: "https://example.com/private-path" }];
    f.setRows(rows);
    const wire = vi.spyOn(globalThis, "fetch").mockRejectedValue(new Error("Unexpected provider request"));
    try {
      expect(await execute(f.input)).toBe("FAILED");
      expect(wire).not.toHaveBeenCalled();
      const errors = (await admin.query("SELECT message FROM newjitsu.task_log WHERE level='ERROR'")).rows;
      expect(errors).toHaveLength(1);
      expect(errors[0].message).toContain('Client user agent is required when Action source is "website"');
      expect(errors[0].message).toContain("Map a column to Client user agent");
      expect(errors[0].message).toContain("This row was not submitted");
      expect(errors[0].message).not.toContain("private");
      const task = (await admin.query("SELECT error FROM newjitsu.source_task WHERE task_id=$1", [f.input.taskId]))
        .rows[0];
      expect(task.error).toBe(errors[0].message);
    } finally {
      wire.mockRestore();
    }
  });
  it.each([false, true])(
    "provisions once and mirrors snapshots with additions before removals (lost receipt: %s)",
    async lostReceipt => {
      const f = fixture();
      f.input.config.destination = { destinationType: "facebook-conversions", accessToken: "meta-token" };
      f.input.config.model.cursor = undefined;
      f.input.config.options = {
        ...f.input.config.options,
        stream: "audience",
        mode: "mirror",
        mapping: { email: "id" },
        streamOptions: {
          accountId: "123",
          audience: { kind: "managed", name: "Test" },
          exclusiveManagementConfirmed: true,
        },
      };
      const token = vi.fn(async () => "unused-oauth");
      f.input.adapters = createAdapterRegistry(token);
      let marker = "";
      let loseResponse = lostReceipt;
      let lastSession: { session_id: string; num_received: number; num_invalid_entries: number } | undefined;
      const calls: { method: string; payload?: any }[] = [];
      const wire = vi.spyOn(globalThis, "fetch").mockImplementation(async (url, init) => {
        const method = init!.method!;
        const body = init?.body ? JSON.parse(String(init.body)) : undefined;
        if (String(url).includes("/customaudiences") && method === "POST") {
          marker = body.description;
          calls.push({ method: "create" });
          return metaResponse({ id: "456" });
        }
        if (String(url).includes("/456/users")) {
          calls.push({ method, payload: body.payload });
          lastSession = {
            session_id: String(body.session.session_id),
            num_received: body.payload.data.length,
            num_invalid_entries: 0,
          };
          if (loseResponse) {
            loseResponse = false;
            throw new Error("Response lost after ingestion");
          }
          return metaResponse({ audience_id: "456", ...lastSession });
        }
        if (String(url).includes("/456/sessions")) {
          expect(String(url)).toContain(`session_id=${lastSession!.session_id}`);
          return metaResponse({ data: [lastSession] });
        }
        return metaResponse({
          id: "456",
          account_id: "123",
          subtype: "CUSTOM",
          is_value_based: false,
          description: marker,
        });
      });
      try {
        f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }]);
        expect(await execute(f.input)).toBe(lostReceipt ? "FAILED" : "COMPLETE");
        if (lostReceipt) {
          f.input.taskId = "meta-resume";
          f.calls.length = 0;
          expect(await execute(f.input)).toBe("COMPLETE");
          expect(f.calls).not.toContain("reader");
        }
        f.input.taskId = "meta-second";
        f.setRows([{ id: "b@example.com" }, { id: "c@example.com" }]);
        expect(await execute(f.input)).toBe("COMPLETE");
        expect(calls.map(c => c.method)).toEqual(["create", "POST", "POST", "DELETE"]);
        expect(calls.slice(1).map(c => c.payload.data.length)).toEqual([2, 1, 1]);
        f.input.taskId = "meta-empty";
        f.setRows([]);
        expect(await execute(f.input)).toBe("COMPLETE");
        expect(calls.at(-1)).toMatchObject({ method: "DELETE" });
        expect(calls.at(-1)!.payload.data).toHaveLength(2);
        expect((await durable()).members).toHaveLength(0);
        expect(token).not.toHaveBeenCalled();
      } finally {
        wire.mockRestore();
      }
    }
  );
  it("deduplicates Meta conversion source keys across runs without audience state or OAuth", async () => {
    const f = fixture();
    f.input.config.destination = { destinationType: "facebook-conversions", accessToken: "meta-token" };
    f.input.config.model.cursor = undefined;
    f.input.config.options = {
      ...f.input.config.options,
      stream: "conversions",
      mode: "upsert",
      mapping: { email: "id" },
      streamOptions: { pixelId: "789", actionSource: "physical_store", eventName: "Lead" },
    };
    const token = vi.fn(async () => "unused-oauth");
    f.input.adapters = createAdapterRegistry(token);
    const uploads: any[][] = [];
    const wire = vi.spyOn(globalThis, "fetch").mockImplementation(async (url, init) => {
      expect(String(url)).toBe("https://graph.facebook.com/v26.0/789/events");
      const body = JSON.parse(String(init!.body));
      uploads.push(body.data);
      return metaResponse({ events_received: body.data.length });
    });
    try {
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }]);
      expect(await execute(f.input)).toBe("COMPLETE");
      f.input.taskId = "meta-conversions-next";
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }, { id: "c@example.com" }]);
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(uploads.map(batch => batch.length)).toEqual([2, 1]);
      expect(token).not.toHaveBeenCalled();
      expect(
        (await admin.query("SELECT count(*) FROM newjitsu.source_state WHERE stream='_REVERSE_ETL_META_AUDIENCE_' "))
          .rows[0].count
      ).toBe("0");
    } finally {
      wire.mockRestore();
    }
  });
});

describe("OpenAI Ads Reverse ETL runner integration", () => {
  const json = (body: unknown, status = 200) =>
    new Response(JSON.stringify(body), { status }) as Awaited<ReturnType<typeof globalThis.fetch>>;
  const openAiFixture = () => {
    const f = fixture();
    f.input.config.destination = { destinationType: "openai-ads", conversionsApiKey: "openai-key-secret" };
    f.input.config.model.cursor = undefined;
    f.input.config.options = {
      ...f.input.config.options,
      stream: "conversions",
      mode: "upsert",
      mapping: { email: "id" },
      streamOptions: { pixelId: "px_runner", actionSource: "offline", eventType: "lead_created" },
    };
    f.input.adapters = createAdapterRegistry(vi.fn());
    return f;
  };
  it("sends only new source keys to the pixel, with the key as a bearer token and no raw email on the wire", async () => {
    const f = openAiFixture();
    const sent: { url: string; auth: unknown; body: any }[] = [];
    const wire = vi.spyOn(globalThis, "fetch").mockImplementation(async (url, init) => {
      const body = JSON.parse(String(init!.body));
      sent.push({ url: String(url), auth: (init!.headers as Record<string, string>).Authorization, body });
      return json({ accepted_events: body.events.length });
    });
    try {
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }]);
      expect(await execute(f.input)).toBe("COMPLETE");
      f.input.taskId = "openai-conversions-next";
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }, { id: "c@example.com" }]);
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(sent.map(s => s.body.events.length)).toEqual([2, 1]);
      expect(sent.every(s => s.url === "https://bzr.openai.com/v1/events?pid=px_runner")).toBe(true);
      expect(sent.every(s => s.auth === "Bearer openai-key-secret")).toBe(true);
      expect(sent.every(s => s.body.integration_source === "jitsu")).toBe(true);
      const ids = sent.flatMap(s => s.body.events.map((e: any) => e.id));
      expect(new Set(ids).size).toBe(3);
      expect(JSON.stringify(sent)).not.toContain("@example.com");
      const logs = (await taskLogs("task")) + (await taskLogs("openai-conversions-next"));
      expect(logs).not.toContain("openai-key-secret");
      expect(logs).not.toContain("@example.com");
    } finally {
      wire.mockRestore();
    }
  });
  it("fails the run on a definite rejection with a controlled code and field path, not OpenAI's message", async () => {
    const f = openAiFixture();
    const wire = vi.spyOn(globalThis, "fetch").mockResolvedValue(
      json(
        {
          error: {
            message: "private detail a@example.com",
            type: "invalid_request_error",
            param: "events[0]",
            code: "invalid_event",
            errors: [{ message: "private detail", param: "events[0].type", code: "unknown_event_type" }],
          },
        },
        400
      )
    );
    try {
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }]);
      expect(await execute(f.input)).toBe("FAILED");
      expect(wire).toHaveBeenCalledTimes(1);
      const error = String((await task()).error);
      const logs = await taskLogs("task");
      for (const text of [error, logs]) {
        expect(text).not.toContain("private detail");
        expect(text).not.toContain("@example.com");
        expect(text).not.toContain("openai-key-secret");
      }
      expect(error).toContain("OPENAI_INVALID_EVENT");
      expect(logs).toContain("2 confirmed submitted in 1 batches; 0 accepted, 0 pending processing, 2 rejected");
    } finally {
      wire.mockRestore();
    }
  });
  it("stops on an unconfirmed delivery and never sends the batch again on the next run", async () => {
    const f = openAiFixture();
    const wire = vi
      .spyOn(globalThis, "fetch")
      .mockResolvedValue(json({ error: { message: "slow down", type: "rate_limit", code: null, param: null } }, 429));
    try {
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }]);
      expect(await execute(f.input)).toBe("FAILED");
      expect(wire).toHaveBeenCalledTimes(1);
      const error = String((await task()).error);
      expect(error).toContain("did not confirm a submitted batch");
      expect(error).toContain("do not reset sync state");
      expect(error).not.toContain("slow down");
      f.input.taskId = "openai-after-unconfirmed";
      await execute(f.input);
      expect(wire).toHaveBeenCalledTimes(1);
    } finally {
      wire.mockRestore();
    }
  });
});

describe("OpenAI Ads audience runner integration", () => {
  const reply = (body: unknown, status = 200) =>
    new Response(JSON.stringify(body), { status }) as Awaited<ReturnType<typeof globalThis.fetch>>;
  const audienceFixture = () => {
    const f = fixture();
    f.input.config.destination = { destinationType: "openai-ads", adsApiKey: "ads-key-secret" };
    f.input.config.model.cursor = undefined;
    f.input.config.options = {
      ...f.input.config.options,
      stream: "audience",
      mode: "mirror",
      mapping: { email: "id" },
      streamOptions: { audience: { kind: "managed", name: "Test audience" }, exclusiveManagementConfirmed: true },
    };
    f.input.adapters = createAdapterRegistry(vi.fn());
    return f;
  };
  /** OpenAI's audience API, answering at once: ready on create, operations succeeded on the first poll. */
  const openAi = (
    options: { loseAddResponses?: number; failOperations?: boolean; failFrom?: number; refuseSubmit?: number } = {}
  ) => {
    const audienceId = `caud_${"0".repeat(31)}1`;
    let description = "";
    let lose = options.loseAddResponses ?? 0;
    const operations = new Map<string, { id: string; action: string; identifiers: any[]; fails: boolean }>();
    const byKey = new Map<string, string>();
    const requests: { method: string; path: string; key?: string }[] = [];
    const fetch = vi.spyOn(globalThis, "fetch").mockImplementation(async (url, init) => {
      const path = String(url).replace("https://api.ads.openai.com/v1", "");
      const headers = init!.headers as Record<string, string>;
      requests.push({ method: init!.method!, path, key: headers["Idempotency-Key"] });
      expect(headers.Authorization).toBe("Bearer ads-key-secret");
      if (path === "/ad_account") return reply({ id: "adacct_runner" });
      if (path === "/custom_audiences" && init!.method === "POST") {
        description = JSON.parse(String(init!.body)).description;
        return reply({ id: audienceId, status: "ready", description });
      }
      if (path === `/custom_audiences/${audienceId}`) return reply({ id: audienceId, status: "ready", description });
      const submit = path.match(new RegExp(`^/custom_audiences/${audienceId}/(add|remove)$`));
      // Access revoked or the like: a definite refusal of the request itself, with OpenAI's error envelope.
      if (submit && options.refuseSubmit)
        return reply(
          { error: { message: "private detail", type: "invalid_request_error", code: "permission_denied" } },
          options.refuseSubmit
        );
      if (submit) {
        const key = headers["Idempotency-Key"];
        let id = byKey.get(key);
        if (!id) {
          id = `caudop_${operations.size + 1}`;
          byKey.set(key, id);
          // Whether an operation fails is fixed when it is created, as at OpenAI: a replay under the same key returns
          // the same failed operation, whatever the fake's options say afterwards.
          operations.set(id, {
            id,
            action: submit[1],
            identifiers: JSON.parse(String(init!.body)).identifiers,
            fails:
              !!options.failOperations || (options.failFrom !== undefined && operations.size + 1 >= options.failFrom),
          });
        }
        if (lose > 0) {
          lose--;
          throw new Error("Response lost after the request was applied");
        }
        return reply({ operation_id: id, custom_audience_id: audienceId, operation: submit[1], status: "processing" });
      }
      const poll = path.match(new RegExp(`^/custom_audiences/${audienceId}/operations/([^/]+)$`));
      if (poll)
        return reply({
          operation_id: poll[1],
          custom_audience_id: audienceId,
          operation: operations.get(poll[1])?.action,
          status: operations.get(poll[1])?.fails ? "failed" : "succeeded",
          ...(operations.get(poll[1])?.fails
            ? { failure_reason: { code: "audience_too_small", message: "private detail" } }
            : {}),
        });
      throw new Error(`Unexpected OpenAI request ${init!.method} ${path}`);
    });
    return { fetch, requests, operations, options };
  };
  const sha = (value: string) => createHash("sha256").update(value).digest("hex");
  const emails = (identifiers: any[]) => identifiers.map(i => i.identifier);

  it("creates the audience once and mirrors snapshots, adding new members before removing missing ones", async () => {
    const f = audienceFixture();
    const wire = openAi();
    try {
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }]);
      expect(await execute(f.input)).toBe("COMPLETE");
      f.input.taskId = "openai-audience-second";
      f.setRows([{ id: "b@example.com" }, { id: "c@example.com" }]);
      expect(await execute(f.input)).toBe("COMPLETE");
      const sent = [...wire.operations.values()];
      expect(sent.map(o => o.action)).toEqual(["add", "add", "remove"]);
      expect(emails(sent[0].identifiers).sort()).toEqual([sha("a@example.com"), sha("b@example.com")].sort());
      expect(emails(sent[1].identifiers)).toEqual([sha("c@example.com")]);
      expect(emails(sent[2].identifiers)).toEqual([sha("a@example.com")]);
      expect(wire.requests.filter(r => r.method === "POST" && r.path === "/custom_audiences")).toHaveLength(1);
      f.input.taskId = "openai-audience-empty";
      f.setRows([]);
      expect(await execute(f.input)).toBe("COMPLETE");
      expect([...wire.operations.values()].at(-1)).toMatchObject({ action: "remove" });
      expect(emails([...wire.operations.values()].at(-1)!.identifiers).sort()).toEqual(
        [sha("b@example.com"), sha("c@example.com")].sort()
      );
      expect((await durable()).members).toHaveLength(0);
      const logs = (await taskLogs("task")) + (await taskLogs("openai-audience-second"));
      expect(logs).not.toContain("ads-key-secret");
      expect(logs).not.toContain("@example.com");
      expect(JSON.stringify(wire.requests)).not.toContain("@example.com");
    } finally {
      wire.fetch.mockRestore();
    }
  });
  it("keeps a member two rows share until both rows are gone: one identifier type means one member", async () => {
    const f = audienceFixture();
    f.input.config.options = { ...f.input.config.options, mapping: { email: "email" } };
    const wire = openAi();
    try {
      f.setRows([
        { id: "1", email: "shared@example.com" },
        { id: "2", email: "shared@example.com" },
      ]);
      expect(await execute(f.input)).toBe("COMPLETE");
      const added = [...wire.operations.values()].flatMap(o => emails(o.identifiers));
      expect(added).toEqual([sha("shared@example.com")]);
      f.input.taskId = "openai-audience-shared-one-row-gone";
      f.setRows([{ id: "2", email: "shared@example.com" }]);
      expect(await execute(f.input)).toBe("COMPLETE");
      expect([...wire.operations.values()].map(o => o.action)).toEqual(["add"]);
      f.input.taskId = "openai-audience-shared-both-gone";
      f.setRows([]);
      expect(await execute(f.input)).toBe("COMPLETE");
      expect([...wire.operations.values()].map(o => o.action)).toEqual(["add", "remove"]);
    } finally {
      wire.fetch.mockRestore();
    }
  });
  it("refuses a mirror that maps two identifier types, before reading the source or calling OpenAI", async () => {
    const f = audienceFixture();
    f.input.config.options = { ...f.input.config.options, mapping: { email: "id", phone: "id" } };
    const wire = openAi();
    try {
      f.setRows([{ id: "a@example.com" }]);
      f.calls.length = 0;
      expect(await execute(f.input)).toBe("FAILED");
      expect(String((await task()).error)).toContain("one identifier type");
      expect(f.calls).not.toContain("reader");
      expect(wire.requests).toHaveLength(0);
    } finally {
      wire.fetch.mockRestore();
    }
  });
  it("recovers a lost submit response by replaying the same key, applying the batch once and not reading the source again", async () => {
    const f = audienceFixture();
    const wire = openAi({ loseAddResponses: 1 });
    try {
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }]);
      expect(await execute(f.input)).toBe("FAILED");
      const stopped = String((await task()).error);
      expect(stopped).toContain("Snapshot mirror stopped during delivery");
      expect(stopped).toContain("do not reset sync state");
      f.input.taskId = "openai-audience-resume";
      f.calls.length = 0;
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(f.calls).not.toContain("reader");
      const adds = wire.requests.filter(r => r.path.endsWith("/add"));
      expect(adds).toHaveLength(2);
      expect(adds[1].key).toBe(adds[0].key);
      expect(wire.operations.size).toBe(1);
    } finally {
      wire.fetch.mockRestore();
    }
  });
  /** OpenAI is healthy again for NEW operations; the same fake keeps its audience, its marker and its failed operations. */
  const heal = (wire: ReturnType<typeof openAi>) => {
    wire.options.failOperations = false;
    wire.options.failFrom = undefined;
  };
  const adds = (wire: ReturnType<typeof openAi>) =>
    [...wire.operations.values()].map(o => `${o.action}:${o.identifiers.length}`);

  it("fails the run when OpenAI reports an operation failed, with no provider text, key or address in the run", async () => {
    const f = audienceFixture();
    const wire = openAi({ failOperations: true });
    try {
      f.setRows([{ id: "a@example.com" }]);
      expect(await execute(f.input)).toBe("FAILED");
      const error = String((await task()).error);
      const logs = await taskLogs("task");
      for (const text of [error, logs]) {
        expect(text).not.toContain("private detail");
        expect(text).not.toContain("@example.com");
        expect(text).not.toContain("ads-key-secret");
      }
      expect(error).toContain("do not reset");
      expect(logs).toContain("part of it may have been applied");
    } finally {
      wire.fetch.mockRestore();
    }
  });
  it("stays stopped after a failed operation: the next run replays it, reads nothing and creates no operation", async () => {
    const f = audienceFixture();
    const wire = openAi({ failOperations: true });
    try {
      f.setRows([{ id: "a@example.com" }]);
      expect(await execute(f.input)).toBe("FAILED");
      // OpenAI is healthy again and the member has left the model, but the failed Add may have applied part of its
      // change. The run is not discarded: the next run replays the same request (the same operation comes back, still
      // failed), reads nothing and sends no removal.
      heal(wire);
      f.input.taskId = "openai-audience-after-failed";
      f.setRows([]);
      f.calls.length = 0;
      expect(await execute(f.input)).toBe("FAILED");
      expect(f.calls).not.toContain("reader");
      expect(adds(wire)).toEqual(["add:1"]);
    } finally {
      wire.fetch.mockRestore();
    }
  });
  it("stays stopped after a failed Add in a later run, even when the member has left the model since", async () => {
    const f = audienceFixture();
    const wire = openAi();
    try {
      f.setRows([{ id: "a@example.com" }]);
      expect(await execute(f.input)).toBe("COMPLETE");
      // Run 2 adds b, and OpenAI reports the operation failed: part of it may have been applied.
      wire.options.failOperations = true;
      f.input.taskId = "openai-audience-failed-add";
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }]);
      expect(await execute(f.input)).toBe("FAILED");
      // b has left the model. If b was applied it must not be forgotten: no fresh read of the source, no completion.
      heal(wire);
      f.input.taskId = "openai-audience-after-failed-add";
      f.setRows([{ id: "a@example.com" }]);
      f.calls.length = 0;
      expect({ outcome: await execute(f.input), readSource: f.calls.includes("reader") }).toEqual({
        outcome: "FAILED",
        readSource: false,
      });
      expect(adds(wire)).toEqual(["add:1", "add:1"]);
    } finally {
      wire.fetch.mockRestore();
    }
  });
  it("stays stopped after a failed Remove: members that came back are not forgotten by a fresh read", async () => {
    const f = audienceFixture();
    const wire = openAi();
    try {
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }]);
      expect(await execute(f.input)).toBe("COMPLETE");
      // Run 2 removes b; OpenAI reports the Remove failed, so b may or may not still be in the audience.
      wire.options.failFrom = 2;
      f.input.taskId = "openai-audience-failed-remove";
      f.setRows([{ id: "a@example.com" }]);
      expect(await execute(f.input)).toBe("FAILED");
      heal(wire);
      f.input.taskId = "openai-audience-after-failed-remove";
      f.setRows([{ id: "a@example.com" }, { id: "b@example.com" }]);
      f.calls.length = 0;
      expect({ outcome: await execute(f.input), readSource: f.calls.includes("reader") }).toEqual({
        outcome: "FAILED",
        readSource: false,
      });
      expect(adds(wire)).toEqual(["add:2", "remove:1"]);
    } finally {
      wire.fetch.mockRestore();
    }
  });
  it("stays stopped after the second of two batches fails: the first batch was accepted", async () => {
    const f = audienceFixture();
    // A journal record budget of 2 makes three members two Add batches, without thousands of rows (a 3,001-row
    // version took over 30 s on a loaded CI runner, past the 10 s idle-in-transaction limit, and was cancelled).
    const bounded = new Database(
      {
        host: container.getHost(),
        port: container.getMappedPort(5432),
        database: "runner_test",
        user: "runner_runtime",
        password: "runtime",
      },
      { ...storage, limits: { batchRecords: 2 } }
    );
    f.input.db = bounded;
    const wire = openAi({ failFrom: 2 });
    try {
      f.setRows([{ id: "m0@example.com" }, { id: "m1@example.com" }, { id: "m2@example.com" }]);
      expect(await execute(f.input)).toBe("FAILED");
      // The first batch (2 members) was accepted, the second (1 member) failed.
      expect(adds(wire)).toEqual(["add:2", "add:1"]);
      heal(wire);
      f.input.taskId = "openai-audience-after-failed-second-batch";
      f.setRows([{ id: "m0@example.com" }]);
      f.calls.length = 0;
      expect({ outcome: await execute(f.input), readSource: f.calls.includes("reader") }).toEqual({
        outcome: "FAILED",
        readSource: false,
      });
      expect(adds(wire)).toEqual(["add:2", "add:1"]);
    } finally {
      wire.fetch.mockRestore();
      await bounded.close();
    }
  });
  it("keeps a refused replay uncertain: the original request may have been applied, so the member is still removed later", async () => {
    const f = audienceFixture();
    const wire = openAi({ loseAddResponses: 1 });
    try {
      // Run 1: the Add is applied at OpenAI but its answer is lost, so the batch is unknown.
      f.setRows([{ id: "a@example.com" }]);
      expect(await execute(f.input)).toBe("FAILED");
      expect(adds(wire)).toEqual(["add:1"]);
      // Run 2: recovery replays the request and OpenAI refuses it (access revoked, say). That does not show the original
      // request applied nothing, so the batch must stay unknown, not become a rejection.
      wire.options.refuseSubmit = 403;
      f.input.taskId = "openai-audience-replay-refused";
      expect(await execute(f.input)).toBe("FAILED");
      // Access is back and the member has left the model. Recovery finishes the original run, and the member, applied by
      // the original Add, is then removed.
      wire.options.refuseSubmit = undefined;
      f.setRows([]);
      f.input.taskId = "openai-audience-replay-recovered";
      await execute(f.input);
      f.input.taskId = "openai-audience-after-recovered";
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(adds(wire)).toEqual(["add:1", "remove:1"]);
    } finally {
      wire.fetch.mockRestore();
    }
  });
  it("validates the first snapshot before creating the audience, so a wrong mapping stays correctable", async () => {
    const f = audienceFixture();
    const wire = openAi();
    try {
      // A value that is not an email: the mapping passes save-time validation but the first row cannot be normalized.
      f.setRows([{ id: "not-an-email" }]);
      expect(await execute(f.input)).toBe("FAILED");
      expect(wire.requests.filter(r => r.method === "POST" && r.path === "/custom_audiences")).toHaveLength(0);
      const state = await admin.query("SELECT count(*)::int AS n FROM newjitsu.source_state WHERE stream=$1", [
        "_REVERSE_ETL_OPENAI_AUDIENCE_",
      ]);
      expect(state.rows[0].n).toBe(0);
      // The user fixes the data (or the mapping): nothing was locked, and the next run creates the audience once.
      f.input.taskId = "openai-audience-corrected";
      f.setRows([{ id: "a@example.com" }]);
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(wire.requests.filter(r => r.method === "POST" && r.path === "/custom_audiences")).toHaveLength(1);
    } finally {
      wire.fetch.mockRestore();
    }
  });
});

describe("runner-owned Google audience provisioning", () => {
  it.each(["invalid-row", "duplicate-key", "source-failure", "identifier-type"])(
    "validates the whole first snapshot before creating an audience: %s",
    async scenario => {
      const f = fixture();
      f.input.config.destination = {
        destinationType: "google-ads",
        authorized: true,
        oauthConnectionId: "destination.destination",
        customerId: "1234567890",
      };
      f.input.config.options = {
        ...f.input.config.options,
        mode: "mirror",
        mapping: { email: "email" },
        streamOptions: {
          audience: { kind: "managed", displayName: "Test" },
          customerMatchTermsAccepted: true,
          exclusiveManagementConfirmed: true,
        },
      };
      f.input.adapters = createAdapterRegistry(async () => "token");
      if (scenario === "identifier-type") f.input.config.options.streamOptions.identifierType = "CRM_ID";
      const inputRows = [
        { id: "a", email: "good@example.com" },
        {
          id: scenario === "duplicate-key" ? "a" : "b",
          email: scenario === "invalid-row" ? "invalid" : "valid@example.com",
        },
      ];
      f.setRows(inputRows);
      if (scenario === "source-failure") {
        const reader = f.input.reader;
        f.input.reader = config => {
          const original = reader(config);
          return {
            ...original,
            stream: async function* (model, after, signal) {
              yield* original.stream(model, after, signal);
              throw new Error("private SQL error");
            },
          };
        };
      }
      const wire = vi.spyOn(globalThis, "fetch").mockRejectedValue(new Error("Unexpected Google call"));
      try {
        expect(await execute(f.input)).toBe("FAILED");
        expect(wire).not.toHaveBeenCalled();
        expect(
          (await admin.query("SELECT count(*) FROM newjitsu.source_state WHERE stream=$1", [googleAudienceStateStream]))
            .rows[0].count
        ).toBe("0");
        expect(await control()).toBeUndefined();
        expect(await taskLogs("task")).not.toContain("private SQL");
      } finally {
        wire.mockRestore();
      }
    }
  );
  it("does not persist an impossible mobile audience creation intent", async () => {
    const f = fixture();
    f.input.config.destination = {
      destinationType: "google-ads",
      authorized: true,
      oauthConnectionId: "destination.destination",
      customerId: "1234567890",
    };
    f.input.config.options = {
      ...f.input.config.options,
      mode: "mirror",
      stream: "audience",
      mapping: { mobileAdvertisingId: "id" },
      streamOptions: {
        audience: { kind: "managed", displayName: "Mobile" },
        identifierType: "MOBILE_ADVERTISING_ID",
        customerMatchTermsAccepted: true,
        exclusiveManagementConfirmed: true,
      },
    };
    f.input.adapters = createAdapterRegistry(async () => "token");
    expect(await execute(f.input)).toBe("FAILED");
    expect((await task()).error).toContain("Set the App ID and mobile platform");
    expect(
      (await admin.query("SELECT count(*) FROM newjitsu.source_state WHERE stream=$1", [googleAudienceStateStream]))
        .rows[0].count
    ).toBe("0");
    expect((await admin.query("SELECT count(*) FROM newjitsu.reverse_sync_control")).rows[0].count).toBe("0");
    expect(f.calls).not.toContain("reader");
  });
  it("stops polling terminal ambiguous conversion results without reopening the model", async () => {
    const f = asynchronousFixture();
    expect(await execute(f.input)).toBe("PENDING");
    const bind = f.adapter.recovery!;
    f.adapter.recovery = state => ({
      ...bind(state),
      reconcileBatch: async () => {
        throw new ReverseEtlManualReconciliationError();
      },
    });
    await admin.query(
      `UPDATE newjitsu.source_task SET metrics=jsonb_set(metrics,'{reverseRecovery,nextCheckAt}',to_jsonb('2000-01-01T00:00:00.000Z'::text)) WHERE status='PENDING'`
    );
    f.input = { ...f.input, taskId: "terminal-check", recoveryOf: "task", trigger: "recovery" };
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("FAILED");
    expect((await task()).error).toContain("partial or unverified conversion results");
    expect(f.calls).not.toContain("reader");
  });
  it.each(["click-conversions", "call-conversions", "conversion-adjustments"])(
    "runs %s and skips previously submitted primary keys on subsequent extraction",
    async stream => {
      const f = fixture();
      f.input.config.destination = {
        destinationType: "google-ads",
        authorized: true,
        oauthConnectionId: "destination.destination",
        customerId: "1234567890",
        developerToken: "test-token",
      };
      f.input.config.model.cursor = undefined;
      f.input.config.options = {
        ...f.input.config.options,
        mode: "upsert",
        stream,
        mapping:
          stream === "click-conversions"
            ? { gclid: "id", conversionTimestamp: "time" }
            : stream === "call-conversions"
            ? { callerId: "caller", callTimestamp: "time", conversionTimestamp: "time" }
            : { orderId: "id", adjustmentTimestamp: "time" },
        streamOptions: {
          conversionActionId: "123",
          ...(stream === "conversion-adjustments" ? { adjustmentType: "RETRACTION" } : {}),
        },
      };
      f.input.adapters = createAdapterRegistry(async () => "token");
      const rows = (ids: string[]) => ids.map(id => ({ id, time: "2026-09-20T12:30:00Z", caller: "+14155552671" }));
      f.setRows(rows(["a", "b"]));
      const uploads: any[] = [];
      const wire = vi.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) => {
        if (String(_url).includes("requestStatus:retrieve"))
          return new Response(
            JSON.stringify({
              requestStatusPerDestination: [
                {
                  destination: {
                    operatingAccount: { accountType: "GOOGLE_ADS", accountId: "1234567890" },
                    productDestinationId: "123",
                  },
                  requestStatus: "SUCCESS",
                  eventsIngestionStatus: { recordCount: "2" },
                },
              ],
            })
          ) as Awaited<ReturnType<typeof globalThis.fetch>>;
        const body = JSON.parse(init!.body as string);
        uploads.push(body);
        return new Response(
          JSON.stringify(
            stream === "click-conversions"
              ? { requestId: `req-${uploads.length}` }
              : {
                  results: (body.conversions ?? body.conversionAdjustments).map(() => ({
                    conversionAction: "customers/1234567890/conversionActions/123",
                  })),
                }
          )
        ) as Awaited<ReturnType<typeof globalThis.fetch>>;
      });
      try {
        expect(await execute(f.input)).toBe(stream === "click-conversions" ? "PENDING" : "COMPLETE");
        if (stream === "click-conversions") {
          await admin.query(
            `UPDATE newjitsu.source_task SET metrics=jsonb_set(metrics,'{reverseRecovery,nextCheckAt}',to_jsonb('2000-01-01T00:00:00.000Z'::text)) WHERE task_id='task'`
          );
          expect(await execute({ ...f.input, taskId: "accepted-check", trigger: "recovery", recoveryOf: "task" })).toBe(
            "COMPLETE"
          );
        }
        f.input.taskId = "second-event-run";
        // Different payload, same primary key: insert-only must still omit a/b.
        f.setRows(rows(["a", "b", "c"]).map(row => ({ ...row, time: "2026-09-20T13:30:00Z" })));
        expect(await execute(f.input)).toBe(stream === "click-conversions" ? "PENDING" : "COMPLETE");
        expect(uploads).toHaveLength(2);
        const key =
          stream === "click-conversions"
            ? "events"
            : stream === "call-conversions"
            ? "conversions"
            : "conversionAdjustments";
        expect(uploads[0][key]).toHaveLength(2);
        expect(uploads[1][key]).toHaveLength(1);
        expect(await taskLogs("second-event-run")).toContain("2 previously submitted events omitted");
        expect(
          (await admin.query("SELECT count(*) FROM newjitsu.source_state WHERE stream=$1", [googleAudienceStateStream]))
            .rows[0].count
        ).toBe("0");
      } finally {
        wire.mockRestore();
      }
    }
  );
  it.each(["normal", "lost-response", "empty-discovery", "oauth-failure"])(
    "persists first-run provisioning and does not create twice: %s",
    async scenario => {
      const f = fixture();
      f.input.config.destination = {
        destinationType: "google-ads",
        authorized: true,
        oauthConnectionId: "destination.destination",
        customerId: "1234567890",
      };
      f.input.config.options = {
        ...f.input.config.options,
        mode: "mirror",
        mapping: { email: "id" },
        streamOptions: {
          audience: { kind: "managed", displayName: "Test" },
          customerMatchTermsAccepted: true,
          exclusiveManagementConfirmed: true,
          mirrorStrategy: "snapshot-diff",
        },
      };
      f.setRows([]);
      let tokenFails = scenario === "oauth-failure",
        creates = 0,
        remote: any;
      f.input.adapters = createAdapterRegistry(async () => {
        if (tokenFails) throw new Error("private-token");
        return "token";
      });
      const saved = async () =>
        (
          await admin.query("SELECT state FROM newjitsu.source_state WHERE sync_id='sync' AND stream=$1", [
            googleAudienceStateStream,
          ])
        ).rows[0]?.state;
      const response = (body: unknown) => new Response(JSON.stringify(body)) as Awaited<ReturnType<typeof fetch>>;
      const wire = vi.spyOn(globalThis, "fetch").mockImplementation(async (url, init) => {
        expect(f.calls).toContain("lease");
        if (init?.method === "POST") {
          creates++;
          expect((await saved()).phase).toBe("submitting");
          expect((await admin.query("SELECT count(*) FROM newjitsu.reverse_sync_control")).rows[0].count).toBe("0");
          remote = {
            ...JSON.parse(init.body as string),
            id: "123",
            name: "accountTypes/GOOGLE_ADS/accounts/1234567890/userLists/123",
            accessReason: "OWNED",
          };
          if (scenario === "lost-response" || scenario === "empty-discovery") throw new Error("lost response");
          return response(remote);
        }
        if (String(url).includes("?")) return response({ userLists: scenario === "empty-discovery" ? [] : [remote] });
        return response(remote);
      });
      try {
        expect(await execute(f.input)).toBe(scenario === "normal" ? "COMPLETE" : "FAILED");
        expect(f.calls.filter(call => call === "reader")).toHaveLength(1);
        expect((await saved()).phase).toBe(
          scenario === "normal" ? "ready" : scenario === "oauth-failure" ? "prepared" : "submitting"
        );
        tokenFails = false;
        f.input.taskId = "next";
        expect(await execute(f.input)).toBe(scenario === "empty-discovery" ? "FAILED" : "COMPLETE");
        expect(creates).toBe(1);
        if (scenario !== "empty-discovery") {
          expect((await saved()).audienceId).toBe("123");
          expect((await control()).revision).toBe(f.input.config.configRevision);
          expect((await control()).target_hash).toBeDefined();
        }
        expect(f.input.config.options.streamOptions.audienceId).toBeUndefined();
      } finally {
        wire.mockRestore();
      }
    }
  );
});

let container: StartedTestContainer;
let admin: Client;
let db: Database;
beforeAll(async () => {
  container = await new GenericContainer("postgres:18-alpine")
    .withEnvironment({ POSTGRES_PASSWORD: "test", POSTGRES_DB: "runner_test" })
    .withExposedPorts(5432)
    .withWaitStrategy(Wait.forLogMessage(/database system is ready to accept connections/, 2))
    .start();
  const config = {
    host: container.getHost(),
    port: container.getMappedPort(5432),
    database: "runner_test",
    user: "postgres",
    password: "test",
  };
  admin = new Client(config);
  await admin.connect();
  execFileSync(
    process.execPath,
    [
      createRequire(import.meta.url).resolve("prisma/build/index.js"),
      "db",
      "push",
      `--schema=${fileURLToPath(new URL("../../../webapps/console/prisma/schema.prisma", import.meta.url))}`,
      "--skip-generate",
    ],
    {
      env: {
        // eslint-disable-next-line no-restricted-properties -- disposable test database only.
        ...process.env,
        DATABASE_URL: `postgresql://postgres:test@${config.host}:${config.port}/${config.database}?schema=newjitsu`,
      },
      stdio: "inherit",
    }
  );
  await admin.query(
    "CREATE ROLE runner_runtime LOGIN PASSWORD 'runtime'; GRANT USAGE ON SCHEMA newjitsu TO runner_runtime"
  );
  for (const table of ["control", "target_owner"])
    await admin.query(`GRANT SELECT,INSERT,UPDATE,DELETE ON newjitsu.reverse_sync_${table} TO runner_runtime`);
  await admin.query(
    "GRANT SELECT,INSERT,UPDATE ON newjitsu.source_state,newjitsu.source_task TO runner_runtime; GRANT INSERT ON newjitsu.task_log TO runner_runtime"
  );
  db = new Database({ ...config, user: "runner_runtime", password: "runtime" }, storage);
}, 60_000);
afterAll(async () => {
  await db?.close();
  await admin?.end();
  await container?.stop();
});
beforeEach(async () => {
  await admin.query(
    "TRUNCATE newjitsu.reverse_sync_control,newjitsu.reverse_sync_target_owner,newjitsu.source_state,newjitsu.source_task,newjitsu.task_log"
  );
});
const config = () =>
  ReverseRunConfig.parse({
    version: 1,
    kind: "reverse",
    id: "sync",
    workspaceId: "workspace",
    fromId: "model",
    toId: "destination",
    configRevision: "a".repeat(64),
    updatedAt: new Date().toISOString(),
    model: { warehouseId: "warehouse", query: "SELECT id FROM users", primaryKey: ["id"] },
    warehouse: { destinationType: "postgres" },
    destination: { destinationType: "test" },
    options: { stream: "audience", mode: "upsert", mapping: { id: "id" } },
  });
const accepted = (batch: WriteBatch<JsonObject>) => ({
  outcomes: batch.records.map(row => ({ operationId: row.operationId, status: "accepted" as const })),
});
function fixture() {
  const calls: string[] = [];
  const writes: WriteBatch<JsonObject>[] = [];
  let pending = false;
  let failInit = false;
  let failBatch = false;
  let rows: { id: string; email?: string }[] = [{ id: "a" }, { id: "b" }];
  const writer = (ctx: ReverseEtlContext<JsonObject, JsonObject>) => ({
    init: async () => {
      calls.push("init");
      if (failInit) throw new Error("private-token");
      await ctx.delivery.saveProviderState({ session: "existing" });
    },
    upsert: async (batch: WriteBatch<JsonObject>) => {
      calls.push("upsert");
      writes.push(batch);
      if (failBatch) throw new Error("lost-response");
      return accepted(batch);
    },
    remove: async (batch: WriteBatch<JsonObject>) => {
      calls.push("remove");
      return accepted(batch);
    },
    finish: async () => {
      calls.push("finish");
      return pending ? { delivery: "pending" as const, remoteJobIds: ["job"] } : { delivery: "accepted" as const };
    },
    abort: async () => {
      calls.push("abort");
    },
  });
  const adapter: RuntimeAdapter = {
    credentials: {},
    targetIdentity: "test/account/audience",
    project: (_action, row: any) => [{ identity: String(row.id), upsert: { id: row.id }, remove: { id: row.id } }],
    stream: {
      name: "audience",
      displayName: "Audience",
      rowType: z.object({ id: z.string() }),
      removeRowType: z.object({ id: z.string() }),
      options: z.object({}),
      batchSize: 2,
      capabilities: {
        supportsUpsert: true,
        supportsExplicitRemove: true,
        mirror: "snapshot-diff",
        replay: "idempotent-operation",
      },
      createWriter: async ctx => {
        calls.push("create");
        return writer(ctx);
      },
    },
    verifyMirrorBaseline: async () => "tracked",
    recovery: () => ({
      attachWriter: async ctx => {
        calls.push("attach");
        return writer(ctx);
      },
      reconcileBatch: async batch => {
        calls.push("reconcile");
        expect(batch).toEqual(writes.at(-1));
        return accepted(batch);
      },
      reconcileFinish: async () => {
        calls.push("reconcileFinish");
        return { delivery: "accepted" };
      },
      reconcileInit: async () => {
        calls.push("reconcileInit");
        return "absent";
      },
      reconcileAbort: async () => {
        calls.push("reconcileAbort");
      },
    }),
  };
  adapter.mirror = {
    stream: adapter.stream,
    batchDelivery: "accepted",
    projection: { rowType: adapter.stream.rowType, project: row => adapter.project("upsert", row) },
  };
  const input: ExecuteOptions = {
    config: config(),
    db,
    taskId: "task",
    trigger: "scheduled",
    adapters: new Map([["test", () => adapter]]),
    controller: new AbortController(),
    lease: {
      acquire: async () => {
        calls.push("lease");
      },
      renew: async () => {
        calls.push("renew");
      },
      release: async () => {
        calls.push("release");
      },
    },
    admit: async () => input.config,
    reader: () => {
      calls.push("reader");
      return {
        sql: {} as any,
        columns: async () => [],
        preview: async () => ({ rows: [], columns: [], truncated: false }),
        close: async () => {
          calls.push("close");
        },
        stream: async function* (_model, after, signal) {
          calls.push("source");
          expect(signal).toBe(input.controller.signal);
          for (const row of rows)
            yield {
              row,
              deleted: false,
              ...(input.config.model.cursor ? { checkpoint: { value: row.id, primaryKeyValues: [row.id] } } : {}),
            };
        },
      };
    },
  };
  return {
    input,
    calls,
    adapter,
    writes,
    setPending: () => {
      pending = true;
    },
    setFailInit: () => {
      failInit = true;
    },
    setFailBatch: () => {
      failBatch = true;
    },
    setRows: (value: typeof rows) => {
      rows = value;
    },
  };
}
async function task(id = "task") {
  return (await admin.query("SELECT * FROM newjitsu.source_task WHERE task_id=$1", [id])).rows[0];
}
async function taskLogs(id: string) {
  return (await admin.query("SELECT message FROM newjitsu.task_log WHERE task_id=$1 ORDER BY timestamp", [id])).rows
    .map(row => row.message)
    .join("\n");
}
async function control() {
  return (await admin.query("SELECT * FROM newjitsu.reverse_sync_control ORDER BY run_order DESC LIMIT 1")).rows[0];
}

function asynchronousFixture() {
  const f = fixture();
  const receipts = new Map<string, BatchResult>();
  f.adapter.stream.batchDelivery = "asynchronous";
  const create = f.adapter.stream.createWriter;
  f.adapter.stream.createWriter = async ctx => ({
    ...(await create(ctx)),
    upsert: async batch => {
      f.calls.push("upsert");
      f.writes.push(batch);
      const result: BatchResult = {
        outcomes: batch.records.map(row => ({ operationId: row.operationId, status: "staged" })),
        remoteJobIds: [batch.batchId],
      };
      receipts.set(batch.batchId, result);
      return result;
    },
  });
  const recovery = f.adapter.recovery!;
  f.adapter.recovery = state => ({
    ...recovery(state),
    reconcileBatch: async batch => {
      f.calls.push("reconcile");
      return receipts.get(batch.batchId)!;
    },
  });
  return { ...f, receipts };
}

describe("retention bucket routing through the run loop", () => {
  // With a retention bucket the projected effects, which stay in the main store, must hold no row data (as for webhook).
  const payloadFree = (f: ReturnType<typeof fixture>) => {
    f.adapter.project = (_action, row: any) => [
      { identity: createHash("sha256").update(String(row.id)).digest("hex"), upsert: {}, remove: {} },
    ];
  };
  const decoded = (store: MemoryObjects) =>
    [...store.objects.entries()].map(([key, value]) => ({
      key,
      value: JSON.parse(gunzipSync(value).toString()).value as any,
    }));
  // A batch carries rows either inline (no retention bucket) or as a separate artifact; the manifest has null rows.
  const carriesRows = (value: any) =>
    (Array.isArray(value?.rows) && value.rows.length > 0) ||
    !!value?.batch?.records?.some((record: any) => record.row !== null && record.row !== undefined);
  function retentionDb(main: MemoryObjects, retention: MemoryObjects) {
    return new Database(
      {
        host: container.getHost(),
        port: container.getMappedPort(5432),
        database: "runner_test",
        user: "runner_runtime",
        password: "runtime",
      },
      { objectStorage: { store: main, signal: new AbortController().signal, retention } }
    );
  }

  it("stores the rows, and only the rows, in the retention bucket", async () => {
    const main = new MemoryObjects(),
      retention = new MemoryObjects(),
      rdb = retentionDb(main, retention);
    try {
      const f = fixture();
      payloadFree(f);
      f.input.db = rdb;
      expect(await execute(f.input)).toBe("COMPLETE");
      const kept = decoded(retention);
      expect(kept.length).toBeGreaterThan(0);
      expect(kept.every(object => object.key.startsWith("r1/") && Array.isArray(object.value.rows))).toBe(true);
      expect(kept.flatMap(object => object.value.rows.map((row: any) => row.id)).sort()).toEqual(["a", "b"]);
      const rest = decoded(main);
      expect(rest.length).toBeGreaterThan(0);
      expect(rest.every(object => object.key.startsWith("v1/") && !carriesRows(object.value))).toBe(true);
      // The batch manifest stays in the main store, with its rows replaced by a reference.
      const manifests = rest.filter(object => object.value?.batch?.records);
      expect(manifests.length).toBeGreaterThan(0);
      expect(
        manifests.every(object => object.value.rows && object.value.batch.records.every((r: any) => r.row === null))
      ).toBe(true);
    } finally {
      await rdb.close();
    }
  });

  it("a destination whose projections hold row data is refused with a readable message when retention is configured", async () => {
    const main = new MemoryObjects(),
      retention = new MemoryObjects(),
      rdb = retentionDb(main, retention);
    try {
      const f = fixture(); // the default projection keeps the row in the effects, which would never expire
      f.input.db = rdb;
      expect(await execute(f.input)).toBe("FAILED");
      expect((await task()).error).toContain("cannot expire");
      expect(retention.objects.size).toBe(0);
      expect(f.calls).not.toContain("upsert");
    } finally {
      await rdb.close();
    }
  });

  it("a run without a retention bucket stores everything in the main bucket, as before", async () => {
    const main = new MemoryObjects(),
      rdb = retentionDb(main, undefined as any);
    try {
      const f = fixture();
      payloadFree(f);
      f.input.db = rdb;
      expect(await execute(f.input)).toBe("COMPLETE");
      const all = decoded(main);
      expect(all.every(object => object.key.startsWith("v1/"))).toBe(true);
      expect(all.some(object => carriesRows(object.value))).toBe(true);
    } finally {
      await rdb.close();
    }
  });

  it("after the retention window the next run still works: only an unresolved batch needs its rows", async () => {
    const main = new MemoryObjects(),
      retention = new MemoryObjects(),
      rdb = retentionDb(main, retention);
    try {
      const f = fixture();
      payloadFree(f);
      f.input.db = rdb;
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(retention.objects.size).toBeGreaterThan(0);
      retention.objects.clear(); // the bucket's lifecycle rule has deleted every expired row object
      f.input = { ...f.input, taskId: "task-after-expiry" };
      expect(await execute(f.input)).toBe("COMPLETE");
      // The new run wrote its own rows; the expired objects were never needed.
      expect(retention.objects.size).toBeGreaterThan(0);
    } finally {
      await rdb.close();
    }
  });

  it("a poison record stops the run; after its rows expire and the endpoint is fixed the next run delivers everything", async () => {
    const main = new MemoryObjects(),
      retention = new MemoryObjects(),
      rdb = retentionDb(main, retention);
    try {
      const f = fixture();
      payloadFree(f);
      f.input.db = rdb;
      // First run: the destination rejects one record, which stops the run (a rejected batch stays in the head).
      const createWriter = f.adapter.stream.createWriter;
      f.adapter.stream.createWriter = async ctx => {
        const writer = await createWriter(ctx);
        return {
          ...writer,
          upsert: async batch => ({
            outcomes: batch.records.map((record, i) =>
              i === 0
                ? {
                    operationId: record.operationId,
                    status: "rejected" as const,
                    code: "invalid",
                    safeReason: "Invalid",
                  }
                : { operationId: record.operationId, status: "accepted" as const }
            ),
          }),
        };
      };
      expect(await execute(f.input)).toBe("FAILED");
      expect(retention.objects.size).toBeGreaterThan(0);
      // The window passes before anyone fixes the endpoint: the bucket's lifecycle rule deletes the rows.
      retention.objects.clear();
      // Endpoint fixed. The next scheduled run must not need the expired rows.
      f.adapter.stream.createWriter = createWriter;
      f.writes.length = 0;
      f.input = { ...f.input, taskId: "task-after-fix" };
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(f.writes.flatMap(batch => batch.records.map(record => (record.row as any).id)).sort()).toEqual(["a", "b"]);
    } finally {
      await rdb.close();
    }
  });

  it("replay refuses rows that do not match the batch manifest instead of sending them", async () => {
    const main = new MemoryObjects(),
      retention = new MemoryObjects(),
      rdb = retentionDb(main, retention);
    // A writer bug: the stored rows artifact is internally valid but is not the batch the manifest describes.
    const original = Artifacts.prototype.put;
    const spy = vi
      .spyOn(Artifacts.prototype, "put")
      .mockImplementation(function (this: Artifacts, value: any, kind?: any) {
        return original.call(
          this,
          kind === "rows" ? { rows: value.rows.map(() => ({ id: "not-the-batch" })) } : value,
          kind
        );
      });
    try {
      const f = fixture();
      payloadFree(f);
      f.input.db = rdb;
      f.setFailBatch(); // leaves the batch unknown, so the next run must replay it
      expect(await execute(f.input)).toBe("FAILED");
      spy.mockRestore();
      f.input = { ...f.input, taskId: "task-replay" };
      f.calls.length = 0;
      expect(await execute(f.input)).toBe("FAILED");
      // Stopped before the provider was asked to replay anything.
      expect(f.calls).not.toContain("reconcile");
      expect(f.writes.every(batch => batch.records.every(record => (record.row as any).id !== "not-the-batch"))).toBe(
        true
      );
    } finally {
      spy.mockRestore();
      await rdb.close();
    }
  });

  it("a batch that is still unresolved when its rows expire blocks delivery with the recovery message", async () => {
    const main = new MemoryObjects(),
      retention = new MemoryObjects(),
      rdb = retentionDb(main, retention);
    try {
      const f = fixture();
      payloadFree(f);
      f.input.db = rdb;
      f.setFailBatch(); // the destination loses the response: the batch is left unknown
      expect(await execute(f.input)).toBe("FAILED");
      retention.objects.clear();
      f.input = { ...f.input, taskId: "task-after-expiry" }; // the next scheduled run starts recovery, as in production
      const result = await execute(f.input);
      expect(result).toBe("FAILED");
      expect((await task("task-after-expiry")).error).toContain("Saved sync data is missing or unreadable");
    } finally {
      await rdb.close();
    }
  });
});

describe("executable runner", () => {
  it("records actionable duplicate-identity failures in task details and logs without member values", async () => {
    const f = asynchronousFixture();
    // Distinct model primary keys can normalize to the same destination identity.
    f.adapter.project = () => [
      {
        identity: "private@example.com",
        upsert: { email: "private@example.com" },
        remove: { email: "private@example.com" },
      },
    ];
    expect(await execute(f.input)).toBe("FAILED");
    const failed = await task();
    expect(failed.error).toContain("Multiple source rows identify the same audience member");
    expect(failed.error).toContain("Update the model");
    expect(failed.error).toContain("Run ID: task");
    expect(failed.description).toBe(failed.error);
    expect(failed.error).not.toContain("private@example.com");
    expect((await admin.query("SELECT message FROM newjitsu.task_log WHERE level='ERROR'")).rows).toEqual([
      { message: failed.error },
    ]);
    expect(f.writes).toHaveLength(0);
  });
  it("explains audience ownership conflicts before opening the warehouse", async () => {
    const mirror = fixture();
    mirror.input.config.options.mode = "mirror";
    expect(await execute(mirror.input)).toBe("COMPLETE");
    const f = fixture();
    f.input.config.id = "another-sync";
    f.input.taskId = "another-task";
    expect(await execute(f.input)).toBe("FAILED");
    expect((await task("another-task")).error).toContain("This audience is reserved by another sync");
    expect(f.calls).not.toContain("source");
  });
  it("gives a support reference instead of exposing unexpected provider errors", async () => {
    const f = fixture();
    f.setFailInit();
    expect(await execute(f.input)).toBe("FAILED");
    const failed = await task();
    expect(failed.error).toContain("Contact support or your Jitsu administrator");
    expect(failed.error).toContain("Run ID: task");
    expect(failed.error).not.toMatch(/private-token|inspect.*state/);
  });
  const makeDue = () =>
    admin.query(
      `UPDATE newjitsu.source_task SET metrics=jsonb_set(metrics,'{reverseRecovery,nextCheckAt}',to_jsonb('2000-01-01T00:00:00.000Z'::text)) WHERE status='PENDING'`
    );
  async function refresh(f: { input: ExecuteOptions }, worker: string, parent = "task") {
    await makeDue();
    f.input = { ...f.input, taskId: worker, recoveryOf: parent, trigger: "recovery" };
  }
  it("refreshes paused delivery without extraction and rejects new manual or scheduled runs", async () => {
    const f = asynchronousFixture();
    expect(await execute(f.input)).toBe("PENDING");
    f.input.config.options.disabled = true;
    for (const trigger of ["manual", "scheduled"] as const) {
      f.calls.length = 0;
      expect(await execute({ ...f.input, taskId: trigger, trigger })).toBe("FAILED");
      expect(f.calls).not.toContain("reader");
      expect(f.calls).not.toContain("upsert");
    }
    for (const batch of f.writes) f.receipts.set(batch.batchId, accepted(batch));
    await refresh(f, "paused-check");
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("COMPLETE");
    expect(f.calls).toContain("reconcile");
    expect(f.calls).not.toContain("reader");
  });
  it("refuses an automatic refresh queued before the sync was paused", async () => {
    const f = asynchronousFixture();
    expect(await execute(f.input)).toBe("PENDING");
    await refresh(f, "stale-auto-check");
    f.input.admit = async () => ({ ...f.input.config, options: { ...f.input.config.options, disabled: true } });
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("PENDING");
    expect(f.calls).not.toContain("reader");
    expect(f.calls).not.toContain("reconcile");
    expect(await taskLogs("task")).toContain("Status refresh failed without updating delivery outcomes");
  });
  it("retries rejected conversion keys after explicit cleanup without replaying accepted keys", async () => {
    const f = asynchronousFixture();
    f.input.config.model.cursor = undefined;
    f.adapter.insertOnly = true;
    f.adapter.project = (_action, row: any) => [
      { identity: { eventKey: recordKey([row.id]) }, upsert: row, remove: {} },
    ];
    expect(await execute(f.input)).toBe("PENDING");
    const batch = f.writes[0];
    f.input.taskId = "overlap";
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.writes).toHaveLength(1);
    f.receipts.set(batch.batchId, {
      outcomes: batch.records.map((row, i) =>
        i === 0
          ? { operationId: row.operationId, status: "accepted" }
          : { operationId: row.operationId, status: "rejected", code: "invalid", safeReason: "Invalid" }
      ),
    });
    await refresh(f, "rejected-check");
    expect(await execute(f.input)).toBe("FAILED");
    expect((await control()).phase).toBe("batches_pending");
    // Explicit retry cleans up the failed saved run, but does not re-upload it.
    await admin.query("UPDATE newjitsu.source_task SET status='PENDING' WHERE task_id='task'");
    await refresh(f, "cleanup-check");
    expect(await execute(f.input)).toBe("FAILED");
    expect((await control()).phase).toBe("aborted");
    f.input = { ...f.input, taskId: "retry", trigger: "manual", recoveryOf: undefined };
    expect(await execute(f.input)).toBe("PENDING");
    expect(f.writes).toHaveLength(2);
    expect(f.writes[1].records.map(row => row.row.id)).toEqual(["b"]);
  });
  it.each([false, true])(
    "blocks insert-only overlap until acceptance without advancing the cursor (saved cursor: %s)",
    async seeded => {
      const f = asynchronousFixture();
      f.input.config.model.cursor = { column: "id", type: "string" };
      f.adapter.insertOnly = true;
      f.adapter.project = (_action, row: any) => [
        {
          identity: { eventKey: recordKey([row.id]) },
          upsert: row,
          remove: {},
        },
      ];
      const createWriter = f.adapter.stream.createWriter;
      f.adapter.stream.createWriter = async ctx => {
        const writer = await createWriter(ctx);
        return {
          ...writer,
          init: async () => {
            await writer.init();
            ctx.store.set("latestRun", ctx.logicalRunId);
          },
        };
      };
      const reader = f.input.reader;
      const afters: unknown[] = [];
      f.input.reader = config => {
        const original = reader(config);
        return {
          ...original,
          stream: async function* (model, after, signal) {
            afters.push(after);
            for await (const row of original.stream(model, after, signal))
              if (!after || row.checkpoint!.value > after.value) yield row;
          },
        };
      };
      const saved = async () => {
        const envelope = (await admin.query("SELECT state FROM newjitsu.source_state WHERE stream='_REVERSE_ETL_'"))
          .rows[0]?.state;
        return envelope && { ...envelope, decoded: JSON.parse(envelope.value).value };
      };
      if (seeded) {
        f.setRows([{ id: "0" }]);
        f.input.taskId = "seed";
        expect(await execute(f.input)).toBe("PENDING");
        for (const batch of f.writes) f.receipts.set(batch.batchId, accepted(batch));
        await refresh(f, "seed-check", "seed");
        expect(await execute(f.input)).toBe("COMPLETE");
        // Older version-2 state has only runOrder; upgrades must preserve its fence.
        await admin.query(
          "UPDATE newjitsu.source_state SET state=state-'checkpointRunOrder' WHERE stream='_REVERSE_ETL_'"
        );
      }
      const before = await saved();
      f.setRows([{ id: "a" }, { id: "b" }]);
      f.input = { ...f.input, taskId: "task", trigger: "manual", recoveryOf: undefined };
      const firstWrite = f.writes.length;
      expect(await execute(f.input)).toBe("PENDING");
      const runOrder = (await control()).run_order;
      const count = f.writes.length;
      for (const taskId of ["noop-1", "noop-2"]) {
        f.input.taskId = taskId;
        expect(await execute(f.input)).toBe("FAILED");
        expect(f.writes).toHaveLength(count);
        expect(await saved()).toEqual(before);
        expect((await task(taskId)).error).toContain("previous conversion run is still unresolved");
        expect((await task()).status).toBe("PENDING");
      }
      expect((await control()).run_order).toBe(runOrder);
      for (const batch of f.writes.slice(firstWrite)) f.receipts.set(batch.batchId, accepted(batch));
      await refresh(f, "accepted-check");
      expect(await execute(f.input)).toBe("COMPLETE");
      const acceptedState = await saved();
      expect(acceptedState.checkpointRunOrder).toBe(runOrder);
      expect(acceptedState.decoded.store.latestRun).toBe((await control()).run_id);
      expect(acceptedState.decoded.point.cursor).toEqual({ value: "b", primaryKeyValues: ["b"] });
      f.input = { ...f.input, taskId: "after-acceptance", trigger: "manual", recoveryOf: undefined };
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(afters.at(-1)).toEqual({ value: "b", primaryKeyValues: ["b"] });
      expect(f.writes).toHaveLength(count);
    }
  );
  it.each(["upsert", "mirror"] as const)(
    "allows a new %s upload while an older run is pending and isolates late refresh",
    async mode => {
      const f = asynchronousFixture();
      f.input.config.options.mode = mode;
      if (mode === "mirror") f.adapter.mirror!.batchDelivery = "asynchronous";
      expect(await execute(f.input)).toBe("PENDING");
      const firstRun = (await control()).run_id;
      const firstBatches = [...f.writes];
      f.input.taskId = "second";
      f.setRows([{ id: "b" }, { id: "c" }]);
      expect(await execute(f.input)).toBe("PENDING");
      const secondRun = (await control()).run_id;
      expect(secondRun).not.toBe(firstRun);
      expect((await admin.query("SELECT count(*) FROM newjitsu.reverse_sync_control")).rows[0].count).toBe("2");
      expect((await task()).status).toBe("PENDING");
      expect((await task("second")).status).toBe("PENDING");
      const secondBatches = f.writes.slice(firstBatches.length);
      expect(secondBatches.flatMap(batch => batch.records)).toHaveLength(2);
      for (const batch of secondBatches) f.receipts.set(batch.batchId, accepted(batch));
      await refresh(f, "second-check", "second");
      expect(await execute(f.input)).toBe("COMPLETE");
      const checkpoint = (await admin.query("SELECT state FROM newjitsu.source_state")).rows[0].state;
      expect(checkpoint.runOrder).toBe("1");
      expect((await task()).status).toBe("PENDING");
      for (const batch of firstBatches) f.receipts.set(batch.batchId, accepted(batch));
      await refresh(f, "first-check");
      f.calls.length = 0;
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(f.calls).not.toContain("reader");
      expect((await admin.query("SELECT state FROM newjitsu.source_state")).rows[0].state).toEqual(checkpoint);
      expect((await task()).status).toBe("COMPLETE");
      expect((await task("second")).status).toBe("COMPLETE");
      expect((await admin.query("SELECT count(*) FROM newjitsu.source_task")).rows[0].count).toBe("2");
      expect((await durable()).members.map(m => m.effect.identity).sort()).toEqual(
        mode === "mirror" ? ["b", "c"] : ["a", "b", "c"]
      );
    }
  );
  it("persists WAITING without errors, and a separate recovery trigger polls without opening SQL", async () => {
    const f = asynchronousFixture();
    const before = Date.now();
    expect(await execute(f.input)).toBe("PENDING");
    const waiting = await task();
    expect(waiting.error).toBeNull();
    expect(waiting.metrics.reverseDelivery).toMatchObject({
      upsert: { total: 1, pending: 1 },
      records: { accepted: 0, pending: 2 },
    });
    expect(waiting.started_by.workspaceId).toBe("workspace");
    expect(waiting.metrics.reverseRecovery).toMatchObject({
      runId: (await control()).run_id,
      revision: f.input.config.configRevision,
      attempt: 0,
    });
    expect(Date.parse(waiting.metrics.reverseRecovery.nextCheckAt)).toBeGreaterThanOrEqual(before + 30 * 60_000);
    expect((await admin.query("SELECT 1 FROM newjitsu.task_log WHERE level='ERROR'")).rowCount).toBe(0);
    await makeDue();
    f.input = { ...f.input, taskId: "automatic-check", trigger: "recovery", recoveryOf: "task" };
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("PENDING");
    expect(f.calls).toEqual(["lease", "renew", "reconcile", "release"]);
    expect((await task()).status).toBe("PENDING");
    const polled = await task();
    expect(polled.started_by).toEqual(waiting.started_by);
    expect(polled.started_at).toEqual(waiting.started_at);
    expect(polled.metrics.reverseRecovery.attempt).toBe(1);
    expect(polled.metrics.reverseDelivery).toMatchObject({ upsert: { total: 1, pending: 1 } });
    expect(polled.metrics.reverseRecovery.deadline).toBe(waiting.metrics.reverseRecovery.deadline);
    for (const batch of f.writes) f.receipts.set(batch.batchId, accepted(batch));
    await makeDue();
    f.input.taskId = "automatic-complete";
    f.input.recoveryOf = "task";
    expect(await execute(f.input)).toBe("COMPLETE");
    expect((await task()).status).toBe("COMPLETE");
    expect((await task()).metrics.reverseDelivery).toMatchObject({
      upsert: { total: 1, accepted: 1, pending: 0 },
      records: { accepted: 2, pending: 0 },
    });
    expect((await admin.query("SELECT count(*) FROM newjitsu.source_task")).rows[0].count).toBe("1");
    expect((await control()).phase).toBe("complete");
    expect(f.calls).not.toContain("reader");
    const waitingLogs = (await admin.query("SELECT message FROM newjitsu.task_log WHERE task_id='task'")).rows
      .map(r => r.message)
      .join("\n");
    expect(waitingLogs).toContain("2 confirmed submitted in 1 batches; 0 accepted, 2 pending");
    expect(waitingLogs).toContain("2 rows read in this attempt, 2 upsert rows, 0 explicit removal rows");
    expect(waitingLogs.match(/Delivery totals/g)).toHaveLength(3);
    expect(waitingLogs).toContain("Preparing 2 additions/upserts.");
    expect(waitingLogs).toContain("Submitted 2 additions/upserts; 0 accepted, 2 pending processing");
    expect(waitingLogs).not.toContain("may have reached the destination");
    const recoveryLogs = waitingLogs;
    expect(recoveryLogs).toContain("warehouse SQL is not re-read");
    expect(recoveryLogs).toContain("2 confirmed submitted in 1 batches; 2 accepted, 0 pending");
    expect(recoveryLogs.match(/Delivery totals/g)).toHaveLength(3);
    expect(recoveryLogs).toContain("Status updated for 2 additions/upserts: 2 accepted");
    expect(recoveryLogs.match(/Submitted 2 additions\/upserts/g)).toHaveLength(1);
    const unchangedLogs = await taskLogs("automatic-check");
    expect(unchangedLogs).toBe("");
  });
  it.each(["early", "cancelled", "revision", "run", "complete", "foreign", "duplicate"])(
    "does not start an obsolete or unauthorized recovery (%s)",
    async reason => {
      const f = asynchronousFixture();
      await execute(f.input);
      if (reason !== "early") await makeDue();
      if (reason === "cancelled") await admin.query("UPDATE newjitsu.source_task SET status='CANCELLED'");
      if (reason === "revision") f.input.config.configRevision = "b".repeat(64);
      if (reason === "run") await admin.query("UPDATE newjitsu.reverse_sync_control SET run_id='new-run'");
      if (reason === "complete") await admin.query("UPDATE newjitsu.reverse_sync_control SET phase='complete'");
      if (reason === "foreign") f.input.config.workspaceId = "foreign";
      f.input.trigger = "recovery";
      f.input.recoveryOf = "task";
      f.input.taskId = "automatic";
      if (reason === "duplicate") {
        expect(await execute(f.input)).toBe("PENDING");
        f.input.taskId = "duplicate";
      }
      f.calls.length = 0;
      expect(await execute(f.input)).toBe("FAILED");
      expect(f.calls).toEqual(["lease", "release"]);
      expect(await task(f.input.taskId)).toBeUndefined();
      if (reason === "cancelled") expect((await task()).status).toBe("CANCELLED");
      else if (reason !== "duplicate") expect((await task()).status).toBe("PENDING");
    }
  );
  it("keeps real provider rejection FAILED without scheduling another check", async () => {
    const f = asynchronousFixture();
    await execute(f.input);
    await makeDue();
    const batch = f.writes[0];
    f.receipts.set(batch.batchId, {
      outcomes: batch.records.map(row => ({
        operationId: row.operationId,
        status: "rejected",
        code: "invalid",
        safeReason: "Invalid",
      })),
    });
    f.input.trigger = "recovery";
    f.input.recoveryOf = "task";
    f.input.taskId = "rejected-check";
    expect(await execute(f.input)).toBe("FAILED");
    const rejectedMetrics = (await task()).metrics;
    expect(rejectedMetrics.reverseRecovery.runId).toBe((await control()).run_id);
    expect(rejectedMetrics.reverseDelivery).toMatchObject({
      upsert: { total: 1, rejected: 1 },
      records: { rejected: 2 },
    });
    expect((await admin.query("SELECT 1 FROM newjitsu.source_task WHERE status='PENDING'")).rowCount).toBe(0);
    expect((await control()).phase).toBe("batches_pending");
    expect(await taskLogs("task")).toContain(
      "2 confirmed submitted in 1 batches; 0 accepted, 0 pending processing, 2 rejected"
    );
  });
  it("retries a transient refresh failure on the original task without reopening SQL", async () => {
    const f = asynchronousFixture();
    expect(await execute(f.input)).toBe("PENDING");
    const bind = f.adapter.recovery!;
    let fail = true;
    f.adapter.recovery = state => ({
      ...bind(state),
      reconcileBatch: async batch => {
        if (fail) throw new Error("private provider response");
        return accepted(batch);
      },
    });
    await refresh(f, "failed-poll");
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("PENDING");
    expect((await task()).metrics.reverseRecovery.attempt).toBe(1);
    expect(await taskLogs("task")).not.toContain("private provider response");
    fail = false;
    await refresh(f, "next-poll");
    expect(await execute(f.input)).toBe("COMPLETE");
    expect(f.calls).not.toContain("reader");
    expect((await admin.query("SELECT count(*) FROM newjitsu.source_task")).rows[0].count).toBe("1");
  });
  it.each(["manual", "scheduled"] as const)(
    "does not create another task when %s admission encounters a pending non-detached run",
    async trigger => {
      const f = asynchronousFixture();
      expect(await execute(f.input)).toBe("PENDING");
      // Represents an interrupted extraction or migrated WAITING run.
      await admin.query("UPDATE newjitsu.reverse_sync_control SET detached=false");
      const before = await task();
      f.input = { ...f.input, trigger, taskId: "new-attempt" };
      f.calls.length = 0;
      expect(await execute(f.input)).toBe("FAILED");
      expect((await admin.query("SELECT task_id,status FROM newjitsu.source_task")).rows).toEqual([
        { task_id: "task", status: "PENDING" },
      ]);
      expect((await task()).metrics).toEqual(before.metrics);
      expect(f.calls).not.toContain("reader");
      for (const batch of f.writes) f.receipts.set(batch.batchId, accepted(batch));
      await refresh(f, "continue-original");
      expect(await execute(f.input)).toBe("COMPLETE");
      expect((await task()).status).toBe("COMPLETE");
      expect((await admin.query("SELECT count(*) FROM newjitsu.source_task")).rows[0].count).toBe("1");
    }
  );
  it.each(["FAILED", "CANCELLED"])("retains a %s detached run across failed explicit refresh", async previousStatus => {
    const f = asynchronousFixture();
    expect(await execute(f.input)).toBe("PENDING");
    const initial = await task();
    expect((await control()).detached).toBe(true);
    const bind = f.adapter.recovery!;
    f.adapter.recovery = state => ({
      ...bind(state),
      reconcileBatch: async () => {
        throw new ReverseEtlManualReconciliationError();
      },
    });
    await refresh(f, "failed-check");
    expect(await execute(f.input)).toBe("FAILED");
    expect((await task()).metrics.reverseRecovery.runId).toBe(initial.metrics.reverseRecovery.runId);
    const error = (await task()).error;
    // The controller's explicit-refresh transition: same task, new worker.
    await admin.query(
      `
      UPDATE newjitsu.source_task SET status='PENDING',metrics=jsonb_set(
        jsonb_set(metrics,'{reverseRecovery,nextCheckAt}',to_jsonb('2000-01-01T00:00:00.000Z'::text)),
        '{reverseRecovery,previousStatus}',to_jsonb($1::text)) WHERE task_id='task'`,
      [previousStatus]
    );
    f.input.taskId = "manual-failed-check";
    f.adapter.recovery = state => ({
      ...bind(state),
      reconcileBatch: async () => {
        throw new Error("temporary lookup failure");
      },
    });
    f.calls.length = 0;
    expect(await execute(f.input)).toBe(previousStatus);
    expect((await task()).status).toBe(previousStatus);
    expect((await task()).metrics.reverseWorker.active).toBe(false);
    expect((await task()).error).toBe(error);
    expect((await task()).metrics.reverseRecovery.runId).toBe(initial.metrics.reverseRecovery.runId);
    await admin.query(`
      UPDATE newjitsu.source_task SET status='PENDING',metrics=jsonb_set(
        metrics,'{reverseRecovery,nextCheckAt}',to_jsonb('2000-01-01T00:00:00.000Z'::text)) WHERE task_id='task'`);
    f.input.taskId = "manual-successful-check";
    f.adapter.recovery = bind;
    for (const batch of f.writes) f.receipts.set(batch.batchId, accepted(batch));
    expect(await execute(f.input)).toBe("COMPLETE");
    expect((await task()).error).toBeNull();
    expect(f.calls).not.toContain("reader");
    expect(f.calls).not.toContain("upsert");
    expect((await admin.query("SELECT count(*) FROM newjitsu.source_task")).rows[0].count).toBe("1");
  });
  it("does not let a queued refresh bypass suspension", async () => {
    const f = asynchronousFixture();
    expect(await execute(f.input)).toBe("PENDING");
    await refresh(f, "late-queued-check");
    await admin.query(
      "UPDATE newjitsu.source_task SET metrics=jsonb_set(metrics,'{reverseRecovery,suspended}','true')"
    );
    const before = await task();
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("FAILED");
    expect(await task()).toEqual(before);
    expect(f.calls).not.toContain("reconcile");
    expect(f.calls).not.toContain("reader");
  });
  it.each([
    "Google replacement cleanup failed or is unverified; manual reconciliation required, no automatic replay",
    "Malformed Google replacement status; manual reconciliation required",
    "Google replacement status target mismatch",
    "Google replacement cleanup receipt unavailable; manual reconciliation required, no automatic replay",
  ])("stops polling a terminal native-replacement result: %s", async reason => {
    const f = fixture();
    f.input.config.options.mode = "mirror";
    f.adapter.stream.capabilities.mirror = "native-replace";
    f.adapter.verifyMirrorBaseline = async () => "replace";
    f.setPending();
    expect(await execute(f.input)).toBe("PENDING");
    const bind = f.adapter.recovery!;
    f.adapter.recovery = state => ({
      ...bind(state),
      reconcileFinish: async () => {
        throw new Error(reason);
      },
    });
    const before = await task();
    await refresh(f, "replacement-failed");
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("FAILED");
    const failed = await task();
    expect(failed.error).toContain("Google audience replacement could not be confirmed");
    expect(failed.metrics.reverseRecovery.runId).toBe(before.metrics.reverseRecovery.runId);
    expect(failed.metrics.reverseRecovery.attempt).toBe(before.metrics.reverseRecovery.attempt);
    expect(f.calls).not.toContain("reader");
    expect(f.calls).not.toContain("finish");
  });
  it("stops automatic polling at its original deadline without losing receipts", async () => {
    const f = asynchronousFixture();
    await execute(f.input);
    await makeDue();
    await admin.query(
      `UPDATE newjitsu.source_task SET metrics=jsonb_set(jsonb_set(metrics,'{reverseRecovery,startedAt}',to_jsonb('2000-01-01T00:00:00.000Z'::text)),'{reverseRecovery,deadline}',to_jsonb('2000-01-03T00:00:00.000Z'::text))`
    );
    f.input.trigger = "recovery";
    f.input.recoveryOf = "task";
    f.input.taskId = "last-check";
    expect(await execute(f.input)).toBe("PENDING");
    expect((await task()).description).toContain("after 48 hours");
    expect((await task()).error).toBeNull();
    expect((await task()).metrics.reverseRecovery.suspended).toBe(true);
    expect((await control()).phase).toBe("batches_pending");
    expect((await durable()).batches.length).toBe(1);
    expect((await admin.query("SELECT 1 FROM newjitsu.source_task WHERE status='PENDING'")).rowCount).toBe(1);
    const expiredSchedule = (await task()).metrics.reverseRecovery;
    await admin.query(
      `UPDATE newjitsu.source_task SET metrics=jsonb_set(metrics,'{reverseRecovery,suspended}','false') WHERE task_id='task'`
    );
    f.input.taskId = "explicit-after-deadline";
    expect(await execute(f.input)).toBe("PENDING");
    expect((await task()).metrics.reverseRecovery).toEqual(expiredSchedule);
    expect(f.writes).toHaveLength(1);
    // A new extraction has its own deadline; it does not reopen this expired run.
    f.input.trigger = "manual";
    f.input.taskId = "manual-after-timeout";
    delete f.input.recoveryOf;
    expect(await execute(f.input)).toBe("PENDING");
    expect((await task()).status).toBe("PENDING");
  });
  it("still cleans up incomplete legacy finish-staged sessions through verified abort", async () => {
    const f = asynchronousFixture();
    f.adapter.stream.batchDelivery = undefined;
    const create = f.adapter.stream.createWriter;
    f.adapter.stream.createWriter = async ctx => {
      const writer = await create(ctx);
      return {
        ...writer,
        upsert: async batch => {
          await writer.upsert(batch);
          throw new Error("lost staged response");
        },
      };
    };
    expect(await execute(f.input)).toBe("FAILED");
    expect((await control()).phase).toBe("running");
    f.input.taskId = "legacy-cleanup";
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.calls).toEqual(["lease", "renew", "reconcile", "reconcileAbort", "release"]);
    expect((await control()).phase).toBe("aborted");
    expect([...new Set((await durable()).operations.map(row => row.status))].map(status => ({ status }))).toEqual([
      { status: "cancelled" },
    ]);
  });
  it("blocks direct batch acknowledgement, cleanup and finish while sealed jobs remain pending", async () => {
    const f = asynchronousFixture();
    await execute(f.input);
    const run = await openPersistence(
      db,
      {
        workspaceId: "workspace",
        syncId: "sync",
        logicalRunId: (await control()).run_id,
        taskId: "probe",
        configRevision: f.input.config.configRevision,
        targetIdentity: f.adapter.targetIdentity,
        mode: "upsert",
        extraction: "full",
      },
      f.adapter.project
    );
    await expect(run.delivery.prepareFinish(2, {})).rejects.toThrow();
    await expect(run.delivery.prepareAbort()).rejects.toThrow();
    await expect(run.delivery.acknowledge(f.writes[0].batchId, accepted(f.writes[0]), {})).rejects.toThrow();
    expect((await control()).phase).toBe("batches_pending");
  });
  it.each([false, true])("settles independent batches without re-extraction (cursor=%s)", async cursor => {
    const f = asynchronousFixture();
    if (cursor) f.input.config.model.cursor = { column: "id", type: "string" };
    f.setRows([{ id: "a" }, { id: "b" }, { id: "c" }]);
    expect(await execute(f.input)).toBe("PENDING");
    expect((await control()).phase).toBe("batches_pending");
    expect(f.calls).not.toContain("finish");
    expect((await admin.query("SELECT count(*) FROM newjitsu.source_state")).rows[0].count).toBe("0");

    await refresh(f, "poll");
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("PENDING");
    expect(f.calls).toEqual(["lease", "renew", "reconcile", "reconcile", "release"]);
    expect((await control()).phase).toBe("batches_pending");

    for (const batch of f.writes) f.receipts.set(batch.batchId, accepted(batch));
    await refresh(f, "complete");
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("COMPLETE");
    expect(f.calls).toEqual(["lease", "renew", "reconcile", "reconcile", "attach", "finish", "release"]);
    expect((await control()).phase).toBe("complete");
    const state = (await admin.query("SELECT state FROM newjitsu.source_state")).rows[0].state;
    expect(JSON.parse(state.value).value.point).toEqual({
      sourceSequence: 3,
      ...(cursor ? { cursor: { value: "c", primaryKeyValues: ["c"] } } : {}),
    });
    expect(f.writes).toHaveLength(2);
  });
  it("retains partial accepted effects after independent jobs reject rows, without finalizing", async () => {
    const f = asynchronousFixture();
    await execute(f.input);
    const batch = f.writes[0];
    f.receipts.set(batch.batchId, {
      outcomes: [
        { operationId: batch.records[0].operationId, status: "accepted" },
        { operationId: batch.records[1].operationId, status: "rejected", code: "invalid", safeReason: "Invalid" },
      ],
    });
    await refresh(f, "rejected");
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.calls).toEqual(["lease", "renew", "reconcile", "release"]);
    expect((await control()).phase).toBe("batches_pending");
    expect(String((await durable()).members.length)).toBe("1");
    expect((await task()).status).toBe("FAILED");
    expect(f.calls).not.toContain("finish");
  });
  it("rejects overlapping async identities before the conflicting provider request", async () => {
    const f = asynchronousFixture();
    f.adapter.stream.batchSize = 1;
    f.adapter.project = () => [{ identity: "shared", upsert: { id: "shared" }, remove: { id: "shared" } }];
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.writes).toHaveLength(1);
    expect((await control()).phase).toBe("running");
    expect(f.calls).not.toContain("abort");
    f.input.taskId = "still-pending";
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("PENDING");
    expect(f.calls).toEqual(["lease", "renew", "reconcile", "release"]);
    expect((await control()).phase).toBe("running");
    f.input.taskId = "do-not-duplicate-pending";
    expect(await execute(f.input)).toBe("FAILED");
    expect(await task("do-not-duplicate-pending")).toBeUndefined();
  });
  it("records lease contention without opening delivery persistence", async () => {
    const f = fixture();
    f.input.lease.acquire = async () => {
      throw new Error("Reverse sync already running");
    };
    expect(await execute(f.input)).toBe("FAILED");
    expect(await control()).toBeUndefined();
    expect(await task()).toMatchObject({ status: "FAILED" });
    const logs = (await admin.query("SELECT message FROM newjitsu.task_log WHERE task_id=$1", [f.input.taskId])).rows;
    expect(logs).toHaveLength(1);
    expect(logs[0].message).toContain("Another worker is running this sync");
    expect(f.calls).toEqual([]);
  });
  it("does not overwrite an existing task when a duplicate worker loses admission", async () => {
    const f = fixture();
    expect(await execute(f.input)).toBe("COMPLETE");
    const before = await task();
    f.input.lease.acquire = async () => {
      throw new Error("Reverse sync already running");
    };
    expect(await execute(f.input)).toBe("FAILED");
    expect(await task()).toEqual(before);
  });
  it("shares an in-flight heartbeat renewal when slow provisioning finishes", async () => {
    const f = fixture();
    let ready!: () => void;
    const heartbeatStarted = new Promise<void>(resolve => {
      ready = resolve;
    });
    let inFlight = 0;
    let maximum = 0;
    f.input.heartbeatMs = 5;
    f.input.lease.renew = async () => {
      maximum = Math.max(maximum, ++inFlight);
      ready();
      await new Promise(resolve => setTimeout(resolve, 40));
      inFlight--;
    };
    f.input.adapters = new Map([
      [
        String(f.input.config.destination.destinationType),
        async () => {
          await heartbeatStarted;
          return f.adapter;
        },
      ],
    ]);
    expect(await execute(f.input)).toBe("COMPLETE");
    expect(maximum).toBe(1);
  });
  it("runs upsert with restricted DB grants, task logs and Kubernetes admission", async () => {
    const f = fixture();
    expect(await execute(f.input)).toBe("COMPLETE");
    expect((await task()).status).toBe("COMPLETE");
    expect((await control()).phase).toBe("complete");
    expect(f.calls).toEqual([
      "lease",
      "renew",
      "create",
      "init",
      "reader",
      "source",
      "upsert",
      "finish",
      "close",
      "release",
    ]);
    expect((await admin.query("SELECT message FROM newjitsu.task_log")).rows).toContainEqual({
      message: "Reverse ETL delivery committed",
    });
  });
  it("runs mirror including empty generations", async () => {
    const f = fixture();
    f.input.config.options.mode = "mirror";
    expect(await execute(f.input)).toBe("COMPLETE");
    const logs = (await admin.query("SELECT message FROM newjitsu.task_log ORDER BY timestamp")).rows.map(
      r => r.message
    );
    expect(logs).toContain("Extracted 2 source rows into snapshot; no audience changes submitted yet");
    expect(logs).toContain(
      "Snapshot complete: 2 source rows, 2 projected audience members, 2 unique audience members, 0 duplicates collapsed. Comparing audience membership and submitting changes."
    );
    f.input.taskId = "empty";
    f.setRows([]);
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("COMPLETE");
    expect(f.calls).toContain("remove");
    expect(String((await durable()).members.length)).toBe("0");
    expect((await admin.query("SELECT message FROM newjitsu.task_log WHERE task_id='empty'")).rows).toContainEqual({
      message:
        "Snapshot complete: 0 source rows, 0 projected audience members, 0 unique audience members, 0 duplicates collapsed. Comparing audience membership and submitting changes.",
    });
  });
  it("logs deduplication of projected members rather than subtracting source rows", async () => {
    const f = fixture();
    f.input.config.options.mode = "mirror";
    f.setRows([{ id: "a" }, { id: "b" }, { id: "excluded" }]);
    f.adapter.project = (_action, row: any) =>
      row.id === "excluded"
        ? []
        : ["private-member-1", "private-member-2"].map(id => ({ identity: id, upsert: { id }, remove: { id } }));
    expect(await execute(f.input)).toBe("COMPLETE");
    const logs = (await admin.query("SELECT message FROM newjitsu.task_log")).rows;
    expect(logs).toContainEqual({
      message:
        "Snapshot complete: 3 source rows, 4 projected audience members, 2 unique audience members, 2 duplicates collapsed. Comparing audience membership and submitting changes.",
    });
    expect(JSON.stringify(logs)).not.toContain("private-member");
    expect(JSON.stringify(logs)).toContain("2 duplicates collapsed, 1 source rows excluded by projection");
    expect(f.writes.flatMap(batch => batch.records)).toHaveLength(2);
  });
  it("preserves the original mirror comparison while a later attempt accepts additions and removes old members", async () => {
    const baseline = fixture();
    baseline.input.config.options.mode = "mirror";
    expect(await execute(baseline.input)).toBe("COMPLETE");

    const f = asynchronousFixture();
    f.input.config.options.mode = "mirror";
    f.adapter.mirror!.batchDelivery = "asynchronous";
    f.input.taskId = "diff";
    f.setRows([{ id: "b" }, { id: "c" }, { id: "duplicate-c" }]);
    f.adapter.project = (_action, row: any) => {
      const id = row.id === "duplicate-c" ? "c" : row.id;
      return [{ identity: id, upsert: { id }, remove: { id } }];
    };
    expect(await execute(f.input)).toBe("PENDING");
    const initialLogs = await taskLogs("diff");
    expect(initialLogs).toContain(
      "2 baseline members; 1 new, 0 changed, 0 unchanged due for expiry refresh, 0 unconfirmed requiring refresh, 1 unchanged skipped, 1 to remove"
    );
    expect(initialLogs).toContain("3 source rows, 2 unique members, 3 projected members, 1 duplicates collapsed");
    expect(initialLogs).toContain("1 confirmed submitted in 1 batches; 0 accepted, 1 pending");
    expect(initialLogs).toContain("Removals are blocked until all additions/updates/refreshes are accepted");
    expect(initialLogs.match(/Delivery totals/g)).toHaveLength(1);
    expect(f.calls).not.toContain("remove");

    for (const batch of f.writes) f.receipts.set(batch.batchId, accepted(batch));
    await refresh(f, "diff-status-check", "diff");
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("COMPLETE");
    const resumedLogs = await taskLogs("diff");
    expect(resumedLogs).toContain("warehouse SQL is not re-read");
    expect(resumedLogs).toContain("2 baseline members; 1 new");
    expect(resumedLogs).toContain("Removals: 1 confirmed submitted in 1 batches; 1 accepted, 0 pending");
    expect(resumedLogs.match(/Delivery totals/g)).toHaveLength(2);
    expect(resumedLogs).toContain("Preparing 1 removals.");
    expect(resumedLogs).toContain("Submitted 1 removals; 1 accepted");
    expect(f.calls).toContain("remove");
    expect(f.calls).not.toContain("reader");
    expect((await durable()).members).toHaveLength(2);
  });
  it("reports the failing mirror stage and counters without exposing warehouse errors", async () => {
    const f = fixture();
    f.input.config.options.mode = "mirror";
    const reader = f.input.reader(f.input.config.warehouse);
    f.input.reader = () => ({
      ...reader,
      stream: async function* () {
        yield { row: { id: "one" }, deleted: false };
        throw new Error("private-user@example.com secret-token");
      },
    });
    expect(await execute(f.input)).toBe("FAILED");
    const failed = await task();
    expect(failed.error).toContain("during extraction (read 1 rows, saved 0)");
    expect(failed.error).toContain("Check warehouse connectivity and query timeouts");
    expect(failed.error).not.toMatch(/private-user|secret-token/);
    expect(f.writes).toHaveLength(0);
  });
  it("caps provider batches to the journal record budget", async () => {
    const f = fixture();
    const bounded = new Database(
      {
        host: container.getHost(),
        port: container.getMappedPort(5432),
        database: "runner_test",
        user: "runner_runtime",
        password: "runtime",
      },
      { ...storage, limits: { batchRecords: 2 } }
    );
    f.input.db = bounded;
    f.adapter.stream.batchSize = 1000;
    f.setRows([{ id: "a" }, { id: "b" }, { id: "c" }]);
    try {
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(f.writes.map(batch => batch.records.length)).toEqual([2, 1]);
    } finally {
      await bounded.close();
    }
  });
  it("rejects stale admission before constructing provider or reader", async () => {
    const f = fixture();
    f.input.admit = async () => ({ ...f.input.config, configRevision: "b".repeat(64) });
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.calls).toEqual(["lease", "release"]);
    expect((await task()).status).toBe("FAILED");
  });
  describe("retention changed between startup and admission", () => {
    // The object storage is fixed from the startup configuration, but retention is outside the revision hash, so an
    // admitted configuration with a different retention bucket must not run with the old storage choice.
    const admitting = (startup: { bucket: string } | undefined, admitted: { bucket: string } | undefined) => {
      const f = fixture();
      f.input.config = { ...f.input.config, ...(startup ? { retention: startup } : {}) };
      f.input.admit = async () => ({
        ...f.input.config,
        ...(admitted ? { retention: admitted } : { retention: undefined }),
      });
      return f;
    };
    it.each([
      // The only bucket a run may name is its own workspace's (the fixture's workspace is "workspace").
      ["enabled after startup", undefined, { bucket: "jitsu-retl-workspace" }],
      ["disabled after startup", { bucket: "jitsu-retl-workspace" }, undefined],
    ])(
      "refuses a run when retention was %s, before anything is constructed or sent",
      async (_name, startup, admitted) => {
        const f = admitting(startup, admitted);
        expect(await execute(f.input)).toBe("FAILED");
        expect(f.calls).toEqual(["lease", "release"]);
        expect((await task()).error).toContain("retention settings changed");
      }
    );
    it("refuses retention on a mirror run at admission, before anything is extracted or stored", async () => {
      // A mirror seals its full snapshot, with the projected provider data, before its first batch, so the per-batch guard
      // would be too late: the combination is refused up front.
      const f = fixture();
      f.input.config = {
        ...f.input.config,
        options: { ...f.input.config.options, mode: "mirror" },
        retention: { bucket: "jitsu-retl-workspace" },
      };
      expect(await execute(f.input)).toBe("FAILED");
      expect(f.calls).toEqual(["lease", "release"]);
      expect((await task()).error).toContain("snapshot of its data that cannot expire");
    });
    it("admits a run whose retention is unchanged, or absent on both sides", async () => {
      expect(await execute(admitting(undefined, undefined).input)).toBe("COMPLETE");
    });
  });
  it("does not admit missing provider bindings", async () => {
    const f = fixture();
    f.input.adapters = new Map();
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.calls).not.toContain("reader");
  });
  it("does not overwrite a task cancelled before startup", async () => {
    const f = fixture();
    await new Tasks(db, "sync", "task").start("manual");
    await new Tasks(db, "sync", "task").finish("CANCELLED", "Cancelled");
    expect(await execute(f.input)).toBe("FAILED");
    expect((await task()).status).toBe("CANCELLED");
    expect(f.calls).toEqual(["lease", "release"]);
  });
  it("reconciles pending finish without a fresh source or session", async () => {
    const f = fixture();
    f.setPending();
    expect(await execute(f.input)).toBe("PENDING");
    const logical = (await control()).run_id;
    await refresh(f, "recovery");
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("COMPLETE");
    expect(f.calls).toEqual(["lease", "renew", "reconcileFinish", "release"]);
    expect((await control()).run_id).toBe(logical);
    expect((await task()).status).toBe("COMPLETE");
  });
  it("does not restore pending after committed delivery when task completion bookkeeping fails", async () => {
    const f = fixture();
    f.setPending();
    expect(await execute(f.input)).toBe("PENDING");
    await refresh(f, "completion-check");
    const finish = vi
      .spyOn(Tasks.prototype, "finish")
      .mockRejectedValueOnce(new Error("database temporarily unavailable"));
    try {
      expect(await execute(f.input)).toBe("FAILED");
      expect((await control()).phase).toBe("complete");
      const saved = await task();
      expect(saved.status).toBe("FAILED");
      expect(saved.metrics.reverseWorker.active).toBe(false);
      expect(finish).toHaveBeenCalledTimes(2);
    } finally {
      finish.mockRestore();
    }
  });
  it("executes managed Google mirror, resumes exact wire payloads and refreshes only when due", async () => {
    const f = fixture();
    const managed = {
      id: `retl-google-${"b".repeat(64)}`,
      syncId: "sync",
      customerId: "1234567890",
      audienceId: "123",
      integrationCode: `jitsu-retl-${"b".repeat(64)}`,
      displayName: "Managed",
      membershipDays: 540,
    };
    f.input.config.destination = {
      destinationType: "google-ads",
      authorized: true,
      oauthConnectionId: "destination.destination",
      customerId: "1234567890",
      reverseManagedAudience: managed,
    };
    f.input.config.options.mode = "mirror";
    f.input.config.options.mapping = { email: "id", adUserData: "consent", adPersonalization: "consent" };
    f.input.config.options.streamOptions = {
      audienceId: "123",
      customerMatchTermsAccepted: true,
      managedAudienceId: managed.id,
    };
    const originalReader = f.input.reader;
    f.input.reader = connection => ({
      ...originalReader(connection),
      stream: async function* () {
        f.calls.push("google-source");
        yield { row: { id: "Private.Person+tag@gmail.com", consent: "GRANTED" }, deleted: false };
      },
    });
    const response = (body: unknown) => new Response(JSON.stringify(body)) as Awaited<ReturnType<typeof fetch>>;
    let submits = 0;
    const bodies: unknown[] = [];
    const wire = vi.spyOn(globalThis, "fetch").mockImplementation(async (url, init) => {
      if (String(url).endsWith("/userLists/123"))
        return response({
          name: "accountTypes/GOOGLE_ADS/accounts/1234567890/userLists/123",
          id: "123",
          displayName: managed.displayName,
          integrationCode: managed.integrationCode,
          membershipDuration: "46656000s",
          membershipStatus: "OPEN",
          accessReason: "OWNED",
          ingestedUserListInfo: {
            uploadKeyTypes: ["CONTACT_ID"],
            contactIdInfo: { dataSourceType: "DATA_SOURCE_TYPE_FIRST_PARTY" },
          },
        });
      if (String(url).includes("audienceMembers:ingest")) {
        bodies.push(JSON.parse(init!.body as string));
        return response({ requestId: `job-${++submits}` });
      }
      expect(String(url)).toContain("requestStatus:retrieve?requestId=job-");
      return response({
        requestStatusPerDestination: [
          {
            destination: {
              operatingAccount: { accountType: "GOOGLE_ADS", accountId: "1234567890" },
              productDestinationId: "123",
            },
            requestStatus: "SUCCESS",
            audienceMembersIngestionStatus: { userDataIngestionStatus: { recordCount: "1" } },
          },
        ],
      });
    });
    try {
      f.input.adapters = createAdapterRegistry(async () => "access-token");
      expect(await execute(f.input)).toBe("PENDING");
      expect(submits).toBe(1);
      await refresh(f, "resume");
      f.input.adapters = createAdapterRegistry(async () => "access-token");
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(f.calls.filter(c => c === "google-source")).toHaveLength(1);
      f.input.taskId = "fresh";
      f.input.trigger = "manual";
      delete f.input.recoveryOf;
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(submits).toBe(1);
      expect(await taskLogs("fresh")).toContain("1 unchanged skipped");
      expect(await taskLogs("fresh")).toContain("No audience changes or expiry refreshes needed");
      vi.useFakeTimers({ toFake: ["Date"] });
      vi.setSystemTime(Date.now() + 31 * 86400_000);
      f.input.taskId = "refresh";
      expect(await execute(f.input)).toBe("PENDING");
      expect(submits).toBe(2);
      expect(await taskLogs("refresh")).toContain(
        "1 unchanged due for expiry refresh, 0 unconfirmed requiring refresh, 0 unchanged skipped"
      );
      expect(bodies[1]).toEqual(bodies[0]);
      expect(JSON.stringify(bodies)).not.toContain("Private.Person");
      await refresh(f, "refresh-resume", "refresh");
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(await taskLogs("refresh")).toContain(
        "1 unchanged due for expiry refresh, 0 unconfirmed requiring refresh, 0 unchanged skipped"
      );
      expect(await taskLogs("refresh")).toContain("1 confirmed submitted in 1 batches; 1 accepted, 0 pending");
    } finally {
      wire.mockRestore();
    }
  });

  it("replaces an existing Google audience across upload and cleanup status checks without replay or re-extraction", async () => {
    const f = fixture();
    f.input.config.destination = {
      destinationType: "google-ads",
      authorized: true,
      oauthConnectionId: "destination.destination",
      customerId: "1234567890",
    };
    f.input.config.options.mode = "mirror";
    f.input.config.options.mapping = { email: "id" };
    f.input.config.options.streamOptions = {
      audienceId: "123",
      customerMatchTermsAccepted: true,
      mirrorStrategy: "full-replace",
      exclusiveManagementConfirmed: true,
    };
    f.setRows([{ id: "one@example.com" }, { id: "two@example.com" }]);
    f.input.adapters = createAdapterRegistry(async () => "access-token");
    const cutoff = "2026-09-20T07:00:00.000Z";
    let uploads = 0,
      cleanups = 0,
      uploadsAccepted = false,
      cleanupAccepted = false;
    const response = (body: unknown) =>
      new Response(JSON.stringify(body), { headers: { Date: new Date(cutoff).toUTCString() } }) as Awaited<
        ReturnType<typeof fetch>
      >;
    const wire = vi.spyOn(globalThis, "fetch").mockImplementation(async (url, init) => {
      if (String(url).endsWith("/userLists/123"))
        return response({
          id: "123",
          name: "accountTypes/GOOGLE_ADS/accounts/1234567890/userLists/123",
          displayName: "Existing",
          membershipDuration: "46656000s",
          membershipStatus: "OPEN",
          accessReason: "OWNED",
          ingestedUserListInfo: {
            uploadKeyTypes: ["CONTACT_ID"],
            contactIdInfo: { dataSourceType: "DATA_SOURCE_TYPE_FIRST_PARTY" },
          },
        });
      if (String(url).endsWith("audienceMembers:ingest")) {
        expect(JSON.parse((await control()).store.toString()).value.googleAudienceReplacement.cutoff).toBe(cutoff);
        expect(JSON.parse(init!.body as string).audienceMembers).toHaveLength(2);
        return response({ requestId: `upload-${++uploads}` });
      }
      if (String(url).endsWith("audienceMembers:removeAll")) {
        expect(uploadsAccepted).toBe(true);
        expect((await control()).phase).toBe("finish_prepared");
        expect(JSON.parse(init!.body as string)).toEqual({
          destinations: [
            { operatingAccount: { accountType: "GOOGLE_ADS", accountId: "1234567890" }, productDestinationId: "123" },
          ],
          removeAsOfTime: cutoff,
        });
        return response({ requestId: `cleanup-${++cleanups}` });
      }
      expect(String(url)).toContain("requestStatus:retrieve?requestId=");
      expect(init?.method).toBe("GET");
      const cleanup = String(url).includes("requestId=cleanup-");
      return response({
        requestStatusPerDestination: [
          {
            destination: {
              operatingAccount: { accountType: "GOOGLE_ADS", accountId: "1234567890" },
              productDestinationId: "123",
            },
            requestStatus: (cleanup ? cleanupAccepted : uploadsAccepted) ? "SUCCESS" : "PROCESSING",
            ...(cleanup
              ? { removeAllAudienceMembersStatus: {} }
              : { audienceMembersIngestionStatus: { userDataIngestionStatus: { recordCount: "2" } } }),
          },
        ],
      });
    });
    try {
      expect(await execute(f.input)).toBe("PENDING");
      expect(uploads).toBe(1);
      expect(cleanups).toBe(0);
      expect(await taskLogs("task")).toContain("All 2 members will be uploaded");
      expect(await taskLogs("task")).not.toContain("unchanged skipped");
      await makeDue();
      f.input = { ...f.input, taskId: "upload-check", trigger: "recovery", recoveryOf: "task" };
      expect(await execute(f.input)).toBe("PENDING");
      uploadsAccepted = true;
      await makeDue();
      f.input = { ...f.input, taskId: "start-cleanup", recoveryOf: "task" };
      expect(await execute(f.input)).toBe("PENDING");
      expect(cleanups).toBe(1);
      expect(await taskLogs("task")).toContain("Google is processing full-audience cleanup");
      expect(await taskLogs("task")).toContain("Full-audience cleanup: pending");
      expect(await taskLogs("task")).not.toContain("Removals: 0");
      await makeDue();
      f.input = { ...f.input, taskId: "cleanup-check", recoveryOf: "task" };
      expect(await execute(f.input)).toBe("PENDING");
      cleanupAccepted = true;
      await makeDue();
      f.input = { ...f.input, taskId: "complete-cleanup", recoveryOf: "task" };
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(await taskLogs("task")).toContain("Full-audience cleanup: accepted");
      expect((await taskLogs("task")).match(/Delivery totals/g)).toHaveLength(5);
      expect(uploads).toBe(1);
      expect(cleanups).toBe(1);
      expect(f.calls.filter(c => c === "source")).toHaveLength(1);
      expect((await control()).phase).toBe("complete");
    } finally {
      wire.mockRestore();
    }
  });

  it("rejects Google mirroring without an exact sync/account/audience binding", () => {
    const cfg = config();
    cfg.destination = {
      destinationType: "google-ads",
      authorized: true,
      oauthConnectionId: "destination.destination",
      customerId: "1234567890",
    };
    cfg.options.mode = "mirror";
    cfg.options.streamOptions = { audienceId: "123", customerMatchTermsAccepted: true };
    const create = createAdapterRegistry(async () => "token").get("google-ads")!;
    expect(() => create(cfg)).toThrow("cannot be mirrored");
    cfg.options.streamOptions.managedAudienceId = `retl-google-${"a".repeat(64)}`;
    cfg.destination.reverseManagedAudience = {
      id: cfg.options.streamOptions.managedAudienceId,
      syncId: "other",
      customerId: "1234567890",
      audienceId: "123",
      integrationCode: `jitsu-retl-${"a".repeat(64)}`,
      displayName: "Managed",
      membershipDays: 540,
    };
    expect(() => create(cfg)).toThrow("binding mismatch");
  });

  it("recovers a Google request after restart using the durable normalized receipt, without source replay", async () => {
    const f = fixture();
    f.input.config.destination = {
      destinationType: "google-ads",
      authorized: true,
      oauthConnectionId: "destination.destination",
      customerId: "1234567890",
    };
    f.input.config.options.mapping = { email: "id", adUserData: "consent", adPersonalization: "consent" };
    f.input.config.options.streamOptions = { audienceId: "123", customerMatchTermsAccepted: true };
    const originalReader = f.input.reader;
    f.input.reader = connection => ({
      ...originalReader(connection),
      stream: async function* () {
        f.calls.push("google-source");
        yield { row: { id: "Private.Person+tag@gmail.com", consent: "GRANTED" }, deleted: false };
      },
    });
    let polls = 0;
    // Node's Response constructor and the workspace's ambient fetch declaration
    // differ only in json()'s generic signature; runtime responses are native.
    const response = (body: unknown) => new Response(JSON.stringify(body)) as Awaited<ReturnType<typeof fetch>>;
    const wire = vi.spyOn(globalThis, "fetch").mockImplementation(async (url, init) => {
      expect(init?.headers).toEqual({ Authorization: "Bearer access-token", "Content-Type": "application/json" });
      if (String(url).includes("audienceMembers:ingest")) return response({ requestId: "durable-job" });
      expect(String(url)).toContain("requestStatus:retrieve?requestId=durable-job");
      polls++;
      return response({
        requestStatusPerDestination: [
          {
            destination: {
              operatingAccount: { accountType: "GOOGLE_ADS", accountId: "1234567890" },
              productDestinationId: "123",
            },
            requestStatus: polls === 1 ? "PROCESSING" : "SUCCESS",
            ...(polls === 1
              ? {}
              : { audienceMembersIngestionStatus: { userDataIngestionStatus: { recordCount: "1" } } }),
          },
        ],
      });
    });
    try {
      f.input.adapters = createAdapterRegistry(async () => "access-token");
      expect(await execute(f.input)).toBe("PENDING");
      expect((await control()).phase).toBe("batches_pending");
      const logicalRun = (await control()).run_id;
      const saved = (await durable()).batches;
      expect(saved).toHaveLength(1);
      const serialized = JSON.stringify(saved);
      expect(serialized).toContain("durable-job");
      expect(serialized).not.toContain("Private.Person");
      expect(serialized).not.toContain("access-token");
      // New registry/client instances simulate a different worker process.
      f.input.adapters = createAdapterRegistry(async () => "access-token");
      await refresh(f, "google-pending");
      expect(await execute(f.input)).toBe("PENDING");
      f.input.adapters = createAdapterRegistry(async () => "access-token");
      await refresh(f, "google-accepted");
      expect(await execute(f.input)).toBe("COMPLETE");
      expect((await control()).run_id).toBe(logicalRun);
      expect((await control()).checkpoint_sequence).toBe("1");
      expect(f.calls.filter(call => call === "google-source")).toHaveLength(1);
      expect(wire.mock.calls.filter(([, init]) => init?.method === "POST")).toHaveLength(1);
      expect(String((await durable()).members.length)).toBe("1");
    } finally {
      wire.mockRestore();
    }
  });
  it("preserves cursor when recovered finish commits the exact last manifest", async () => {
    const f = fixture();
    f.input.config.model.cursor = { column: "id", type: "string" };
    f.setPending();
    expect(await execute(f.input)).toBe("PENDING");
    await refresh(f, "recovery");
    expect(await execute(f.input)).toBe("COMPLETE");
    expect((await control()).checkpoint_sequence).toBe("2");
  });
  it("recovers an empty cursor run from its plain JSON checkpoint without keys", async () => {
    const f = fixture();
    f.input.config.model.cursor = { column: "id", type: "string" };
    f.setRows([{ id: "a\u0000b" }]);
    expect(await execute(f.input)).toBe("COMPLETE");
    const saved = (await admin.query("SELECT state FROM newjitsu.source_state")).rows[0].state;
    expect(saved.version).toBe(2);
    const point = JSON.parse(saved.value).value.point;
    expect(point.cursor).toEqual({ value: "a\u0000b", primaryKeyValues: ["a\u0000b"] });

    f.setRows([]);
    f.setPending();
    f.input.taskId = "empty";
    expect(await execute(f.input)).toBe("PENDING");
    await refresh(f, "recovery", "empty");
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("COMPLETE");
    expect(f.calls).toEqual(["lease", "renew", "reconcileFinish", "release"]);
    const recovered = (await admin.query("SELECT state FROM newjitsu.source_state")).rows[0].state;
    expect(JSON.parse(recovered.value).value.point).toEqual(point);
  });
  it("blocks unknown initialization without proof, then resets with explicit reconciliation", async () => {
    const f = fixture();
    f.setFailInit();
    expect(await execute(f.input)).toBe("FAILED");
    const recovery = f.adapter.recovery;
    f.adapter.recovery = undefined;
    f.input.taskId = "blocked";
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.calls).toEqual(["lease", "renew", "release"]);
    expect((await control()).phase).toBe("init_prepared");
    f.adapter.recovery = recovery;
    f.input.taskId = "reset";
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.calls).toContain("reconcileInit");
    expect((await control()).phase).toBe("new");
  });
  it("reconciles an ambiguous upsert then cleans up before any fresh extraction", async () => {
    const f = fixture();
    f.setFailBatch();
    expect(await execute(f.input)).toBe("FAILED");
    const failedLogs = await taskLogs("task");
    expect(failedLogs.match(/Delivery totals/g)).toHaveLength(1);
    expect(failedLogs).toContain("Confirmation missing for 2 additions/upserts");
    expect(failedLogs).toContain("2 prepared/unconfirmed");
    expect(failedLogs).toContain("may have reached the destination");
    f.input.taskId = "recovery";
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.calls).toEqual(["lease", "renew", "reconcile", "reconcileAbort", "release"]);
    expect((await control()).phase).toBe("aborted");
    expect(String((await durable()).members.length)).toBe("2");
  });
  it("does not release ownership until an aborted in-flight callback settles", async () => {
    const f = fixture();
    f.input.heartbeatMs = 5;
    let unblock!: () => void;
    let entered!: () => void;
    const enteredPromise = new Promise<void>(resolve => {
      entered = resolve;
    });
    const barrier = new Promise<void>(resolve => {
      unblock = resolve;
    });
    const create = f.adapter.stream.createWriter;
    f.adapter.stream.createWriter = async ctx => {
      const writer = await create(ctx);
      return {
        ...writer,
        upsert: async batch => {
          entered();
          await barrier;
          return accepted(batch);
        },
      };
    };
    const running = execute(f.input);
    await enteredPromise;
    f.input.controller.abort();
    expect(f.calls).not.toContain("release");
    expect((await task()).status).toBe("RUNNING");
    unblock();
    expect(await running).toBe("CANCELLED");
    expect(f.calls.at(-1)).toBe("release");
  });
  it("aborts a live run when Kubernetes renewal fails", async () => {
    const f = fixture();
    f.input.heartbeatMs = 5;
    let constructing = false;
    f.input.lease.renew = async () => {
      if (constructing) throw new Error("lease lost");
    };
    const create = f.adapter.stream.createWriter;
    f.adapter.stream.createWriter = async ctx => {
      constructing = true;
      await new Promise<void>(resolve => ctx.signal.addEventListener("abort", () => resolve(), { once: true }));
      return create(ctx);
    };
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.calls).not.toContain("init");
    expect((await control()).phase).toBe("init_prepared");
    expect((await task()).error).toContain("ownership");
  });
});

describe("rejection reporting", () => {
  const rejectEverything = (
    f: ReturnType<typeof fixture>,
    code: string,
    details?: Array<{ reason: string; field?: string }>
  ) => {
    const create = f.adapter.stream.createWriter;
    f.adapter.stream.createWriter = async ctx => {
      const writer = await create(ctx);
      writer.upsert = async batch => ({
        outcomes: batch.records.map(row => ({
          operationId: row.operationId,
          status: "rejected" as const,
          code,
          safeReason: "HTTP 422 with private-token in the body",
          ...(details ? { details } : {}),
        })),
      });
      return writer;
    };
  };
  it("persists structured provider details in both the task error and ERROR log", async () => {
    const f = fixture();
    rejectEverything(f, "GOOGLE_HTTP_400_INVALID_ARGUMENT", [
      { reason: "INVALID_CURRENCY_CODE", field: "events[0].currency" },
    ]);
    expect(await execute(f.input)).toBe("FAILED");
    const error = (await task()).error as string;
    expect(error).toContain("events[0].currency: INVALID_CURRENCY_CODE");
    const logs = await admin.query("SELECT message FROM newjitsu.task_log WHERE task_id=$1 AND level='ERROR'", [
      f.input.taskId,
    ]);
    expect(logs.rows.some(row => row.message === error)).toBe(true);
    expect(await taskLogs("task")).not.toContain("private-token");
  });
  it("shows the rejection code in the task error and never the provider's text", async () => {
    const f = fixture();
    rejectEverything(f, "http_422");
    expect(await execute(f.input)).toBe("FAILED");
    const error = (await task()).error as string;
    expect(error).toContain("The destination rejected a row, so the sync stopped.");
    expect(error).toContain("Reason code: http_422.");
    expect(error).not.toContain("identifier mappings");
    expect(error).not.toContain("private-token");
    expect(await taskLogs("task")).not.toContain("private-token");
  });
  it("shows the rejection code when recovery of an unresolved batch finds a rejected row", async () => {
    const f = fixture();
    f.setFailBatch(); // the response was lost: the batch is left unresolved
    expect(await execute(f.input)).toBe("FAILED");
    const recovery = f.adapter.recovery!;
    // reconciliation learns the destination's answer: the row was rejected
    f.adapter.recovery = providerState => ({
      ...recovery(providerState),
      reconcileBatch: async batch => ({
        outcomes: batch.records.map(row => ({
          operationId: row.operationId,
          status: "rejected" as const,
          code: "http_422",
          safeReason: "HTTP 422 with private-token in the body",
        })),
      }),
    });
    f.input = { ...f.input, taskId: "recovery" };
    expect(await execute(f.input)).toBe("FAILED");
    const error = (await task("recovery")).error as string;
    expect(error).toContain("The destination rejected a row, so the sync stopped.");
    expect(error).toContain("Reason code: http_422.");
    expect(error).not.toContain("private-token");
  });
  it("shows the rejection code when a mirror run's recovery finds a rejected row", async () => {
    const f = fixture();
    f.input.config.options = { ...f.input.config.options, mode: "mirror" };
    f.setFailBatch();
    expect(await execute(f.input)).toBe("FAILED");
    const recovery = f.adapter.recovery!;
    f.adapter.recovery = providerState => ({
      ...recovery(providerState),
      reconcileBatch: async batch => ({
        outcomes: batch.records.map(row => ({
          operationId: row.operationId,
          status: "rejected" as const,
          code: "http_422",
          safeReason: "HTTP 422 with private-token in the body",
        })),
      }),
    });
    f.input = { ...f.input, taskId: "recovery" };
    expect(await execute(f.input)).toBe("FAILED");
    const error = (await task("recovery")).error as string;
    expect(error).toContain("The destination rejected a row, so the sync stopped.");
    expect(error).toContain("Reason code: http_422.");
    expect(error).not.toContain("private-token");
  });
  it("the run that cleans up after a rejected batch was acknowledged still shows its reason code", async () => {
    const f = fixture();
    f.setFailBatch(); // unresolved batch
    expect(await execute(f.input)).toBe("FAILED");
    const recovery = f.adapter.recovery!;
    f.adapter.recovery = providerState => ({
      ...recovery(providerState),
      reconcileBatch: async batch => ({
        outcomes: batch.records.map(row => ({
          operationId: row.operationId,
          status: "rejected" as const,
          code: "http_422",
          safeReason: "HTTP 422 with private-token in the body",
          details: [{ reason: "INVALID_CURRENCY_CODE", field: "events[0].currency" }],
        })),
      }),
    });
    f.input = { ...f.input, taskId: "recovery-1" };
    expect(await execute(f.input)).toBe("FAILED"); // the rejection is acknowledged durably, with its code shown
    // The next run finds the rejection already saved: it cleans up and restarts, and must not lose the code. Raising here
    // again would make recovery fail every time, so the code is added to the cleanup message instead.
    f.input = { ...f.input, taskId: "recovery-2" };
    f.calls.length = 0;
    expect(await execute(f.input)).toBe("FAILED");
    expect(f.calls).toContain("reconcileAbort");
    const error = (await task("recovery-2")).error as string;
    expect(error).toContain("Recovery completed cleanup; next run will restart extraction");
    expect(error).toContain("Reason code: http_422.");
    expect(error).toContain("events[0].currency: INVALID_CURRENCY_CODE");
    expect(error).not.toContain("private-token");
    // and the sync is not stuck: the run after that completes (the destination is healthy again)
    const create = f.adapter.stream.createWriter;
    f.adapter.stream.createWriter = async ctx => {
      const writer = await create(ctx);
      writer.upsert = async batch => accepted(batch);
      return writer;
    };
    f.input = { ...f.input, taskId: "recovery-3" };
    expect(await execute(f.input)).toBe("COMPLETE");
  });
  it("shows the rejection code for a mirror run too", async () => {
    const f = fixture();
    rejectEverything(f, "http_422");
    f.input.config.options = { ...f.input.config.options, mode: "mirror" };
    expect(await execute(f.input)).toBe("FAILED");
    const error = (await task()).error as string;
    expect(error).toContain("The destination rejected a row, so the sync stopped.");
    expect(error).toContain("Reason code: http_422.");
    expect(error).not.toContain("private-token");
  });
  it("drops a malformed code instead of showing it", async () => {
    const f = fixture();
    rejectEverything(f, "Bearer private-token");
    expect(await execute(f.input)).toBe("FAILED");
    const error = (await task()).error as string;
    expect(error).toContain("The destination rejected a row, so the sync stopped.");
    expect(error).not.toContain("Reason code");
    expect(error).not.toContain("private-token");
  });
  it("describes delivery counts without Google-specific wording", async () => {
    const f = fixture();
    expect(await execute(f.input)).toBe("COMPLETE");
    const logs = await taskLogs("task");
    expect(logs).toContain("Delivery counts describe records the destination's API accepted");
    expect(logs).not.toContain("Google");
  });
});

describe("webhook destination through the runner", () => {
  let server: http.Server;
  let received: any[];
  let status: number;
  beforeEach(async () => {
    received = [];
    status = 200;
    server = http.createServer((req, res) => {
      const chunks: Buffer[] = [];
      req.on("data", chunk => chunks.push(chunk));
      req.on("end", () => {
        received.push({ headers: req.headers, body: JSON.parse(Buffer.concat(chunks).toString()) });
        res.statusCode = status;
        res.end();
      });
    });
    await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
  });
  afterEach(() => new Promise<void>(resolve => server.close(() => resolve())));

  function webhookFixture() {
    const f = fixture();
    const url = `http://127.0.0.1:${(server.address() as AddressInfo).port}/hook`;
    f.input.config.destination = { destinationType: "webhook", url, method: "POST" };
    f.input.config.options = {
      ...f.input.config.options,
      stream: "rows",
      mode: "upsert",
      mapping: { id: "id" },
      streamOptions: { deliveryAttested: true, allowInsecureHttp: true },
    };
    f.setRows([{ id: "a" }, { id: "b" }, { id: "c" }]);
    // The test server is on loopback, which the real guard refuses; everything else is the production delivery path.
    const deps = {
      ...defaultDeliveryDeps,
      send: createGuardedRequest({ isBlocked: () => false }),
      sleep: async () => {},
    };
    f.input.adapters = new Map([["webhook", cfg => createWebhookRuntime(cfg as any, undefined, deps)]]);
    return f;
  }

  it("delivers every row once, in the documented envelope", async () => {
    const f = webhookFixture();
    expect(await execute(f.input)).toBe("COMPLETE");
    const records = received.flatMap(r => r.body.records);
    expect(records.map((r: any) => r.data.id).sort()).toEqual(["a", "b", "c"]);
    expect(received[0].body).toMatchObject({ syncId: "sync" });
    expect(records.every((r: any) => r.operation === "upsert" && r.idempotencyKey)).toBe(true);
  });

  it("a long URL (for example one that carries a token in its query string) does not stop the run from starting", async () => {
    const f = webhookFixture();
    const long = `${f.input.config.destination.url}?token=${"t".repeat(700)}`;
    f.input.config.destination = { ...f.input.config.destination, url: long };
    expect(await execute(f.input)).toBe("COMPLETE");
    expect(received.flatMap(r => r.body.records).length).toBe(3);
  });

  it("stops at a rejected record with its reason code, then re-sends the same keys on the next run", async () => {
    const f = webhookFixture();
    status = 400;
    expect(await execute(f.input)).toBe("FAILED");
    expect((await task()).error).toContain("Reason code: http_400");
    const firstKeys = received.flatMap(r => r.body.records.map((x: any) => x.idempotencyKey)).sort();
    received.length = 0;
    status = 200;
    f.input = { ...f.input, taskId: "task-2" };
    expect(await execute(f.input)).toBe("COMPLETE");
    const secondKeys = received.flatMap(r => r.body.records.map((x: any) => x.idempotencyKey)).sort();
    expect([...new Set(secondKeys)]).toEqual([...new Set(firstKeys)]);
  });

  it("an unreachable endpoint fails the run with a readable code and no thrown crash", async () => {
    const f = webhookFixture();
    await new Promise<void>(resolve => server.close(() => resolve()));
    server = http.createServer();
    expect(await execute(f.input)).toBe("FAILED");
    expect((await task()).error).toMatch(/Reason code: (unconfirmed|connection_error)/);
  });
  describe("Postgres column types arrive as plain JSON", () => {
    // [name, column type, value expression, expected JSON value as received by the endpoint]
    // Expectations were reviewed one by one against the plan's type table (H11): types that fail a run without
    // conversion (float NaN/Infinity, interval, bytea) must be converted. The warehouse reader also
    // preserves exact numeric arrays; JSON payloads intentionally use native JS values.
    const cases: Array<[string, string, string, unknown]> = [
      ["int4", "int4", "42", 42],
      ["int8", "int8", "123456789012345678", "123456789012345678"],
      ["numeric", "numeric", "1.50", "1.50"],
      ["float8", "float8", "1.25", 1.25],
      ["float4", "float4", "1.5", 1.5],
      ["float8 NaN", "float8", "'NaN'", "NaN"],
      ["float8 Infinity", "float8", "'Infinity'", "Infinity"],
      ["float8 -Infinity", "float8", "'-Infinity'", "-Infinity"],
      ["float4 NaN", "float4", "'NaN'", "NaN"],
      ["numeric NaN", "numeric", "'NaN'", "NaN"],
      ["bool", "bool", "true", true],
      ["text", "text", "'héllo'", "héllo"],
      ["char(3)", "char(3)", "'a'", "a  "],
      ["null", "text", "NULL", null],
      ["date", "date", "'2026-01-01'", "2026-01-01"],
      ["timestamp", "timestamp", "'2026-01-01 12:00:00.123456'", "2026-01-01 12:00:00.123456"],
      ["timestamptz", "timestamptz", "'2026-01-01 12:00:00+00'", "2026-01-01 12:00:00+00"],
      ["timestamptz infinity", "timestamptz", "'infinity'", "infinity"],
      ["time", "time", "'12:34:56'", "12:34:56"],
      ["interval", "interval", "'1 day 02:03:04'", "P0Y0M1DT2H3M4S"],
      ["uuid", "uuid", "'11111111-2222-3333-4444-555555555555'", "11111111-2222-3333-4444-555555555555"],
      ["json", "json", `'{"a":[1,2]}'`, { a: [1, 2] }],
      ["jsonb", "jsonb", `'{"a":[1,2]}'`, { a: [1, 2] }],
      ["json JS number", "json", `'{"id":9007199254740993}'`, { id: 9007199254740992 }],
      ["jsonb JS number", "jsonb", `'{"id":9007199254740993}'`, { id: 9007199254740992 }],
      ["int[]", "int[]", "ARRAY[1,2,3]", [1, 2, 3]],
      ["text[]", "text[]", "ARRAY['a','b']", ["a", "b"]],
      ["int8[]", "int8[]", "ARRAY[9007199254740993]", ["9007199254740993"]],
      ["numeric[]", "numeric[]", "ARRAY[1.5,2.5]", ["1.5", "2.5"]],
      ["bytea", "bytea", "'\\xdeadbeef'", "3q2+7w=="],
      ["bytea[]", "bytea[]", "ARRAY['\\xde'::bytea]", ["3g=="]],
      ["interval[]", "interval[]", "ARRAY['1 day'::interval]", ["P0Y0M1DT0H0M0S"]],
      ["float8[] with NaN", "float8[]", "ARRAY['NaN'::float8, 1]", ["NaN", 1]],
      ["jsonb[]", "jsonb[]", `ARRAY['{"a":1}'::jsonb]`, [{ a: 1 }]],
      [
        "2-D int[]",
        "int[]",
        "ARRAY[[1,2],[3,4]]",
        [
          [1, 2],
          [3, 4],
        ],
      ],
      ["inet", "inet", "'10.0.0.1'", "10.0.0.1"],
      ["cidr", "cidr", "'10.0.0.0/8'", "10.0.0.0/8"],
      ["macaddr", "macaddr", "'08:00:2b:01:02:03'", "08:00:2b:01:02:03"],
      ["money", "money", "'$1.50'", "$1.50"],
      ["bit(3)", "bit(3)", "B'101'", "101"],
      ["xml", "xml", "'<a/>'", "<a/>"],
      ["tsvector", "tsvector", "'a:1 b:2'", "'a':1 'b':2"],
      ["int4range", "int4range", "'[1,5)'", "[1,5)"],
      ["point", "point", "'(1,2)'", { x: 1, y: 2 }],
      ["box", "box", "'((1,1),(2,2))'", "(2,2),(1,1)"],
    ];
    let n = 0;
    it.each(cases)("%s", async (name, type, expr, expected) => {
      const table = `webhook_type_${++n}`;
      await admin.query(
        `CREATE TABLE ${table} (id int PRIMARY KEY, v ${type}); INSERT INTO ${table} VALUES (1, ${expr})`
      );
      const f = webhookFixture();
      f.input.config.warehouse = {
        destinationType: "postgres",
        host: container.getHost(),
        port: container.getMappedPort(5432),
        database: "runner_test",
        username: "postgres",
        password: "test",
        sslMode: "disable",
      };
      f.input.reader = createWarehouseReader as any;
      f.input.config.model = { ...f.input.config.model, query: `SELECT id, v FROM ${table}`, primaryKey: ["id"] };
      f.input.config.options = { ...f.input.config.options, mapping: { id: "id", v: "v" } };
      expect(await execute(f.input)).toBe("COMPLETE");
      const record = received.flatMap(r => r.body.records)[0];
      expect(record.data.id).toBe(1);
      // Whatever arrives must survive a JSON round trip unchanged, i.e. be plain JSON.
      expect(JSON.parse(JSON.stringify(record.data))).toEqual(record.data);
      expect(record.data.v).toEqual(expected);
    });
  });
  describe("ClickHouse column types arrive as plain JSON", () => {
    let ch: StartedTestContainer;
    const chRun = async (query: string) => {
      const response = await fetch(`http://${ch.getHost()}:${ch.getMappedPort(8123)}/?user=default&password=pw`, {
        method: "POST",
        body: query,
      });
      if (!response.ok) throw new Error(await response.text());
    };
    beforeAll(async () => {
      ch = await new GenericContainer("clickhouse/clickhouse-server:25.4-alpine")
        .withExposedPorts(8123)
        .withEnvironment({ CLICKHOUSE_DB: "default", CLICKHOUSE_USER: "default", CLICKHOUSE_PASSWORD: "pw" })
        .withWaitStrategy(Wait.forHttp("/ping", 8123).forStatusCode(200))
        .withStartupTimeout(120_000)
        .start();
    }, 150_000);
    afterAll(async () => {
      await ch?.stop();
    });
    // [name, column type, value expression, expected JSON value as received by the endpoint]
    const cases: Array<[string, string, string, unknown]> = [
      ["Int64", "Int64", "9223372036854775807", "9223372036854775807"],
      ["UInt64", "UInt64", "18446744073709551615", "18446744073709551615"],
      ["Int128", "Int128", "170141183460469231731687303715884105727", "170141183460469231731687303715884105727"],
      [
        "Int256",
        "Int256",
        "57896044618658097711785492504343953926634992332820282019728792003956564819967",
        "57896044618658097711785492504343953926634992332820282019728792003956564819967",
      ],
      ["Decimal(18,4)", "Decimal(18,4)", "1.2345", "1.2345"],
      ["Decimal256(10)", "Decimal256(10)", "123.456", "123.456"],
      ["Float64", "Float64", "1.5", 1.5],
      ["Float64 nan", "Float64", "nan", "nan"],
      ["Float64 inf", "Float64", "inf", "inf"],
      ["Date", "Date", "'2026-01-01'", "2026-01-01"],
      ["Date32", "Date32", "'2026-01-01'", "2026-01-01"],
      ["DateTime", "DateTime('UTC')", "'2026-01-01 12:00:00'", "2026-01-01T12:00:00Z"],
      ["DateTime64(6)", "DateTime64(6,'UTC')", "'2026-01-01 12:00:00.123456'", "2026-01-01T12:00:00.123456Z"],
      ["String", "String", "'héllo'", "héllo"],
      ["FixedString(3)", "FixedString(3)", "'abc'", "abc"],
      ["UUID", "UUID", "'11111111-2222-3333-4444-555555555555'", "11111111-2222-3333-4444-555555555555"],
      ["Bool", "Bool", "true", true],
      ["Enum8", "Enum8('a'=1,'b'=2)", "'a'", "a"],
      ["Enum16", "Enum16('x'=1000,'y'=2)", "'x'", "x"],
      ["LowCardinality(String)", "LowCardinality(String)", "'lc'", "lc"],
      ["Nullable(String) NULL", "Nullable(String)", "NULL", null],
      ["Array(Int64)", "Array(Int64)", "[1,2,9223372036854775807]", ["1", "2", "9223372036854775807"]],
      ["Array(String)", "Array(String)", "['x']", ["x"]],
      ["Array(Nullable(Int64))", "Array(Nullable(Int64))", "[1,NULL]", ["1", null]],
      ["Tuple(Int64,String)", "Tuple(Int64, String)", "(7,'t')", ["7", "t"]],
      ["named Tuple", "Tuple(a Int64, b String)", "(1,'t')", { a: "1", b: "t" }],
      ["Map(String,Int64)", "Map(String, Int64)", "map('k', 5)", { k: "5" }],
      ["IPv4", "IPv4", "'10.0.0.1'", "10.0.0.1"],
      ["IPv6", "IPv6", "'::1'", "::1"],
      ["Variant", "Variant(Int64, String)", "5", "5"],
    ];
    let n = 0;
    it.each(cases)("%s", async (name, type, expr, expected) => {
      const table = `webhook_type_${++n}`;
      await chRun(`CREATE TABLE ${table} (id UInt32, v ${type}) ENGINE = MergeTree ORDER BY id`);
      await chRun(`INSERT INTO ${table} VALUES (1, ${expr})`);
      const f = webhookFixture();
      f.input.config.warehouse = {
        destinationType: "clickhouse",
        protocol: "http",
        hosts: [`${ch.getHost()}:${ch.getMappedPort(8123)}`],
        database: "default",
        username: "default",
        password: "pw",
      };
      f.input.reader = createWarehouseReader as any;
      f.input.config.model = { ...f.input.config.model, query: `SELECT id, v FROM ${table}`, primaryKey: ["id"] };
      f.input.config.options = { ...f.input.config.options, mapping: { id: "id", v: "v" } };
      expect(await execute(f.input)).toBe("COMPLETE");
      const record = received.flatMap(r => r.body.records)[0];
      expect(JSON.parse(JSON.stringify(record.data))).toEqual(record.data);
      expect(record.data.v).toEqual(expected);
    });
    it("Nested columns arrive as arrays when aliased in the model SQL", async () => {
      await chRun(
        `CREATE TABLE webhook_nested (id UInt32, n Nested(a Int64, b String)) ENGINE = MergeTree ORDER BY id`
      );
      await chRun("INSERT INTO webhook_nested (id, `n.a`, `n.b`) VALUES (1, [1,2], ['p','q'])");
      const f = webhookFixture();
      f.input.config.warehouse = {
        destinationType: "clickhouse",
        protocol: "http",
        hosts: [`${ch.getHost()}:${ch.getMappedPort(8123)}`],
        database: "default",
        username: "default",
        password: "pw",
      };
      f.input.reader = createWarehouseReader as any;
      f.input.config.model = {
        ...f.input.config.model,
        query: "SELECT id, `n.a` AS n_a, `n.b` AS n_b FROM webhook_nested",
        primaryKey: ["id"],
      };
      f.input.config.options = { ...f.input.config.options, mapping: { id: "id", n_a: "n_a", n_b: "n_b" } };
      expect(await execute(f.input)).toBe("COMPLETE");
      expect(received.flatMap(r => r.body.records)[0].data).toEqual({ id: 1, n_a: ["1", "2"], n_b: ["p", "q"] });
    });
  });
});

describe("webhook delivery under faults", () => {
  const ROWS = Number(process.env.WEBHOOK_SOAK_ROWS ?? 3000);
  // The suite's console output is swallowed; WEBHOOK_REPORT=<file> collects the measurements instead.
  const report = (line: string) => {
    if (process.env.WEBHOOK_REPORT) require("node:fs").appendFileSync(process.env.WEBHOOK_REPORT, `${line}\n`);
  };
  const secret = "soak-secret";
  type Action =
    | { kind: "ok" }
    | { kind: "status"; status: number; retryAfter?: string }
    | { kind: "reset"; processed: boolean }
    | { kind: "slow"; ms: number }
    | { kind: "hang"; processed: boolean };
  interface Info {
    ids: number[];
    request: number;
  }
  let server: http.Server;
  let sockets: Set<import("node:net").Socket>;
  const state = {
    delivered: new Map<string, number>(),
    ids: new Map<number, number>(),
    requests: 0,
    badSignatures: 0,
    behaviour: (_info: Info): Action => ({ kind: "ok" }),
    onProcessed: (_uniqueIds: number) => {},
  };
  const reset = () => {
    state.delivered.clear();
    state.ids.clear();
    state.requests = 0;
    state.badSignatures = 0;
    state.behaviour = () => ({ kind: "ok" });
    state.onProcessed = () => {};
  };
  // Small deterministic generator so a failing run can be reproduced.
  const seeded = (seed: number) => () => {
    seed = (seed + 0x6d2b79f5) | 0;
    let t = Math.imul(seed ^ (seed >>> 15), 1 | seed);
    t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
    return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
  };
  beforeEach(async () => {
    reset();
    sockets = new Set();
    server = http.createServer((req, res) => {
      const chunks: Buffer[] = [];
      req.on("data", chunk => chunks.push(chunk));
      req.on("end", () => {
        const text = Buffer.concat(chunks).toString();
        const timestamp = String(req.headers["jitsu-signature-timestamp"]);
        const expected = createHmac("sha256", secret).update(`${timestamp}.${text}`).digest("hex");
        if (req.headers["jitsu-signature"] !== expected) state.badSignatures++;
        const records = JSON.parse(text).records as { key: string; idempotencyKey: string; data: { id: number } }[];
        const action = state.behaviour({ ids: records.map(r => r.data.id), request: ++state.requests });
        const process = () => {
          for (const record of records) {
            state.delivered.set(record.idempotencyKey, (state.delivered.get(record.idempotencyKey) ?? 0) + 1);
            state.ids.set(record.data.id, (state.ids.get(record.data.id) ?? 0) + 1);
          }
          state.onProcessed(state.ids.size);
        };
        switch (action.kind) {
          case "ok":
            process();
            res.statusCode = 200;
            res.end();
            break;
          case "slow":
            process();
            setTimeout(() => {
              res.statusCode = 200;
              res.end();
            }, action.ms);
            break;
          case "status":
            res.statusCode = action.status;
            if (action.retryAfter) res.setHeader("retry-after", action.retryAfter);
            res.end();
            break;
          case "reset":
            if (action.processed) process();
            req.socket.destroy();
            break;
          case "hang":
            if (action.processed) process();
            break;
        }
      });
    });
    server.on("connection", socket => {
      sockets.add(socket);
      socket.on("close", () => sockets.delete(socket));
    });
    await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
  });
  afterEach(async () => {
    for (const socket of sockets) socket.destroy();
    await new Promise<void>(resolve => server.close(() => resolve()));
  });

  function soak(rows = ROWS) {
    const f = fixture();
    const url = `http://127.0.0.1:${(server.address() as AddressInfo).port}/hook`;
    f.input.config.destination = {
      destinationType: "webhook",
      url,
      method: "POST",
      signatureMethod: "hmac",
      signatureSecret: secret,
    };
    f.input.config.model = {
      ...f.input.config.model,
      query: "SELECT id, name FROM source",
      primaryKey: ["id"],
      cursor: { column: "id", type: "number" },
    } as any;
    f.input.config.options = {
      ...f.input.config.options,
      stream: "rows",
      mode: "upsert",
      mapping: { id: "id", name: "name" },
      checkpointEvery: 5000,
      streamOptions: { deliveryAttested: true, allowInsecureHttp: true, recordsPerRequest: 50, concurrency: 2 },
    };
    const deps = {
      ...defaultDeliveryDeps,
      send: createGuardedRequest({ isBlocked: () => false }),
      sleep: async () => {},
    };
    f.input.adapters = new Map([["webhook", cfg => createWebhookRuntime(cfg as any, undefined, deps)]]);
    f.input.reader = () =>
      ({
        sql: {} as any,
        columns: async () => [],
        preview: async () => ({ rows: [], columns: [], truncated: false }),
        close: async () => {},
        stream: async function* (_model: any, after: any, signal: AbortSignal) {
          for (let id = after ? Number(after.value) + 1 : 1; id <= rows; id++) {
            signal.throwIfAborted();
            yield {
              row: { id, name: `user-${id}` },
              deleted: false,
              checkpoint: { value: String(id), primaryKeyValues: [String(id)] },
            };
          }
        },
      } as any);
    let run = 0;
    const next = () => {
      run++;
      f.input = { ...f.input, taskId: `soak-${run}`, controller: new AbortController() };
      return f.input;
    };
    return { f, next, rows };
  }
  const deliveredEverything = (rows: number) => {
    const missing: number[] = [];
    for (let id = 1; id <= rows && missing.length < 5; id++) if (!state.ids.has(id)) missing.push(id);
    return missing;
  };
  const duplicates = () => [...state.delivered.values()].reduce((sum, count) => sum + count - 1, 0);
  // After a run is interrupted mid-request the batch is left uncertain. The next scheduled run fails once with the
  // core's "reconcile before retrying" error and the one after it recovers (no human step). Returns each run's result
  // and the failed runs' task errors.
  const continueAfterInterruption = async (next: () => ExecuteOptions) => {
    const results: string[] = [];
    const errors: string[] = [];
    while (results.at(-1) !== "COMPLETE" && results.length < 4) {
      const input = next();
      results.push(await execute(input));
      if (results.at(-1) === "FAILED") errors.push(String((await task(input.taskId)).error));
    }
    return { results, errors };
  };

  it("delivers every row exactly once-or-more under 503, 429, resets and slow responses, with valid signatures", async () => {
    const random = seeded(7);
    state.behaviour = () => {
      const r = random();
      if (r < 0.015) return { kind: "status", status: 503 };
      if (r < 0.02) return { kind: "status", status: 429, retryAfter: "0" };
      if (r < 0.025) return { kind: "reset", processed: false };
      if (r < 0.035) return { kind: "reset", processed: true };
      if (r < 0.045) return { kind: "slow", ms: 30 };
      return { kind: "ok" };
    };
    const { f, rows } = soak();
    expect(await execute(f.input)).toBe("COMPLETE");
    expect(deliveredEverything(rows)).toEqual([]);
    expect(state.ids.size).toBe(rows);
    expect(state.badSignatures).toBe(0);
    // Resets after processing show up as repeats with the same idempotency key, never as different keys for one row.
    expect(new Set([...state.delivered.keys()]).size).toBe(rows);
    report(`SOAK rows=${rows} requests=${state.requests} duplicates=${duplicates()}`);
  }, 600_000);

  it("a poison record stops the run with its code; after the endpoint is fixed the next run delivers everything and re-sends at most one checkpoint", async () => {
    const { f, next, rows } = soak();
    const poison = Math.floor(rows * 0.6);
    state.behaviour = ({ ids }) => (ids.includes(poison) ? { kind: "status", status: 400 } : { kind: "ok" });
    expect(await execute(next())).toBe("FAILED");
    expect((await task("soak-1")).error).toContain("Reason code: http_400");
    expect(state.ids.has(poison)).toBe(false);
    const deliveredBefore = state.ids.size;
    state.behaviour = () => ({ kind: "ok" });
    expect(await execute(next())).toBe("COMPLETE");
    expect(deliveredEverything(rows)).toEqual([]);
    expect(state.badSignatures).toBe(0);
    const resent = duplicates();
    report(`POISON rows=${rows} poison=${poison} deliveredBeforeFix=${deliveredBefore} resent=${resent}`);
    expect(resent).toBeLessThanOrEqual(5000 + 4 * 50);
  }, 600_000);

  it("cancelling mid-run and running again delivers every row", async () => {
    const { next, rows } = soak();
    const run = next();
    const cutoff = Math.floor(rows * 0.3);
    state.onProcessed = unique => {
      if (unique >= cutoff) run.controller.abort();
    };
    expect(await execute(run)).toBe("CANCELLED");
    expect(state.ids.size).toBeLessThan(rows);
    state.onProcessed = () => {};
    const { results, errors } = await continueAfterInterruption(next);
    report(`CANCEL results=${results.join(",")} errors=${JSON.stringify([...new Set(errors)])}`);
    expect(results.at(-1)).toBe("COMPLETE");
    expect(results.filter(result => result === "FAILED").length).toBeLessThanOrEqual(1);
    expect(deliveredEverything(rows)).toEqual([]);
    report(`CANCEL rows=${rows} cutoff=${cutoff} duplicates=${duplicates()}`);
  }, 600_000);

  it("an abort while a request is in flight (endpoint processed it, never answered) is replayed with the same keys and delivers every row", async () => {
    const { next, rows } = soak();
    const run = next();
    const hangAt = Math.floor(rows / 2);
    let hung = false;
    state.behaviour = ({ ids }) => {
      if (!hung && ids.includes(hangAt)) {
        hung = true;
        setTimeout(() => run.controller.abort(), 100);
        return { kind: "hang", processed: true };
      }
      return { kind: "ok" };
    };
    expect(await execute(run)).not.toBe("COMPLETE");
    expect(state.ids.has(hangAt)).toBe(true);
    const { results, errors } = await continueAfterInterruption(next);
    report(`INFLIGHT results=${results.join(",")} errors=${JSON.stringify([...new Set(errors)])}`);
    expect(results.at(-1)).toBe("COMPLETE");
    expect(results.filter(result => result === "FAILED").length).toBeLessThanOrEqual(1);
    expect(deliveredEverything(rows)).toEqual([]);
    const repeated = [...state.delivered.entries()].filter(([, count]) => count > 1);
    expect(repeated.length).toBeGreaterThan(0);
    report(`INFLIGHT rows=${rows} hangAt=${hangAt} duplicates=${duplicates()}`);
  }, 600_000);
});

describe("webhook SSRF protection through the runner (real address guard, nothing bypassed)", () => {
  let server: http.Server;
  let hits: string[];
  let redirectTo: string | undefined;
  beforeEach(async () => {
    hits = [];
    redirectTo = undefined;
    server = http.createServer((req, res) => {
      hits.push(`${req.method} ${req.url}`);
      req.resume();
      if (redirectTo) {
        res.statusCode = 302;
        res.setHeader("location", redirectTo);
      }
      res.end();
    });
    await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
  });
  afterEach(() => new Promise<void>(resolve => server.close(() => resolve())));

  function syncTo(url: string, realGuard = true) {
    const f = fixture();
    f.input.config.destination = { destinationType: "webhook", url, method: "POST" };
    f.input.config.options = {
      ...f.input.config.options,
      stream: "rows",
      mode: "upsert",
      mapping: { id: "id" },
      streamOptions: { deliveryAttested: true, allowInsecureHttp: true },
    };
    f.setRows([{ id: "a" }]);
    // Production deliver path with its real guard; only the wait between retries is skipped.
    const deps = {
      ...defaultDeliveryDeps,
      sleep: async () => {},
      ...(realGuard ? {} : { send: createGuardedRequest({ isBlocked: (address: string) => address !== "127.0.0.1" }) }),
    };
    f.input.adapters = new Map([["webhook", cfg => createWebhookRuntime(cfg as any, undefined, deps)]]);
    return f;
  }

  const blocked = (port: number) => [
    `http://127.0.0.1:${port}/hook`,
    `http://localhost:${port}/hook`,
    `http://[::1]:${port}/hook`,
    `http://[::ffff:127.0.0.1]:${port}/hook`,
    `http://[::ffff:7f00:1]:${port}/hook`,
    `http://2130706433:${port}/hook`,
    `http://0x7f.1:${port}/hook`,
    `http://0.0.0.0:${port}/hook`,
    `http://[::]:${port}/hook`,
    "http://169.254.169.254/latest/meta-data/",
    "http://[fe80::1]/",
    "http://10.0.0.1/",
    "http://172.16.0.1/",
    "http://192.168.1.1/",
    "http://100.64.0.1/",
    "http://224.0.0.1/",
    "http://[64:ff9b::7f00:1]/",
    "http://[fd00::1]/",
  ];

  it("refuses loopback, link-local, private, CGNAT, multicast, mapped and NAT64 targets with a clear code and no request", async () => {
    const port = (server.address() as AddressInfo).port;
    for (const url of blocked(port)) {
      const f = syncTo(url);
      f.input.taskId = `ssrf-${Math.random().toString(36).slice(2)}`;
      expect(await execute(f.input), url).toBe("FAILED");
      const error = String((await task(f.input.taskId)).error);
      expect(error, url).toContain("Reason code: blocked_address");
      expect(error, url).not.toContain("127.0.0.1");
      await admin.query(
        "TRUNCATE newjitsu.reverse_sync_control,newjitsu.reverse_sync_target_owner,newjitsu.source_state,newjitsu.source_task,newjitsu.task_log"
      );
    }
    expect(hits).toEqual([]);
  }, 300_000);

  it("does not follow a redirect to an internal address", async () => {
    redirectTo = "http://169.254.169.254/latest/meta-data/";
    const f = syncTo(`http://127.0.0.1:${(server.address() as AddressInfo).port}/hook`, false);
    expect(await execute(f.input)).toBe("FAILED");
    expect(String((await task()).error)).toContain("Reason code: redirect_refused");
    expect(hits).toEqual(["POST /hook"]);
  });
});