diff --git a/.github/workflows/kestra.yml b/.github/workflows/kestra.yml
new file mode 100644
index 0000000..47bfc46
--- /dev/null
+++ b/.github/workflows/kestra.yml
@@ -0,0 +1,49 @@
+name: Kestra 2.0 candidate fixture
+
+on:
+  push:
+    paths:
+      - "integrations/kestra/**"
+      - ".github/workflows/kestra.yml"
+  pull_request:
+    paths:
+      - "integrations/kestra/**"
+      - ".github/workflows/kestra.yml"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  native-synthetic-fixture:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 15
+    defaults:
+      run:
+        working-directory: integrations/kestra
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - name: Build pinned official runtime and fixture image
+        run: docker build -f tests/Dockerfile -t sanctionskit-kestra-fixture .
+      - name: Execute native flow with only container loopback available
+        run: |
+          mkdir -p test-results
+          docker run --rm --network none --cap-drop ALL \
+            --hostname kestra-fixture --add-host kestra-fixture:127.0.0.1 \
+            --security-opt no-new-privileges:true --read-only \
+            --memory 3g --cpus 2 --pids-limit 512 \
+            --user "$(id -u):$(id -g)" \
+            --tmpfs /tmp:rw,nosuid,nodev,size=1g,mode=1777 \
+            --tmpfs /work:rw,nosuid,nodev,size=1g,mode=1777 \
+            --mount "type=bind,source=$PWD/test-results,target=/evidence" \
+            sanctionskit-kestra-fixture
+      - name: Keep native fixture evidence
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: kestra-2-native-fixture-${{ github.run_id }}
+          path: integrations/kestra/test-results/
+          retention-days: 7
+          if-no-files-found: warn
diff --git a/README.md b/README.md
index f3c3a66..5d6641f 100644
--- a/README.md
+++ b/README.md
@@ -34,6 +34,8 @@ npm --prefix integrations/windmill test
 
 [Node-RED](integrations/node-red/) includes an importable flow built with core nodes. Run a fictional screening, retrieve its evidence, and inspect a review record. It keeps credentials in the process environment and reuses a stable request key for manual retries. The README includes setup instructions and local runtime tests.
 
+[Kestra](integrations/kestra/) submits a fictional batch, waits for its final state, and exports retained evidence and a CSV row summary. It keeps failed and cancelled rows visible for review. The Kestra 2.0.4 flow passed 24 native scenarios against a local HTTP fixture in an isolated container.
+
 ## Make your first request
 
 1. [Create a workspace](https://www.sanctionskit.com/signup), then open [API keys](https://www.sanctionskit.com/dashboard/keys?environment=sandbox).
@@ -87,7 +89,7 @@ These scripts display responses for learning. When adapting them to an applicati
 
 `REQUEST_KEY` is sent as the `Idempotency-Key` header. Generate it once for a new screening. If a request times out or needs a retry, keep **the same key and the same body**. Rerun the script without repeating the key-generation command. A new subject or changed body needs a new key.
 
-The examples make one attempt and exit with a nonzero status on an HTTP or network error. They do not retry automatically. An error is an incomplete request, never a `no_match` result. See [idempotency](https://www.sanctionskit.com/docs/idempotency) and [error handling](https://www.sanctionskit.com/docs/errors) before adding retries to an application.
+The standalone language examples make one attempt and exit with a nonzero status on an HTTP or network error. They do not retry automatically. An error is an incomplete request, never a `no_match` result. See [idempotency](https://www.sanctionskit.com/docs/idempotency) and [error handling](https://www.sanctionskit.com/docs/errors) before adding retries to an application.
 
 ## Adapt the example
 
diff --git a/integrations/kestra/.dockerignore b/integrations/kestra/.dockerignore
new file mode 100644
index 0000000..2e85623
--- /dev/null
+++ b/integrations/kestra/.dockerignore
@@ -0,0 +1,5 @@
+__pycache__
+**/__pycache__
+tests/native-results.json
+tests/last-failure.json
+test-results
diff --git a/integrations/kestra/.gitignore b/integrations/kestra/.gitignore
new file mode 100644
index 0000000..0f67333
--- /dev/null
+++ b/integrations/kestra/.gitignore
@@ -0,0 +1,7 @@
+__pycache__/
+tests/native-results.json
+tests/last-failure.json
+tests/last-failure-diagnostics.json
+test-results/
+synthetic-review-bundle.json
+synthetic-row-summary.csv
diff --git a/integrations/kestra/LICENSE b/integrations/kestra/LICENSE
new file mode 100644
index 0000000..94d712c
--- /dev/null
+++ b/integrations/kestra/LICENSE
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 SanctionsKit
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/integrations/kestra/README.md b/integrations/kestra/README.md
new file mode 100644
index 0000000..a0592c5
--- /dev/null
+++ b/integrations/kestra/README.md
@@ -0,0 +1,83 @@
+# Kestra batch screening and evidence
+
+This Kestra 2.0.4 flow submits two fictional contacts to the SanctionsKit synthetic
+sandbox, waits for the batch to finish, and collects retained evidence with a CSV
+row summary. Completed, failed and cancelled rows stay visible. A successful
+collection does not approve onboarding or mean that every row screened successfully.
+
+All 24 native fixture scenarios passed on Kestra 2.0.4 with serdes 2.0.6 and
+Temurin 25 in an isolated container. The [verified CI run](https://github.com/SanctionsKit/sanctions-kit-examples/actions/runs/37181224466)
+used a local fictional HTTP server with Docker networking disabled. This verifies
+the workflow and its error handling against fixtures; a hosted sandbox run remains
+separate.
+
+## Isolated fixture test
+
+The CI job builds an image from official Python and Eclipse Temurin images pinned
+by digest, downloads Kestra 2.0.4 with a checked SHA-256, and installs serdes 2.0.6
+with a checked artifact hash. [runtime-lock.json](runtime-lock.json) records these
+sources. Network access is needed only while building the image.
+
+Execution uses Docker `--network none`, with no published ports or Docker socket.
+The fixed hostname resolves to `127.0.0.1` through the container's hosts file, so
+Java can identify the local host without DNS. The test runner checks that mapping
+and refuses to start unless its only network interface is `lo`. It
+runs as an unprivileged user with all Linux capabilities dropped, no new privileges,
+a read-only root filesystem, and temporary writable directories. The only service
+key is the invented string `local-mock-only`; no GitHub secret or real API key is
+needed. No hosted SanctionsKit request can leave the container.
+
+From this directory on a machine with Docker:
+
+```sh
+docker build -f tests/Dockerfile -t sanctionskit-kestra-fixture .
+mkdir -p test-results
+docker run --rm --network none --cap-drop ALL \
+  --hostname kestra-fixture --add-host kestra-fixture:127.0.0.1 \
+  --security-opt no-new-privileges:true --read-only \
+  --memory 3g --cpus 2 --pids-limit 512 \
+  --user "$(id -u):$(id -g)" \
+  --tmpfs /tmp:rw,nosuid,nodev,size=1g,mode=1777 \
+  --tmpfs /work:rw,nosuid,nodev,size=1g,mode=1777 \
+  --mount "type=bind,source=$PWD/test-results,target=/evidence" \
+  sanctionskit-kestra-fixture
+```
+
+The fixture imports the actual flow and changes only the API host, polling interval
+and polling budget. Request bodies, evidence checks, HTTP limits and exports stay
+unchanged. It expects all 24 scenarios to pass, including completion, rate-limit
+retry, failed/cancelled rows, timeouts, HTTP errors, unsafe IDs, mismatched evidence,
+pagination, row order and counters. It checks the parent execution and native Loop
+iteration executions, uses the 2.0 outputs API, downloads the JSON/CSV files, and
+checks logs and outputs for the fake key.
+
+The server log, per-case results, isolation report and generated files go into
+`test-results/`. A failed assertion also saves bounded parent and Loop task logs
+and outputs, with fixture credentials redacted. A parser or import pass is not
+a substitute for this run.
+
+## Sandbox setup
+
+Create a free [developer workspace](https://www.sanctionskit.com/signup?workflow=api)
+and a sandbox key with `batches:write` and `results:read`. Save the key as the Kestra
+secret `SANCTIONSKIT_API_KEY`. The sandbox uses synthetic records and needs no card.
+Production screening needs a plan and payment method.
+
+The two fixed contacts keep their partial birth years as strings. Each row requests
+`sandbox@1` and standard retention. Use a fresh `request_key` for a new business
+batch and the same key/body for retries. Redirects are disabled. A 202 only confirms
+acceptance; the bounded loop waits for a terminal batch state.
+
+This starter expects one complete ordered page. Unexpected pagination, counters,
+IDs, contacts or retained requests stop the flow. It fetches same-ID evidence only
+for completed rows and preserves unsuccessful row outcomes. A `potential_match`
+needs review; `no_match` means no candidates within the selected coverage and
+matching rules. Neither is an account approval or legal clearance.
+
+The daily trigger is disabled. Before adapting this to larger inputs or enabling
+scheduling, read the [batch API guide](https://www.sanctionskit.com/docs/batches) and
+choose a durable business-batch key. No external notifications, account decisions
+or database writes are part of this example.
+
+No Kestra catalog submission or placement is claimed. Upstream blueprint licensing
+and catalog acceptance remain separate from company source publication.
diff --git a/integrations/kestra/runtime-lock.json b/integrations/kestra/runtime-lock.json
new file mode 100644
index 0000000..5629c89
--- /dev/null
+++ b/integrations/kestra/runtime-lock.json
@@ -0,0 +1,37 @@
+{
+  "checked_at": "2026-10-04T04:51:28.638389+00:00",
+  "kestra": {
+    "version": "2.0.4",
+    "url": "https://github.com/kestra-io/kestra/releases/download/v2.0.4/kestra-2.0.4",
+    "sha256": "88ee8d44c3d4a234c9d04689e72b8ba903ae8f9f2c6ae5f94bad89f3bcd68513"
+  },
+  "plugin": {
+    "coordinates": "io.kestra.plugin:plugin-serdes:2.0.6",
+    "sha256": "1394bf5ec0e7a57e446adc2b326e4ac72651d3fe525853d4d1c92bef60fe117e",
+    "maven_sha1_matches": true
+  },
+  "docker_images": [
+    {
+      "image": "docker.io/library/python",
+      "tag": "3.13-slim-trixie",
+      "index_digest": "sha256:bb2988715db2cf7ace7b53f38f3cffbef7c7046a656bee66245eb0ed386e2e81",
+      "linux_amd64_digest": "sha256:b92e6b9bb1ea9d826e9956fd8d30bf18bc384c130231d74fdf0ba3460c290b0b"
+    },
+    {
+      "image": "docker.io/library/eclipse-temurin",
+      "tag": "25-jre-noble",
+      "index_digest": "sha256:398f810215757dc1926390014272579fb0e57c41ef1c8aa4f64ae761613a168b",
+      "linux_amd64_digest": "sha256:78498f30dd330b06755c1b134039dcbf8324bedb385136819dd519b842a3816a"
+    }
+  ],
+  "actions": {
+    "checkout": {
+      "version": "v7.0.1",
+      "sha": "3d3c42e5aac5ba805825da76410c181273ba90b1"
+    },
+    "upload_artifact": {
+      "version": "v7.0.1",
+      "sha": "043fb46d1a93c77aae656e7c1c64a875d1fc6a0a"
+    }
+  }
+}
diff --git a/integrations/kestra/sanctionskit-batch-evidence.yaml b/integrations/kestra/sanctionskit-batch-evidence.yaml
new file mode 100644
index 0000000..a4d5abf
--- /dev/null
+++ b/integrations/kestra/sanctionskit-batch-evidence.yaml
@@ -0,0 +1,217 @@
+id: sanctionskit-batch-evidence
+namespace: company.sanctionskit
+description: |
+  Candidate for Kestra 2.0.4 native fixture CI. Not yet qualified for use.
+  Run two fictional contacts and collect row-level retained evidence.
+inputs:
+  - id: request_key
+    type: STRING
+    required: true
+    validator: ^[A-Za-z0-9_:.-]{8,128}$
+    description: A fresh key for a new batch. Reuse it when retrying the same batch.
+  - id: synthetic
+    type: BOOL
+    defaults: false
+    description: Confirm that this example contains fictional contacts only.
+variables:
+  api_base: https://www.sanctionskit.com/api/v1
+  subjects:
+    - subject:
+        name: Alex Morgan
+        entityType: person
+        birthDate: "1984"
+      package: sandbox@1
+      retention: standard
+      reference: example-customer-001
+    - subject:
+        name: Taylor Jordan
+        entityType: person
+        birthDate: "1979"
+      package: sandbox@1
+      retention: standard
+      reference: example-customer-002
+tasks:
+  - id: require_synthetic
+    type: io.kestra.plugin.core.execution.Assert
+    conditions:
+      - "{{ inputs.synthetic == true }}"
+    errorMessage: Confirm the fictional contacts before running this sandbox example.
+  - id: submit_batch
+    type: io.kestra.plugin.core.http.Request
+    method: POST
+    uri: "{{ vars.api_base }}/batches"
+    headers:
+      Authorization: Bearer {{ secret('SANCTIONSKIT_API_KEY') }}
+      Content-Type: application/json
+      Idempotency-Key: "{{ inputs.request_key }}"
+    body: |
+      {"name":"Kestra fictional onboarding example","subjects":{{ vars.subjects | toJson }}}
+    timeout: PT20S
+    options:
+      followRedirects: false
+      allowFailed: false
+      timeout:
+        connectTimeout: PT5S
+        readIdleTimeout: PT10S
+  - id: check_acceptance
+    type: io.kestra.plugin.core.execution.Assert
+    conditions:
+      - "{{ outputs.submit_batch.code == 202 }}"
+      - "{{ fromJson(outputs.submit_batch.body).data.id | toJson | jq('test(\"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$\")') | first }}"
+      - "{{ fromJson(outputs.submit_batch.body).data.total == 2 }}"
+      - "{{ fromJson(outputs.submit_batch.body).data.status == 'pending' }}"
+    errorMessage: Expected acceptance of exactly two rows. Acceptance is not a completed screening.
+  - id: wait_for_batch
+    type: io.kestra.plugin.core.flow.LoopUntil
+    condition: "{{ (fromJson(outputs.poll_batch.body).data.status == 'completed' or fromJson(outputs.poll_batch.body).data.status == 'failed' or fromJson(outputs.poll_batch.body).data.status == 'cancelled') }}"
+    failOnMaxReached: true
+    checkFrequency:
+      interval: PT2S
+      maxIterations: 30
+      maxDuration: PT2M
+    tasks:
+      - id: poll_batch
+        type: io.kestra.plugin.core.http.Request
+        method: GET
+        uri: "{{ vars.api_base }}/batches/{{ fromJson(outputs.submit_batch.body).data.id }}?limit=100&offset=0"
+        timeout: PT20S
+        options:
+          followRedirects: false
+          allowFailed: false
+          timeout:
+            connectTimeout: PT5S
+            readIdleTimeout: PT10S
+        headers:
+          Authorization: Bearer {{ secret('SANCTIONSKIT_API_KEY') }}
+          Content-Type: application/json
+      - id: check_poll_identity
+        type: io.kestra.plugin.core.execution.Assert
+        conditions:
+          - "{{ outputs.poll_batch.code == 200 }}"
+          - "{{ fromJson(outputs.poll_batch.body).data.id == fromJson(outputs.submit_batch.body).data.id }}"
+          - "{{ fromJson(outputs.poll_batch.body).data.environment == 'sandbox' }}"
+          - "{{ fromJson(outputs.poll_batch.body).data.total == 2 }}"
+          - "{{ (fromJson(outputs.poll_batch.body).data.status == 'pending' or fromJson(outputs.poll_batch.body).data.status == 'processing' or fromJson(outputs.poll_batch.body).data.status == 'completed' or fromJson(outputs.poll_batch.body).data.status == 'failed' or fromJson(outputs.poll_batch.body).data.status == 'cancelled') }}"
+        errorMessage: Batch identity, environment or state did not match the sandbox request.
+  - id: check_complete_page
+    type: io.kestra.plugin.core.execution.Assert
+    conditions:
+      - "{{ fromJson(outputs.poll_batch.body).data.nextOffset == null }}"
+      - "{{ fromJson(outputs.poll_batch.body).data.rows | length == 2 }}"
+      - "{{ fromJson(outputs.poll_batch.body).data.rows[0].row_number == 1 }}"
+      - "{{ fromJson(outputs.poll_batch.body).data.rows[1].row_number == 2 }}"
+      - "{{ fromJson(outputs.poll_batch.body).data.completed == (outputs.poll_batch.body | jq('.data.rows | map(select(.status == \"completed\")) | length') | first) }}"
+      - "{{ fromJson(outputs.poll_batch.body).data.failed == (outputs.poll_batch.body | jq('.data.rows | map(select(.status == \"failed\")) | length') | first) }}"
+    errorMessage: This two-row starter requires one complete ordered page. Stop before using partial or unexpected rows.
+  - id: collect_evidence
+    type: io.kestra.plugin.core.flow.Loop
+    values:
+      - "0"
+      - "1"
+    concurrencyLimit: 1
+    tasks:
+      - id: check_row
+        type: io.kestra.plugin.core.execution.Assert
+        conditions:
+          - "{{ (fromJson(outputs.poll_batch.body).data.rows[item.value | number].status == 'completed' or fromJson(outputs.poll_batch.body).data.rows[item.value | number].status == 'failed' or fromJson(outputs.poll_batch.body).data.rows[item.value | number].status == 'cancelled') }}"
+        errorMessage: A terminal batch still contains an unfinished row.
+      - id: completed_row
+        type: io.kestra.plugin.core.flow.If
+        condition: "{{ fromJson(outputs.poll_batch.body).data.rows[item.value | number].status == 'completed' }}"
+        then:
+          - id: check_screening_id
+            type: io.kestra.plugin.core.execution.Assert
+            conditions:
+              - "{{ fromJson(outputs.poll_batch.body).data.rows[item.value | number].screening_id | toJson | jq('test(\"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$\")') | first }}"
+          - id: fetch_evidence
+            type: io.kestra.plugin.core.http.Request
+            method: GET
+            uri: "{{ vars.api_base }}/results/{{ fromJson(outputs.poll_batch.body).data.rows[item.value | number].screening_id }}/evidence"
+            timeout: PT20S
+            options:
+              followRedirects: false
+              allowFailed: false
+              timeout:
+                connectTimeout: PT5S
+                readIdleTimeout: PT10S
+            headers:
+              Authorization: Bearer {{ secret('SANCTIONSKIT_API_KEY') }}
+              Content-Type: application/json
+          - id: check_evidence
+            type: io.kestra.plugin.core.execution.Assert
+            conditions:
+              - "{{ outputs.fetch_evidence.code == 200 }}"
+              - "{{ fromJson(outputs.fetch_evidence.body).format == 'sanctionskit-evidence@1' }}"
+              - "{{ fromJson(outputs.fetch_evidence.body).result.id == fromJson(outputs.poll_batch.body).data.rows[item.value | number].screening_id }}"
+              - "{{ fromJson(outputs.fetch_evidence.body).result.environment == 'sandbox' }}"
+              - "{{ (fromJson(outputs.fetch_evidence.body).result.status == 'no_match' or fromJson(outputs.fetch_evidence.body).result.status == 'potential_match') }}"
+              - "{{ fromJson(outputs.fetch_evidence.body).retention == 'standard' }}"
+              - "{{ fromJson(outputs.fetch_evidence.body).retainedInputs == true }}"
+              - "{{ fromJson(outputs.fetch_evidence.body).result.versions.package == 'sandbox@1' }}"
+              - "{{ fromJson(outputs.fetch_evidence.body).result.status == 'no_match' ? fromJson(outputs.fetch_evidence.body).result.matches | length == 0 : fromJson(outputs.fetch_evidence.body).result.matches | length > 0 }}"
+              - "{{ fromJson(outputs.fetch_evidence.body).reference == vars.subjects[item.value | number].reference }}"
+              - "{{ fromJson(outputs.fetch_evidence.body).subject == vars.subjects[item.value | number].subject }}"
+              - "{{ fromJson(outputs.fetch_evidence.body).request == vars.subjects[item.value | number] }}"
+            errorMessage: Retained evidence does not match this sandbox batch row and fictional contact.
+    fetchType: FETCH
+    outputs:
+      - id: evidence
+        type: JSON
+        required: false
+        value: "{{ outputs.fetch_evidence is defined ? fromJson(outputs.fetch_evidence.body) : null }}"
+  - id: write_review_bundle
+    type: io.kestra.plugin.core.storage.Write
+    extension: .json
+    content: |
+      {
+        "synthetic":true,
+        "onboardingApproved":false,
+        "requestKey":{{ inputs.request_key | toJson }},
+        "batch":{{ fromJson(outputs.poll_batch.body).data | toJson }},
+        "evidenceByRow":{{ (outputs.collect_evidence.outputs | default([])) | toJson | jq('map(select(.outputs.evidence != null) | {key: ((.item.value | tonumber) + 1 | tostring), value: .outputs.evidence}) | from_entries') | first | toJson }},
+        "nextStep":"Review each row and its evidence. A name-screening result does not approve onboarding."
+      }
+  - id: write_row_summary
+    type: io.kestra.plugin.core.storage.Write
+    extension: .json
+    content: |
+      {% set evidenceByRow = (outputs.collect_evidence.outputs | default([])) | toJson | jq('map(select(.outputs.evidence != null) | {key: ((.item.value | tonumber) + 1 | tostring), value: .outputs.evidence}) | from_entries') | first %}
+      [
+        {% for row in fromJson(outputs.poll_batch.body).data.rows %}
+        {"row_number":{{ row.row_number }},
+         "reference":{{ vars.subjects[row.row_number - 1].reference | toJson }},
+         "row_status":{{ row.status | toJson }},
+         "screening_id":{{ row.screening_id | toJson }},
+         "screening_status":{{ (row.status == 'completed' ? evidenceByRow[row.row_number | string].result.status : '') | toJson }},
+         "error_code":{{ (row.error == null ? '' : row.error.code) | toJson }},
+         "synthetic":true,"onboarding_approved":false}{% if not loop.last %},{% endif %}
+        {% endfor %}
+      ]
+  - id: summary_to_ion
+    type: io.kestra.plugin.serdes.json.JsonToIon
+    from: "{{ outputs.write_row_summary.uri }}"
+    newLine: false
+    version: 2.0.6
+  - id: summary_to_csv
+    type: io.kestra.plugin.serdes.csv.IonToCsv
+    from: "{{ outputs.summary_to_ion.uri }}"
+    header: true
+    version: 2.0.6
+outputs:
+  - id: review_bundle
+    type: FILE
+    value: "{{ outputs.write_review_bundle.uri | default(null) }}"
+  - id: row_summary
+    type: FILE
+    value: "{{ outputs.summary_to_csv.uri | default(null) }}"
+  - id: batch_status
+    type: STRING
+    value: "{{ outputs.poll_batch is defined ? fromJson(outputs.poll_batch.body).data.status : 'not_collected' }}"
+triggers:
+  - id: optional_daily_example
+    type: io.kestra.plugin.core.trigger.Schedule
+    cron: 0 9 * * *
+    disabled: true
+    inputs:
+      synthetic: true
+      request_key: kestra-example-{{ trigger.date | date('yyyyMMdd') }}
diff --git a/integrations/kestra/tests/Dockerfile b/integrations/kestra/tests/Dockerfile
new file mode 100644
index 0000000..af792e2
--- /dev/null
+++ b/integrations/kestra/tests/Dockerfile
@@ -0,0 +1,15 @@
+FROM eclipse-temurin:25-jre-noble@sha256:398f810215757dc1926390014272579fb0e57c41ef1c8aa4f64ae761613a168b AS java
+FROM python:3.13-slim-trixie@sha256:bb2988715db2cf7ace7b53f38f3cffbef7c7046a656bee66245eb0ed386e2e81
+COPY --from=java /opt/java/openjdk /opt/java/openjdk
+ENV JAVA_HOME=/opt/java/openjdk
+ENV PATH=/opt/java/openjdk/bin:$PATH
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+COPY . /opt/example
+RUN python /opt/example/tests/bootstrap.py
+RUN java -jar /opt/kestra/kestra-2.0.4 plugins install --locally \
+    -p /opt/kestra/plugins -c /opt/example/tests/local-kestra.yml \
+    io.kestra.plugin:plugin-serdes:2.0.6
+RUN python -c "import hashlib,pathlib; p=pathlib.Path('/opt/kestra/plugins/io_kestra_plugin__plugin-serdes__2_0_6.jar'); assert hashlib.sha256(p.read_bytes()).hexdigest() == '1394bf5ec0e7a57e446adc2b326e4ac72651d3fe525853d4d1c92bef60fe117e'"
+WORKDIR /work
+ENTRYPOINT ["python", "/opt/example/tests/container_runner.py"]
diff --git a/integrations/kestra/tests/bootstrap.py b/integrations/kestra/tests/bootstrap.py
new file mode 100644
index 0000000..3c02dff
--- /dev/null
+++ b/integrations/kestra/tests/bootstrap.py
@@ -0,0 +1,13 @@
+"""Download one pinned official Kestra runtime during image build."""
+import hashlib
+from pathlib import Path
+import urllib.request
+
+url = 'https://github.com/kestra-io/kestra/releases/download/v2.0.4/kestra-2.0.4'
+expected = '88ee8d44c3d4a234c9d04689e72b8ba903ae8f9f2c6ae5f94bad89f3bcd68513'
+path = Path('/opt/kestra/kestra-2.0.4')
+path.parent.mkdir(parents=True, exist_ok=True)
+with urllib.request.urlopen(url, timeout=60) as response, path.open('wb') as output:
+    while chunk := response.read(1024 * 1024):
+        output.write(chunk)
+assert hashlib.sha256(path.read_bytes()).hexdigest() == expected, 'Runtime checksum mismatch'
diff --git a/integrations/kestra/tests/container_runner.py b/integrations/kestra/tests/container_runner.py
new file mode 100644
index 0000000..405d387
--- /dev/null
+++ b/integrations/kestra/tests/container_runner.py
@@ -0,0 +1,58 @@
+"""Run the engine and fixture tests inside a network-none disposable container."""
+import json
+import os
+from pathlib import Path
+import shutil
+import socket
+import subprocess
+import sys
+import time
+import urllib.request
+
+# A network-none Linux container has only loopback. Refuse any other interface.
+interfaces = sorted(p.name for p in Path('/sys/class/net').iterdir())
+assert interfaces == ['lo'], f'Expected isolated loopback-only container, got {interfaces}'
+assert os.geteuid() != 0, 'Run the test container as an unprivileged user'
+hostname = socket.gethostname()
+assert hostname == 'kestra-fixture', 'Run the test container with --hostname kestra-fixture'
+hostname_addresses = sorted({address[4][0] for address in socket.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)})
+assert hostname_addresses == ['127.0.0.1'], f'Expected hostname to resolve only to loopback, got {hostname_addresses}'
+root = Path('/work/example')
+shutil.copytree('/opt/example', root)
+work = Path('/work/runtime')
+work.mkdir()
+log = Path('/evidence/kestra-server.log')
+config = root / 'tests/local-kestra.yml'
+command = ['java', '-Xmx1g', '-Djava.awt.headless=true', '-Duser.home=/work/runtime', '-jar', '/opt/kestra/kestra-2.0.4', 'server', 'local', '-c', str(config), '-p', '/opt/kestra/plugins', '--port=28101', '--worker-thread=4', '--no-tutorials']
+env = dict(os.environ, SECRET_SANCTIONSKIT_API_KEY='bG9jYWwtbW9jay1vbmx5')
+result = 1
+with log.open('w') as stream:
+    process = subprocess.Popen(command, cwd=work, env=env, stdout=stream, stderr=subprocess.STDOUT)
+    try:
+        deadline = time.monotonic() + 120
+        while True:
+            assert process.poll() is None, 'Kestra exited before readiness'
+            assert time.monotonic() < deadline, 'Kestra readiness timed out'
+            try:
+                with urllib.request.urlopen('http://127.0.0.1:28101/ping', timeout=2) as response:
+                    if response.status == 200:
+                        break
+            except (OSError, urllib.error.URLError):
+                time.sleep(1)
+        result = subprocess.run([sys.executable, str(root / 'tests/native_test.py')], cwd=root, timeout=360).returncode
+    finally:
+        process.terminate()
+        try:
+            process.wait(timeout=30)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait(timeout=5)
+        for name in ['tests/native-results.json', 'tests/last-failure.json', 'tests/last-failure-diagnostics.json', 'synthetic-review-bundle.json', 'synthetic-row-summary.csv']:
+            source = root / name
+            if source.exists():
+                shutil.copy2(source, Path('/evidence') / source.name)
+        report = {'network_interfaces': interfaces, 'hostname': hostname, 'hostname_addresses': hostname_addresses, 'runtime': 'Kestra2.0.4', 'fixture_tests_exit': result, 'hosted_api_calls': 0, 'service_credentials_used': False}
+        Path('/evidence/container-boundary.json').write_text(json.dumps(report, indent=2) + '\n')
+        # The only key involved is fake, but logs still must not expose it.
+        assert 'local-mock-only' not in log.read_text(), 'Mock key leaked to server log'
+sys.exit(result)
diff --git a/integrations/kestra/tests/fixtures/synthetic-evidence.json b/integrations/kestra/tests/fixtures/synthetic-evidence.json
new file mode 100644
index 0000000..de12c1f
--- /dev/null
+++ b/integrations/kestra/tests/fixtures/synthetic-evidence.json
@@ -0,0 +1,142 @@
+{
+  "format": "sanctionskit-evidence@1",
+  "result": {
+    "id": "00000000-0000-4000-8000-000000000001",
+    "status": "potential_match",
+    "matches": [
+      {
+        "score": 96,
+        "record": {
+          "id": "sandbox:person-001",
+          "list": "Synthetic demonstration records",
+          "names": [
+            {
+              "kind": "primary",
+              "value": "Alex Morgan"
+            },
+            {
+              "kind": "alias",
+              "value": "\u0410\u043b\u0435\u043a\u0441 \u041c\u043e\u0440\u0433\u0430\u043d"
+            }
+          ],
+          "sourceId": "sandbox-synthetic",
+          "addresses": [],
+          "authority": "SanctionsKit synthetic examples",
+          "sourceUrl": "https://sanctionskit.com/demo",
+          "birthDates": [
+            {
+              "value": "1984",
+              "original": "1984",
+              "precision": "year"
+            }
+          ],
+          "entityType": "person",
+          "extensions": {
+            "synthetic": true
+          },
+          "identifiers": [
+            {
+              "type": "passport",
+              "value": "DEMO-001",
+              "issuer": "Fictional authority"
+            }
+          ],
+          "designations": [
+            {
+              "wording": "Invented example, not a government designation.",
+              "category": "other",
+              "programs": [
+                "Synthetic demonstration only"
+              ],
+              "legalReferences": []
+            }
+          ],
+          "nationalities": []
+        },
+        "evidence": [
+          {
+            "field": "name",
+            "method": "normalized_name_exact",
+            "queryValue": "Alex Morgan",
+            "explanation": "Name similarity is a review signal, not a probability or finding of wrongdoing.",
+            "sourceValue": "Alex Morgan",
+            "contribution": 96
+          }
+        ],
+        "conflicts": [],
+        "assessment": {
+          "name": {
+            "method": "normalized_name_exact",
+            "aliasKind": "primary",
+            "similarity": 96,
+            "aliasQuality": "unknown"
+          },
+          "country": {
+            "status": "not_supplied"
+          },
+          "birthDate": {
+            "status": "not_supplied"
+          },
+          "alertBasis": "name",
+          "entityType": {
+            "status": "agreement"
+          },
+          "reviewTier": "name_only",
+          "identifiers": {
+            "status": "not_supplied",
+            "matched": 0,
+            "conflicts": 0,
+            "issuerUnknown": 0,
+            "issuerAgreements": 0
+          },
+          "hasConflicts": false,
+          "sameNameCandidates": 1
+        }
+      }
+    ],
+    "coverage": [
+      {
+        "fresh": true,
+        "version": "1",
+        "sourceId": "sandbox-synthetic",
+        "retrievedAt": "2026-09-17T12:00:00.000Z"
+      }
+    ],
+    "matching": {
+      "profile": "standard",
+      "version": "review-1.2.0",
+      "nameThreshold": 78,
+      "retrievalVersion": "indexed-1.1.1",
+      "corroboratedNameThreshold": 75
+    },
+    "versions": {
+      "policy": "review-1.2.0",
+      "dataset": "sandbox-synthetic-v1",
+      "package": "sandbox@1",
+      "matchingEngine": "deterministic-1.2.0"
+    },
+    "createdAt": "2026-09-17T12:00:00.000Z",
+    "disclaimer": "Potential matches support analyst review. No match means no match in the selected sources and dataset, not legal clearance. Scores and evidence categories support review; they are not probabilities of identity or wrongdoing.",
+    "environment": "sandbox"
+  },
+  "retention": "standard",
+  "expires_at": null,
+  "subject": {
+    "name": "Alex Morgan",
+    "entityType": "person",
+    "birthDate": "1984"
+  },
+  "reference": "example-customer-001",
+  "request": {
+    "subject": {
+      "name": "Alex Morgan",
+      "entityType": "person",
+      "birthDate": "1984"
+    },
+    "package": "sandbox@1",
+    "reference": "example-customer-001",
+    "retention": "standard"
+  },
+  "retainedInputs": true,
+  "replayLimit": null
+}
diff --git a/integrations/kestra/tests/local-kestra.yml b/integrations/kestra/tests/local-kestra.yml
new file mode 100644
index 0000000..c35979e
--- /dev/null
+++ b/integrations/kestra/tests/local-kestra.yml
@@ -0,0 +1,25 @@
+# Disposable local test instance. These are invented test credentials.
+micronaut:
+  server:
+    host: 127.0.0.1
+    port: 28101
+  executors:
+    io:
+      type: fixed
+      nThreads: 8
+kestra:
+  anonymous-usage-report:
+    enabled: false
+  ui-anonymous-usage-report:
+    enabled: false
+  tutorial-flows:
+    enabled: false
+  plugins:
+    auto-install:
+      enabled: false
+  server:
+    basic-auth:
+      username: local@sanctionskit.invalid
+      password: LocalMockOnly123!
+  controller:
+    port: 28103
diff --git a/integrations/kestra/tests/native_test.py b/integrations/kestra/tests/native_test.py
new file mode 100644
index 0000000..1b294a0
--- /dev/null
+++ b/integrations/kestra/tests/native_test.py
@@ -0,0 +1,281 @@
+"""Execute the actual flow in a loopback-only Kestra 2.0.4 instance.
+
+Only fictional fixtures are sent to an in-process local HTTP server. This file
+never contacts the hosted SanctionsKit API and never needs a real API key.
+"""
+import base64
+import copy
+import csv
+import io
+import json
+import os
+from pathlib import Path
+import re
+import threading
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+ROOT = Path(__file__).resolve().parents[1]
+KESTRA = os.getenv('KESTRA_TEST_URL', 'http://127.0.0.1:28101')
+assert urllib.parse.urlsplit(KESTRA).hostname in ('127.0.0.1', 'localhost')
+AUTH = 'Basic ' + base64.b64encode(b'local@sanctionskit.invalid:LocalMockOnly123!').decode()
+MOCK_KEY = 'local-mock-only'
+BATCH = '11111111-1111-4111-8111-111111111111'
+IDS = ['22222222-2222-4222-8222-222222222222', '33333333-3333-4333-8333-333333333333']
+SUBJECTS = [
+    {'subject': {'name': 'Alex Morgan', 'entityType': 'person', 'birthDate': '1984'}, 'package': 'sandbox@1', 'retention': 'standard', 'reference': 'example-customer-001'},
+    {'subject': {'name': 'Taylor Jordan', 'entityType': 'person', 'birthDate': '1979'}, 'package': 'sandbox@1', 'retention': 'standard', 'reference': 'example-customer-002'},
+]
+TEMPLATE = json.loads((ROOT / 'tests/fixtures/synthetic-evidence.json').read_text())
+
+class State:
+    mode = 'success'
+    calls = []
+    polls = 0
+
+class Mock(BaseHTTPRequestHandler):
+    def log_message(self, *_):
+        pass
+
+    def respond(self, code, body):
+        self.send_response(code)
+        self.send_header('Content-Type', 'application/json')
+        self.end_headers()
+        self.wfile.write(json.dumps(body).encode())
+
+    def request_record(self):
+        assert self.headers.get('Authorization') == 'Bearer ' + MOCK_KEY
+        State.calls.append({'method': self.command, 'path': self.path})
+
+    def do_POST(self):
+        self.request_record()
+        assert self.path == '/api/v1/batches'
+        body = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
+        assert body == {'name': 'Kestra fictional onboarding example', 'subjects': SUBJECTS}
+        assert self.headers['Idempotency-Key'] == 'native-test-' + State.mode
+        State.calls[-1]['idempotency_key'] = self.headers['Idempotency-Key']
+        State.calls[-1]['body'] = body
+        if State.mode == 'rate_limited_once' and len(State.calls) == 1:
+            self.respond(429, {'error': {'code': 'synthetic_retry', 'message': 'Try once more'}})
+            return
+        if State.mode in ('unauthorized', 'rate_limited', 'server_error', 'conflict'):
+            self.respond({'unauthorized': 401, 'rate_limited': 429, 'server_error': 503, 'conflict': 409}[State.mode], {'error': {'code': 'synthetic_test_error', 'message': 'Fictional test response'}})
+            return
+        if State.mode == 'redirect':
+            self.send_response(302)
+            self.send_header('Location', 'http://127.0.0.1:28102/do-not-follow')
+            self.end_headers()
+            return
+        self.respond(202, {'data': {'id': '../not-an-id' if State.mode == 'bad_batch_id' else BATCH, 'status': 'pending', 'total': 2}})
+
+    def do_GET(self):
+        self.request_record()
+        if self.path.startswith('/api/v1/batches/'):
+            assert self.path == f'/api/v1/batches/{BATCH}?limit=100&offset=0'
+            State.polls += 1
+            status = 'processing' if State.polls == 1 or State.mode == 'timeout' else 'completed'
+            rows = [{'row_number': i + 1, 'status': 'pending' if status == 'processing' else 'completed', 'screening_id': None if status == 'processing' else IDS[i], 'error': None} for i in range(2)]
+            if status == 'completed' and State.mode in ('mixed', 'all_failed', 'cancelled'):
+                indexes = [1] if State.mode == 'mixed' else [0, 1]
+                for i in indexes:
+                    rows[i].update(status='cancelled' if State.mode == 'cancelled' else 'failed', screening_id=None, error={'code': 'synthetic_row_error', 'message': 'Fictional row outcome'})
+                if State.mode == 'cancelled': status = 'cancelled'
+                if State.mode == 'all_failed': status = 'failed'
+            if status == 'completed' and State.mode == 'unfinished':
+                rows[1].update(status='pending', screening_id=None)
+            if status == 'completed' and State.mode == 'bad_uuid': rows[0]['screening_id'] = '../not-an-id'
+            if status == 'completed' and State.mode == 'row_order': rows.reverse()
+            self.respond(200, {'data': {'id': IDS[0] if State.mode == 'wrong_batch' else BATCH, 'name': 'Kestra fictional onboarding example', 'environment': 'production' if State.mode == 'wrong_environment' else 'sandbox', 'status': status, 'total': 2, 'completed': sum(r['status'] == 'completed' for r in rows) + (1 if State.mode == 'counter_mismatch' else 0), 'failed': sum(r['status'] == 'failed' for r in rows), 'created_at': '2026-01-01T00:00:00Z', 'cancelled_at': '2026-01-01T00:01:00Z' if State.mode == 'cancelled' else None, 'rows': rows, 'nextOffset': 2 if State.mode == 'extra_page' else None}})
+            return
+        m = re.fullmatch('/api/v1/results/([a-f0-9-]+)/evidence', self.path)
+        assert m and m[1] in IDS
+        i = IDS.index(m[1])
+        evidence = copy.deepcopy(TEMPLATE)
+        evidence['result']['id'] = IDS[i]
+        evidence['subject'] = SUBJECTS[i]['subject']
+        evidence['reference'] = SUBJECTS[i]['reference']
+        evidence['request'] = SUBJECTS[i]
+        if i == 1:
+            evidence['result'].update(status='no_match', matches=[])
+        if State.mode == 'wrong_evidence': evidence['result']['id'] = BATCH
+        if State.mode == 'wrong_package': evidence['result']['versions']['package'] = 'wrong@1'
+        if State.mode == 'wrong_match_count': evidence['result']['matches'] = []
+        if State.mode == 'wrong_contact': evidence['subject'] = {**evidence['subject'], 'birthDate': '1984-01-01'}
+        self.respond(200, evidence)
+
+
+def api(path, data=None, content_type=None, method=None, timeout=15):
+    headers = {'Authorization': AUTH}
+    if content_type: headers['Content-Type'] = content_type
+    req = urllib.request.Request(KESTRA + '/api/v1/main' + path, data=data, headers=headers, method=method)
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as r:
+            payload = r.read()
+            return json.loads(payload) if 'json' in r.headers.get('Content-Type', '') else payload
+    except urllib.error.HTTPError as e:
+        raise RuntimeError(f'Local Kestra HTTP {e.code}: {e.read().decode()[:2000]}') from None
+
+
+def capture_failure(execution):
+    """Save bounded local task diagnostics without replacing the test failure."""
+    def fetch(path):
+        try:
+            return api(path, timeout=3)
+        except Exception as error:
+            return {'diagnostic_error_type': type(error).__name__}
+
+    family = [execution]
+    query = urllib.parse.urlencode({'filters[parentId][EQUALS]': execution['id'], 'filters[kind][EQUALS]': 'LOOP', 'size': 3})
+    children = fetch('/executions/search?' + query)
+    # The fixed fixture has two rows. Bound diagnostics even if an engine is faulty.
+    for child in children.get('results', [])[:2]:
+        member = fetch('/executions/' + child['id'])
+        if 'id' in member:
+            family.append(member)
+    snapshots = []
+    for member in family:
+        task_outputs = {}
+        tasks = [t for t in member.get('taskRunList', []) if t['taskId'] == 'collect_evidence' or t['state']['current'] == 'FAILED']
+        for task in tasks[:3]:
+            task_outputs[task['id']] = fetch('/outputs/tasks/' + member['id'] + '/' + task['id'])
+        snapshots.append({'execution': member, 'logs': fetch('/logs/' + member['id']), 'task_outputs': task_outputs})
+    serialized = json.dumps({'case': State.mode, 'loop_search': children, 'snapshots': snapshots}, indent=2)
+    # These are invented fixture credentials, but diagnostics should not expose them.
+    fixture_credentials = [MOCK_KEY, AUTH, AUTH.removeprefix('Basic '), 'LocalMockOnly123!']
+    credentials_seen = any(value in serialized for value in fixture_credentials)
+    for value in fixture_credentials:
+        serialized = serialized.replace(value, '[fixture credential redacted]')
+    diagnostics = json.loads(serialized)
+    diagnostics['fixture_credentials_redacted'] = credentials_seen
+    (ROOT / 'tests/last-failure-diagnostics.json').write_text(json.dumps(diagnostics, indent=2) + '\n')
+
+
+def execute(mode):
+    State.mode, State.calls, State.polls = mode, [], 0
+    boundary = 'kestra-local-synthetic-test'
+    fields = {'request_key': 'native-test-' + mode, 'synthetic': 'false' if mode == 'no_confirmation' else 'true'}
+    body = ''.join(f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n{value}\r\n' for key, value in fields.items()) + f'--{boundary}--\r\n'
+    execution = api('/executions/company.sanctionskit/sanctionskit-batch-evidence', body.encode(), 'multipart/form-data; boundary=' + boundary, 'POST')
+    execution_id = execution['id']
+    deadline = time.monotonic() + 60
+    while execution['state']['current'] not in ('SUCCESS', 'FAILED', 'KILLED', 'WARNING', 'CANCELLED'):
+        assert time.monotonic() < deadline, 'Native execution exceeded test deadline'
+        time.sleep(.25)
+        execution = api('/executions/' + execution_id)
+    return execution
+
+
+def main():
+    modes = os.getenv('KESTRA_TEST_CASES', 'success,rate_limited_once,mixed,all_failed,cancelled,no_confirmation,timeout,unauthorized,rate_limited,server_error,conflict,wrong_batch,wrong_environment,extra_page,row_order,unfinished,bad_uuid,wrong_evidence,wrong_contact,wrong_package,wrong_match_count,redirect,bad_batch_id,counter_mismatch').split(',')
+    flow = (ROOT / 'sanctionskit-batch-evidence.yaml').read_text()
+    # Only endpoint, polling interval and polling budget change for local tests.
+    # Request bodies, evidence checks, HTTP limits and exports are unchanged.
+    assert flow.count('https://www.sanctionskit.com/api/v1') == 1
+    flow = flow.replace('https://www.sanctionskit.com/api/v1', 'http://127.0.0.1:28102/api/v1').replace('interval: PT2S', 'interval: PT0.1S').replace('maxIterations: 30', 'maxIterations: 3').replace('maxDuration: PT2M', 'maxDuration: PT10S')
+    exists = any(f['id'] == 'sanctionskit-batch-evidence' and f['namespace'] == 'company.sanctionskit' for f in api('/flows/search?' + urllib.parse.urlencode({'filters[namespace][EQUALS]': 'company.sanctionskit', 'size': 100}))['results'])
+    api('/flows/company.sanctionskit/sanctionskit-batch-evidence' if exists else '/flows', flow.encode(), 'application/x-yaml', 'PUT' if exists else 'POST')
+    server = ThreadingHTTPServer(('127.0.0.1', 28102), Mock)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    records = []
+    execution = None
+    try:
+        for mode in modes:
+            execution = None
+            execution = execute(mode)
+            state = execution['state']['current']
+            expected = 'SUCCESS' if mode in ('success', 'rate_limited_once', 'mixed', 'all_failed', 'cancelled') else 'FAILED'
+            record = {'case': mode, 'execution_id': execution['id'], 'state': state, 'expected': expected, 'calls': State.calls, 'polls': State.polls}
+            records.append(record)
+            if state != expected:
+                (ROOT / 'tests/last-failure.json').write_text(json.dumps(execution, indent=2))
+                print(json.dumps(record), flush=True)
+                raise AssertionError(f'{mode}: expected {expected}, got {state}; see last-failure.json')
+            family = [execution]
+            if any(t['taskId'] == 'collect_evidence' for t in execution.get('taskRunList', [])):
+                query = urllib.parse.urlencode({'filters[parentId][EQUALS]': execution['id'], 'filters[kind][EQUALS]': 'LOOP', 'size': 100})
+                children = api('/executions/search?' + query)
+                index_deadline = time.monotonic() + 5
+                while not children['results'] and time.monotonic() < index_deadline:
+                    time.sleep(.25)
+                    children = api('/executions/search?' + query)
+                assert children['total'] <= 100, 'Unexpected iteration pagination'
+                for child in children['results']:
+                    family.append(api('/executions/' + child['id']))
+                assert len(family) > 1, 'Expected native Loop iteration executions'
+            record['loop_iterations_observed'] = len(family) - 1
+            for member in family:
+                logs = api('/logs/' + member['id'])
+                assert MOCK_KEY not in json.dumps(logs), 'Mock key leaked into task logs'
+                assert MOCK_KEY not in json.dumps(member), 'Mock key leaked into execution metadata'
+                for task in member.get('taskRunList', []):
+                    if task['state']['current'] == 'SUCCESS' and task['taskId'] in ('submit_batch', 'poll_batch', 'fetch_evidence'):
+                        task_outputs = api('/outputs/tasks/' + member['id'] + '/' + task['id'])
+                        assert MOCK_KEY not in json.dumps(task_outputs), 'Mock key leaked into HTTP outputs'
+
+            if state == 'FAILED':
+                expected_task = {
+                    'no_confirmation': 'require_synthetic', 'timeout': 'wait_for_batch',
+                    'unauthorized': 'submit_batch', 'rate_limited': 'submit_batch',
+                    'server_error': 'submit_batch', 'conflict': 'submit_batch',
+                    'wrong_batch': 'check_poll_identity', 'wrong_environment': 'check_poll_identity',
+                    'extra_page': 'check_complete_page', 'row_order': 'check_complete_page',
+                    'unfinished': 'check_row', 'bad_uuid': 'check_screening_id',
+                    'wrong_evidence': 'check_evidence', 'wrong_contact': 'check_evidence',
+                    'wrong_package': 'check_evidence', 'wrong_match_count': 'check_evidence',
+                    'redirect': 'check_acceptance', 'bad_batch_id': 'check_acceptance',
+                    'counter_mismatch': 'check_complete_page',
+                }[mode]
+                failed_tasks = [t['taskId'] for member in family for t in member.get('taskRunList', []) if t['state']['current'] == 'FAILED']
+                assert expected_task in failed_tasks, (mode, expected_task, failed_tasks)
+                record['intended_failure_task'] = expected_task
+            record['mock_key_absent_from_logs_and_outputs'] = True
+            if state == 'SUCCESS':
+                outputs = api('/outputs/executions/' + execution['id'])
+                assert MOCK_KEY not in json.dumps(outputs)
+                for key in ['review_bundle', 'row_summary']:
+                    uri = outputs[key]
+                    payload = api('/executions/' + execution['id'] + '/file?path=' + urllib.parse.quote(uri, safe=''))
+                    assert MOCK_KEY not in payload.decode()
+                    if key == 'review_bundle':
+                        bundle = json.loads(payload)
+                        assert bundle['synthetic'] is True and bundle['onboardingApproved'] is False
+                        count = {'success': 2, 'rate_limited_once': 2, 'mixed': 1, 'all_failed': 0, 'cancelled': 0}[mode]
+                        assert len(bundle['evidenceByRow']) == count
+                        assert len(bundle['batch']['rows']) == 2
+                        for row, e in bundle['evidenceByRow'].items():
+                            index = int(row) - 1
+                            assert e['result']['id'] == IDS[index]
+                            assert e['request'] == SUBJECTS[index]
+                        if mode == 'success': (ROOT / 'synthetic-review-bundle.json').write_bytes(payload)
+                    else:
+                        rows = list(csv.DictReader(io.StringIO(payload.decode())))
+                        assert len(rows) == 2 and [r['reference'] for r in rows] == [s['reference'] for s in SUBJECTS]
+                        if mode == 'success': (ROOT / 'synthetic-row-summary.csv').write_bytes(payload)
+            if mode == 'no_confirmation': assert not State.calls
+            if mode in ('unauthorized', 'conflict'): assert len(State.calls) == 1
+            if mode in ('rate_limited', 'server_error'): assert len(State.calls) == 2
+            if mode == 'rate_limited_once':
+                posts = [c for c in State.calls if c['method'] == 'POST']
+                assert len(posts) == 2 and posts[0] == posts[1]
+            if mode in ('bad_uuid', 'extra_page', 'row_order', 'wrong_batch', 'wrong_environment', 'timeout'): assert not any('/results/' in c['path'] for c in State.calls)
+            if mode == 'timeout': assert State.polls == 3
+            if mode in ('redirect', 'bad_batch_id'): assert len(State.calls) == 1
+            if mode == 'counter_mismatch': assert not any('/results/' in c['path'] for c in State.calls)
+            print(f'{mode}: {state}, {len(State.calls)} mock requests', flush=True)
+    except Exception:
+        if execution is not None:
+            try:
+                capture_failure(execution)
+            except Exception as error:
+                print(f'Failure diagnostics unavailable: {type(error).__name__}', flush=True)
+        raise
+    finally:
+        server.shutdown()
+        (ROOT / 'tests/native-results.json').write_text(json.dumps({'runtime': 'Kestra 2.0.4', 'plugin': 'io.kestra.plugin:plugin-serdes:2.0.6', 'java': 'Official Temurin 25 JRE image pinned by digest', 'boundary': 'Native engine against loopback fictional HTTP mock only', 'cases': records}, indent=2) + '\n')
+
+if __name__ == '__main__':
+    main()